Add phase-free spectral pipeline probe

docs/synthid-classifiers.md

@@ -23,6 +23,45 @@ are explicit negatives, a strict rule can abstain, and any watermark claim uses
 an independent oracle. CLIP content embeddings and the 124-d origin-locked
 residual bank are different features for different jobs.
 
+### Phase-free spectral pipeline probe, 2026-09-15
+
+Mark's public
+[`bigdatamark/synthid-research`](https://huggingface.co/datasets/bigdatamark/synthid-research)
+corpus pairs 300 prompts across Gemini 2048-pixel JPEG, OpenAI 1254-pixel PNG,
+and FLUX 1280-pixel PNG exports. It is useful for testing source-pipeline
+features, but it is not a causal SynthID corpus: renderer, dimensions, encoder,
+and watermark state all change together.
+
+The paper-inspired probe in [`scripts/spectral_pipeline_probe.py`](../scripts/spectral_pipeline_probe.py)
+discards Fourier phase, subtracts each image's radial log-magnitude envelope,
+pools luma plus two opponent-color spectra, and fits a dual ridge classifier.
+Matching prompt ids stay together in a grouped five-fold evaluation. The
+256-pixel / 32x32-grid / ridge-1000 configuration was selected after an
+exploratory comparison on these same folds, so this is a development result,
+not blind validation. Its complete-corpus confusion matrix was:
+
+| Actual pipeline | Gemini | OpenAI | FLUX |
+| --- | ---: | ---: | ---: |
+| Gemini | 220 | 25 | 55 |
+| OpenAI | 33 | 219 | 48 |
+| FLUX | 75 | 48 | 177 |
+
+Accuracy was 616/900 (68.4%), against 33.3% chance. Applying each transform
+only to held-out rows gave 67.3% after a two-pixel top-left crop, 67.8% after a
+95% resize, 67.9% after JPEG quality 75, and 68.4% after Gaussian blur radius
+0.7. This is materially more transform-stable than the origin-locked lattice,
+but it remains a source/export classifier. There are no photographs, foreign
+generators beyond FLUX, independently generated 1K Gemini files, or same-image
+SynthID on/off pairs in the corpus. No runtime detector, threshold, or product
+verdict is justified by this result.
+
+The representation follows the frequency-domain premise of Zhang et al.,
+[*Detecting and Simulating Artifacts in GAN Fake Images*](https://arxiv.org/abs/1907.06515),
+not their ResNet-34 implementation. Their own post-processing results also
+support treating augmentation and architecture shift as first-class tests,
+rather than reading an untransformed source-classification score as watermark
+evidence.
+
 Krawetz's Gemini-chat TPR critique is a verifier-quality claim, not a
 feature we can ship. [Lead Stories, 2026-07](https://leadstories.com/analysis/2026/07/google-gemini-synthid-detector-confuses-results-within-same-chat.html)
 documented Gemini repeating the first file's SynthID verdict inside a

scripts/README.md

@@ -50,6 +50,7 @@ explicitly names a tracked canonical result.
 | `visible_recall_sample.py` | Build an unbiased blinded sample for recall measurement. |
 | `visible_sheets.py` | Build blinded contact sheets for relaxation candidates. |
 | `watermark_benchmark_cohort.py` | Build the deterministic synthetic DWT-DCT and TrustMark cohort consumed by `watermark_benchmark.py`. |
+| `spectral_pipeline_probe.py` | Measure a prompt-grouped, phase-free spectral source-pipeline hypothesis without presenting it as SynthID detection. |
 | `watermark_benchmark_real_cohort.py` | Build a publication-cleared, provider-paired and content-stratified real-image cohort. |
 | `retrain_photo_classify.py` | CPU-retrain the 2026-08-31 photo heads from a sha256-keyed cache pack, no images. |
 | `publish_photo_classify_hf.py` | Upload the photo-classify card and freeze weights to Hugging Face `wiltodelta/raiw-photo-classify`. Manual; the Action `publish-photo-classify-hf.yml` is the write-token path. |

scripts/spectral_pipeline_probe.py

@@ -0,0 +1,346 @@
+"""Measure a phase-free spectral source-pipeline hypothesis.
+
+This is a research harness, not a SynthID detector. It classifies complete
+generation/export pipelines whose renderer, output geometry, encoding, and
+watermark state may all be confounded. A watermark claim requires a causal
+marked/clean contrast and an independent oracle.
+
+The input directory contains one subdirectory per class. Matching filename
+stems are treated as one prompt group and always stay in the same fold.
+
+Example:
+    uv run python scripts/spectral_pipeline_probe.py DATASET \
+        --class-dir gemini --class-dir openai --class-dir flux \
+        --attack original --attack crop2 --attack resize95 \
+        --attack jpeg75 --attack blur \
+        --report-out .local-eval/spectral-pipeline-report.json
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import cast
+
+import click
+import numpy as np
+from numpy.typing import NDArray
+from PIL import Image, ImageFilter
+
+log = logging.getLogger(__name__)
+
+IMAGE_SUFFIXES = frozenset({".jpg", ".jpeg", ".png", ".webp"})
+ATTACKS = ("original", "crop2", "resize95", "jpeg75", "blur")
+
+FloatArray = NDArray[np.float64]
+IntArray = NDArray[np.int64]
+BoolArray = NDArray[np.bool_]
+
+
+@dataclass(frozen=True)
+class SpectralGeometry:
+    """Precomputed coordinates for one phase-free spectral representation."""
+
+    size: int
+    grid: int
+    radius: NDArray[np.int32]
+    radius_counts: IntArray
+    passband: BoolArray
+
+
+@dataclass(frozen=True)
+class FoldModel:
+    """One standardized dual-ridge model and its held-out rows."""
+
+    test_rows: BoolArray
+    mean: FloatArray
+    scale: FloatArray
+    active: BoolArray
+    weights: FloatArray
+
+
+def make_geometry(size: int = 256, grid: int = 32) -> SpectralGeometry:
+    """Build the radial detrending and pooling geometry."""
+    if size < 16 or grid < 2 or size % grid:
+        raise ValueError("size must be at least 16 and divisible by grid")
+    yy, xx = np.mgrid[:size, :size]
+    radius = np.rint(np.hypot(yy - size // 2, xx - size // 2)).astype(np.int32)
+    radius_counts = np.bincount(radius.ravel())
+    passband = (radius >= 4) & (radius <= size * 0.48)
+    return SpectralGeometry(size, grid, radius, radius_counts, passband)
+
+
+def phase_free_log_spectrum(channel: FloatArray) -> FloatArray:
+    """Return centered log FFT magnitude, deliberately discarding phase."""
+    if channel.ndim != 2:
+        raise ValueError("channel must be a two-dimensional array")
+    centered = np.asarray(channel, dtype=np.float64) - float(np.mean(channel))
+    return np.log1p(np.abs(np.fft.fftshift(np.fft.fft2(centered))))
+
+
+def spectral_feature(pixels: FloatArray, geometry: SpectralGeometry) -> FloatArray:
+    """Extract pooled radial-residual spectra from luma and opponent color."""
+    if pixels.shape != (geometry.size, geometry.size, 3):
+        raise ValueError(f"pixels must have shape {(geometry.size, geometry.size, 3)}, got {pixels.shape}")
+    red, green, blue = np.moveaxis(np.asarray(pixels, dtype=np.float64), 2, 0)
+    channels = (
+        0.299 * red + 0.587 * green + 0.114 * blue,
+        red - green,
+        blue - (red + green) / 2.0,
+    )
+    block = geometry.size // geometry.grid
+    features: list[FloatArray] = []
+    for channel in channels:
+        spectrum = phase_free_log_spectrum(channel)
+        radial_sum = np.bincount(geometry.radius.ravel(), weights=spectrum.ravel())
+        radial_mean = radial_sum / geometry.radius_counts
+        residual = np.where(
+            geometry.passband,
+            spectrum - radial_mean[geometry.radius],
+            0.0,
+        )
+        pooled = residual.reshape(geometry.grid, block, geometry.grid, block).mean(axis=(1, 3))
+        features.append(pooled.ravel())
+    return np.concatenate(features)
+
+
+def _apply_attack(image: Image.Image, attack: str) -> Image.Image:
+    """Apply one declared robustness transform before canonical resizing."""
+    if attack == "original":
+        return image
+    if attack == "crop2":
+        if image.width <= 4 or image.height <= 4:
+            raise ValueError("crop2 requires both image dimensions to exceed four pixels")
+        return image.crop((2, 2, image.width, image.height))
+    if attack == "resize95":
+        return image.resize(
+            (max(1, round(image.width * 0.95)), max(1, round(image.height * 0.95))),
+            Image.Resampling.LANCZOS,
+        )
+    if attack == "jpeg75":
+        encoded = io.BytesIO()
+        image.save(encoded, format="JPEG", quality=75)
+        encoded.seek(0)
+        with Image.open(encoded) as decoded:
+            return decoded.convert("RGB")
+    if attack == "blur":
+        return image.filter(ImageFilter.GaussianBlur(0.7))
+    raise ValueError(f"unsupported attack: {attack}")
+
+
+def feature_from_path(path: Path, geometry: SpectralGeometry, attack: str) -> FloatArray:
+    """Decode one path, apply ATTACK, and return its phase-free feature."""
+    with Image.open(path) as source:
+        return _feature_from_image(source.convert("RGB"), geometry, attack)
+
+
+def _feature_from_image(image: Image.Image, geometry: SpectralGeometry, attack: str) -> FloatArray:
+    """Extract one attacked feature from an already decoded RGB image."""
+    attacked = _apply_attack(image, attack)
+    canonical = attacked.resize((geometry.size, geometry.size), Image.Resampling.LANCZOS)
+    pixels = np.asarray(canonical, dtype=np.float64) / 255.0
+    return spectral_feature(pixels, geometry)
+
+
+def discover_grouped_paths(
+    root: Path,
+    class_dirs: tuple[str, ...],
+) -> tuple[list[Path], IntArray, IntArray, list[str]]:
+    """Return class-major paths over filename stems shared by every class."""
+    if len(class_dirs) < 2 or len(set(class_dirs)) != len(class_dirs):
+        raise ValueError("at least two unique class directories are required")
+    per_class: list[dict[str, Path]] = []
+    for class_dir in class_dirs:
+        directory = root / class_dir
+        if not directory.is_dir():
+            raise ValueError(f"class directory does not exist: {directory}")
+        mapping: dict[str, Path] = {}
+        for path in sorted(directory.iterdir()):
+            if path.is_file() and path.suffix.lower() in IMAGE_SUFFIXES:
+                if path.stem in mapping:
+                    raise ValueError(f"duplicate filename stem in {directory}: {path.stem}")
+                mapping[path.stem] = path
+        per_class.append(mapping)
+    common_names: set[str] = set(per_class[0])
+    for mapping in per_class[1:]:
+        common_names.intersection_update(mapping)
+    group_names = sorted(common_names)
+    if not group_names:
+        raise ValueError("class directories have no shared filename stems")
+    paths: list[Path] = []
+    labels: list[int] = []
+    groups: list[int] = []
+    for class_index, mapping in enumerate(per_class):
+        for group_index, group_name in enumerate(group_names):
+            paths.append(mapping[group_name])
+            labels.append(class_index)
+            groups.append(group_index)
+    return paths, np.asarray(labels), np.asarray(groups), group_names
+
+
+def fit_grouped_models(
+    features: FloatArray,
+    labels: IntArray,
+    groups: IntArray,
+    *,
+    class_count: int,
+    folds: int = 5,
+    ridge: float = 1000.0,
+) -> list[FoldModel]:
+    """Fit prompt-grouped dual ridge models without test-fold leakage."""
+    if features.ndim != 2 or len(features) != len(labels) or len(labels) != len(groups):
+        raise ValueError("features, labels, and groups must describe the same rows")
+    if folds < 2 or len(np.unique(groups)) < folds:
+        raise ValueError("fold count must be between two and the number of groups")
+    if class_count < 2 or ridge <= 0:
+        raise ValueError("class_count must be at least two and ridge must be positive")
+    targets = np.eye(class_count, dtype=np.float64)[labels]
+    models: list[FoldModel] = []
+    for fold in range(folds):
+        test_rows = groups % folds == fold
+        train_rows = ~test_rows
+        train = features[train_rows]
+        mean = np.mean(train, axis=0)
+        scale = np.std(train, axis=0)
+        active = scale > 1e-8
+        train_features = (train[:, active] - mean[active]) / scale[active]
+        kernel = train_features @ train_features.T
+        dual_weights = cast(
+            "FloatArray",
+            np.linalg.solve(
+                kernel + ridge * np.eye(len(kernel)),
+                targets[train_rows],
+            ),
+        )
+        weights = train_features.T @ dual_weights
+        models.append(FoldModel(test_rows, mean, scale, active, weights))
+    return models
+
+
+def predict_grouped(models: list[FoldModel], features: FloatArray) -> IntArray:
+    """Predict every row only with the model that held out its group."""
+    predictions = np.full(len(features), -1, dtype=np.int64)
+    for model in models:
+        test_features = (features[model.test_rows][:, model.active] - model.mean[model.active]) / model.scale[
+            model.active
+        ]
+        scores = test_features @ model.weights
+        predictions[model.test_rows] = np.argmax(scores, axis=1)
+    if np.any(predictions < 0):
+        raise ValueError("fold models did not cover every row")
+    return predictions
+
+
+def confusion_matrix(labels: IntArray, predictions: IntArray, class_count: int) -> IntArray:
+    """Count actual rows by predicted class."""
+    matrix = np.zeros((class_count, class_count), dtype=np.int64)
+    np.add.at(matrix, (labels, predictions), 1)
+    return matrix
+
+
+def _extract_all_attacks(
+    paths: list[Path],
+    geometry: SpectralGeometry,
+    attacks: tuple[str, ...],
+) -> dict[str, FloatArray]:
+    """Decode each path once and extract every requested attack feature."""
+    log.info("Extracting %s attack views from %s images", len(attacks), len(paths))
+    rows: dict[str, list[FloatArray]] = {attack: [] for attack in attacks}
+    for path in paths:
+        with Image.open(path) as source:
+            image = source.convert("RGB")
+        for attack in attacks:
+            rows[attack].append(_feature_from_image(image, geometry, attack))
+    return {attack: np.asarray(features) for attack, features in rows.items()}
+
+
+@click.command()
+@click.argument("dataset", type=click.Path(exists=True, file_okay=False, path_type=Path))
+@click.option(
+    "class_dirs",
+    "--class-dir",
+    multiple=True,
+    required=True,
+    help="Class subdirectory; repeat in label order.",
+)
+@click.option(
+    "attacks",
+    "--attack",
+    multiple=True,
+    type=click.Choice(ATTACKS),
+    default=("original",),
+    show_default=True,
+)
+@click.option("--size", type=click.IntRange(min=16), default=256, show_default=True)
+@click.option("--grid", type=click.IntRange(min=2), default=32, show_default=True)
+@click.option("--folds", type=click.IntRange(min=2), default=5, show_default=True)
+@click.option("--ridge", type=click.FloatRange(min=0.0, min_open=True), default=1000.0, show_default=True)
+@click.option("--report-out", type=click.Path(dir_okay=False, path_type=Path), required=True)
+def main(
+    dataset: Path,
+    class_dirs: tuple[str, ...],
+    attacks: tuple[str, ...],
+    size: int,
+    grid: int,
+    folds: int,
+    ridge: float,
+    report_out: Path,
+) -> None:
+    """Run grouped spectral pipeline classification over DATASET."""
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    if "original" not in attacks:
+        raise click.UsageError("--attack original is required because every model trains on original exports")
+    unique_attacks = tuple(dict.fromkeys(attacks))
+    try:
+        geometry = make_geometry(size, grid)
+        paths, labels, groups, group_names = discover_grouped_paths(dataset, class_dirs)
+        features_by_attack = _extract_all_attacks(paths, geometry, unique_attacks)
+        original = features_by_attack["original"]
+        models = fit_grouped_models(
+            original,
+            labels,
+            groups,
+            class_count=len(class_dirs),
+            folds=folds,
+            ridge=ridge,
+        )
+    except ValueError as error:
+        raise click.ClickException(str(error)) from error
+
+    results: dict[str, object] = {}
+    for attack in unique_attacks:
+        features = features_by_attack[attack]
+        predictions = predict_grouped(models, features)
+        matrix = confusion_matrix(labels, predictions, len(class_dirs))
+        results[attack] = {
+            "accuracy": float(np.trace(matrix) / np.sum(matrix)),
+            "confusion_matrix": matrix.tolist(),
+        }
+
+    report = {
+        "claim": "phase-free generation/export pipeline classification; not SynthID detection",
+        "causal_limit": (
+            "Renderer, output geometry, encoding, and watermark state vary together. "
+            "Only a marked/clean causal contrast plus an independent oracle can isolate the watermark."
+        ),
+        "evaluation_limit": (
+            "Grouped cross-validation estimates this corpus only. Hyperparameters selected on these folds "
+            "need an independent locked corpus before they support a generalization claim."
+        ),
+        "dataset": str(dataset),
+        "classes": list(class_dirs),
+        "group_count": len(group_names),
+        "image_count": len(paths),
+        "configuration": {"size": size, "grid": grid, "folds": folds, "ridge": ridge},
+        "results": results,
+    }
+    report_out.parent.mkdir(parents=True, exist_ok=True)
+    report_out.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    log.info("Wrote spectral pipeline report: %s", report_out)
+
+
+if __name__ == "__main__":
+    main()

tests/test_spectral_pipeline_probe.py

@@ -0,0 +1,81 @@
+"""Tests for the phase-free spectral pipeline research harness."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import numpy as np
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+
+import spectral_pipeline_probe as probe
+
+
+def test_log_spectrum_discards_translation_phase() -> None:
+    rng = np.random.default_rng(17)
+    channel = rng.normal(size=(32, 32))
+
+    original = probe.phase_free_log_spectrum(channel)
+    translated = probe.phase_free_log_spectrum(np.roll(channel, shift=(5, -7), axis=(0, 1)))
+
+    assert np.allclose(original, translated, rtol=1e-12, atol=1e-12)
+
+
+def test_spectral_feature_is_finite_and_has_declared_shape() -> None:
+    geometry = probe.make_geometry(size=32, grid=4)
+    yy, xx = np.mgrid[:32, :32]
+    carrier = np.cos(2 * np.pi * (5 * xx + 3 * yy) / 32)
+    pixels = np.stack((carrier, np.roll(carrier, 1, axis=0), np.roll(carrier, 2, axis=1)), axis=2)
+
+    feature = probe.spectral_feature(pixels, geometry)
+
+    assert feature.shape == (3 * 4 * 4,)
+    assert np.all(np.isfinite(feature))
+    assert np.linalg.norm(feature) > 0
+
+
+def test_discovery_keeps_only_stems_shared_by_every_class(tmp_path: Path) -> None:
+    for class_name in ("first", "second"):
+        directory = tmp_path / class_name
+        directory.mkdir()
+        for stem in ("001", "002"):
+            Image.new("RGB", (8, 8), "white").save(directory / f"{stem}.png")
+    Image.new("RGB", (8, 8), "black").save(tmp_path / "first" / "unpaired.png")
+
+    paths, labels, groups, names = probe.discover_grouped_paths(tmp_path, ("first", "second"))
+
+    assert names == ["001", "002"]
+    assert [path.stem for path in paths] == ["001", "002", "001", "002"]
+    assert labels.tolist() == [0, 0, 1, 1]
+    assert groups.tolist() == [0, 1, 0, 1]
+
+
+def test_grouped_ridge_never_trains_on_the_test_group() -> None:
+    groups = np.tile(np.arange(14), 3)
+    labels = np.repeat(np.arange(3), 14)
+    class_centers = np.eye(3)[labels] * 8.0
+    group_nuisance = np.column_stack((groups % 2, groups % 3, groups % 5)) * 0.01
+    features = class_centers + group_nuisance
+
+    models = probe.fit_grouped_models(features, labels, groups, class_count=3, folds=5, ridge=1.0)
+    predictions = probe.predict_grouped(models, features)
+
+    assert np.array_equal(predictions, labels)
+    fold_assignment = np.full(len(groups), -1)
+    for fold, model in enumerate(models):
+        assert not np.any(fold_assignment[model.test_rows] >= 0)
+        fold_assignment[model.test_rows] = fold
+    assert np.all(fold_assignment >= 0)
+    for group in np.unique(groups):
+        assert len(np.unique(fold_assignment[groups == group])) == 1
+
+
+def test_confusion_matrix_uses_actual_rows_and_predicted_columns() -> None:
+    labels = np.asarray([0, 0, 1, 1, 2])
+    predictions = np.asarray([0, 1, 1, 2, 0])
+
+    matrix = probe.confusion_matrix(labels, predictions, class_count=3)
+
+    assert matrix.tolist() == [[1, 1, 0], [0, 1, 1], [1, 0, 0]]