From d914e3ac088a52ba3a012a6f58c53178c5b28dfe Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Thu, 10 Sep 2026 17:20:35 -0700 Subject: [PATCH] Add comprehensive signal discovery workflow --- .claude/skills/signal-discovery/SKILL.md | 85 +++++++++ .../references/coverage-and-cadence.md | 57 ++++++ docs/verification-plan.md | 26 +++ scripts/README.md | 2 +- scripts/corpus_gap_scan.py | 175 ++++++++++++------ scripts/sidecar_regression.py | 44 ++++- tests/test_signal_discovery_scripts.py | 76 ++++++++ 7 files changed, 403 insertions(+), 62 deletions(-) create mode 100644 .claude/skills/signal-discovery/SKILL.md create mode 100644 .claude/skills/signal-discovery/references/coverage-and-cadence.md create mode 100644 tests/test_signal_discovery_scripts.py diff --git a/.claude/skills/signal-discovery/SKILL.md b/.claude/skills/signal-discovery/SKILL.md new file mode 100644 index 0000000..d20d8e5 --- /dev/null +++ b/.claude/skills/signal-discovery/SKILL.md @@ -0,0 +1,85 @@ +--- +name: signal-discovery +description: Discover missing AI provenance and watermark coverage in the remove-ai-watermarks library from fresh or historical corpora, then validate detection and removal parity. Use for corpus gap mining, periodic signal audits, new-provider discovery, or deciding whether old retained data must be replayed. Do not use for ordinary per-file identification or removal. +--- + +# Signal discovery + +Work from the `remove-ai-watermarks` repository root. This skill discovers and +closes library coverage gaps. Implementation, tests, fixtures, and +documentation changes belong in this library; do not patch a downstream +consumer to compensate for a detector or cleaner gap. + +Treat retained inputs as sensitive and read-only. Keep reports, contact sheets, +sidecars, and derived samples under the library's gitignored `.local-eval/`. +Never commit a retained input or an artifact derived from one. +Use only publication-cleared or synthetic files for tracked fixtures. + +Before acting, read: + +- `CLAUDE.md` and `.claude/rules/development.md`; +- `docs/verification-plan.md`; +- `docs/supported-signals.md` and `docs/watermarking-landscape.md`; +- [references/coverage-and-cadence.md](references/coverage-and-cadence.md). + +## Choose the run + +Do not call `corpus_gap_scan.py` a complete signal audit. Select every mode the +question requires: + +1. **Fresh discovery** finds new providers, serializations, labels, and formats + in objects added since the preceding review. +2. **Historical replay** reruns current code on the complete retained history. + Use it after detection-path or format-support changes and on the periodic + backstop defined in the reference. +3. **Removal parity** proves that every detected removable signal is cleared + without violating the signal family's fidelity invariant. +4. **Calibration or provider oracle** is required where the corpus has no + self-evident answer key. Never treat a local negative as proof that a + proprietary watermark is absent. + +Start by inventorying the current code, not a remembered list. Enumerate image +and video visible-mark registries, metadata readers and strippers, local +invisible decoders, explicit invisible vendor routes, supported containers, and +the stable signal families emitted by `identify`. Reconcile that inventory with +the coverage matrix before interpreting any green result. A hand-written marker +list or report family map that omits a current registry member is itself a +finding. + +## Run protocol + +1. Record the library commit, version, enabled extras, corpus date range, and + exact commands in an untracked run directory. Preserve the pre-run Git + status so unrelated work is not attributed to the audit. +2. Refresh the local retained corpus through its owning, external collection + process when the task calls for fresh data. Use an explicit gitignored + destination, do not delete local objects, and finish with a zero-delta pass. + Fetch paired outputs only for a removal or complaint investigation. +3. Run the applicable fresh and historical discovery modes from the reference. + Use `corpus_gap_scan.py --since YYYY-MM-DD` for the overlapping fresh window + and omit `--since` for the complete history. Report counts per input shape + and per outcome; an example does not establish coverage. +4. Triage every candidate into one of: parser/serialization gap, missing vendor + attribution, new visible mark, possible invisible signal, format/read error, + expected non-AI provenance, or false candidate. Preserve unresolved items as + unresolved. +5. Reproduce an actionable gap with a failing test before changing code. Derive + the smallest publication-safe fixture from a controlled source or construct + it synthetically; do not move the retained file into a tracked path. +6. Implement the fix in the library subsystem that owns it. Update the library + registry, parser, remover, public docs, and published agent skill when their + contracts move. Updating a downstream consumer to take a released version is + a later, separately scoped task. +7. Rerun the affected focused audit and the complete historical regression + mode. A detection change must name the files expected to move before the + after-run; classify lost signals separately from intended new detections. +8. Run removal parity for every newly detected removable signal. A quiet + post-removal detector is necessary but not always sufficient: check + pixel/stream invariants and visually inspect residual-prone visible marks. +9. Run `bash maintain.sh`. Report evidence limits, incomplete rows, crashes, + unsupported formats, and oracle-unavailable cohorts explicitly. + +External oracle uploads are separate mutations. Use the `provider-oracles` +skill only for a bounded, already-selected candidate, and obtain the required +action-time upload approval. Do not adapt candidates against an oracle, retry +automatically, or substitute one provider's verdict for another's. diff --git a/.claude/skills/signal-discovery/references/coverage-and-cadence.md b/.claude/skills/signal-discovery/references/coverage-and-cadence.md new file mode 100644 index 0000000..72d1e30 --- /dev/null +++ b/.claude/skills/signal-discovery/references/coverage-and-cadence.md @@ -0,0 +1,57 @@ +# Coverage and cadence + +Use this reference to choose the audit modes and decide whether current-only or +historical data is sufficient. + +## Coverage matrix + +| Surface | Discovery evidence | Regression check | Removal proof | +| --- | --- | --- | --- | +| C2PA, EXIF, XMP, IPTC, PNG text, TC260, soft bindings, and container metadata | `scripts/corpus_gap_scan.py` plus review of unattributed and partially attributed reports | `scripts/sidecar_regression.py`; compare record-based and file-based identification when collection changes | `scripts/metadata_removal_audit.py`; signal absent after strip and decoded pixels/streams unchanged | +| Registered visible image marks | `scripts/visible_positives.py`; metadata-derived vendor cohorts and blinded sheets for uncovered labels | `scripts/visible_eval.py`, registered examples, and full-corpus crossfire | `scripts/visible_removal_audit.py`; same-mark redetection plus visual residual review and fill-quality checks | +| Candidate China-AIGC visible marks | `scripts/vendor_cohort_harvest.py` partitions by producer identity without using pixels; inspect full-width top and bottom sheets | `scripts/vendor_mark_calibrate.py` against independently labelled positives, clean negatives, and neighbouring marks | Register only after the candidate gate survives full-corpus crossfire; then use the normal visible-removal audit | +| Registered visible video marks | Review provider-labelled or controlled complete clips, not isolated frames alone | Full-clip tests and the local real-provider audit described in `docs/verification-plan.md` | Preserve sequence, timing, duration, and audio; second-pass detection must be quiet | +| Locally decodable invisible image marks, including open DWT-DCT and TrustMark | Positive-control-gated corpus or constructed carriers; a negative on an uncontrolled carrier is inconclusive | `scripts/watermark_benchmark.py` and the relevant decoder tests with the required extras installed | Re-run the same decoder on output and retain a positive control in the same run | +| Proprietary image marks: Google SynthID, Microsoft InvisMark, Meta Content Seal | Provider provenance can select a cohort but is not a local pixel decode; use bounded provider-labelled candidates | Hash-bound provider-oracle batches when a new route, profile, or claimed signal is evaluated | Source-positive and output-negative verdicts from that same provider; record C2PA separately from the pixel watermark | +| Invisible video and audio marks | Matched provider or open-model cohorts with codec and degradation controls | Benchmark-kernel audio/video rows and temporal aggregation checks | Preserve the non-target stream; use the matching oracle and a control-positive run | +| Explicit vendor regeneration with no detectable local signal | This is a user assertion or research cohort, not a discovered signal | Verify routing and profile selection separately from detection | Quality and provider-oracle evidence may certify a cohort, but must not create a local detection claim | + +`classify` is a pixel-origin guess, not provenance, and must not be counted as a +new signal. `erase --region` is user-directed editing, not automatic watermark +detection, and is outside this audit. + +## Gaps the legacy scanner cannot settle alone + +`corpus_gap_scan.py` is a metadata serialization probe. It reports raw markers, +input shapes, identify errors, integrity clashes, unattributed results, and +blind-marker candidates, but it still cannot settle other useful classes, so +inspect them separately: + +- a known generic signal with missing or wrong vendor attribution; +- a newly coexisting marker on a file that already has another AI verdict; +- an entirely novel marker absent from the scanner's hand-written list; +- visible image or video labels; +- proprietary invisible marks without a local decoder; +- detection/removal parity failures; +- decoder crashes, unsupported extensions, and incomplete reads; +- drift between stored identify sidecars and the current library; +- crossfire where a new detector fires on an older neighbouring cohort. + +Marker matching must remain metadata-region bounded to avoid random hits in +compressed pixels, but its inventory should be reconciled with the current +metadata constants and registry on every audit. Do not infer completeness from +`No gap candidates` until the other rows of the matrix also ran. + +## When to replay old data + +Yes, old data must be rerun. Current code can learn to parse an old +serialization, and a new detector can crossfire on an old provider cohort even +when no new object demonstrates either problem. The canonical intervals, +change triggers, and sampling rules live in +[`docs/verification-plan.md#signal-discovery-cadence`](../../../../docs/verification-plan.md#signal-discovery-cadence). + +Keep dated reports immutable enough to compare runs. Record the library SHA and +version, corpus minimum and maximum dates, input counts per media/format shape, +enabled extras, command line, and output paths. Never commit operational corpus +counts or retained identifiers; summarize only durable conclusions in the +canonical research document. diff --git a/docs/verification-plan.md b/docs/verification-plan.md index 9054366..640bcc2 100644 --- a/docs/verification-plan.md +++ b/docs/verification-plan.md @@ -490,6 +490,32 @@ bridge or reset its start PTS must fail this gate. A separate constant-rate clip with the same non-zero start guards the start-offset routing without relying on the VFR branch. +## Signal discovery cadence + +The project skill [signal-discovery](../.claude/skills/signal-discovery/SKILL.md) +orchestrates the corpus audits in this plan. `scripts/corpus_gap_scan.py` is only +the metadata-serialization probe inside that workflow. It now records raw +markers, input shapes, identify errors, integrity clashes, unattributed results, +and blind-marker candidates, but it does not cover visible marks, proprietary +invisible marks, detection/removal parity, or historical crossfire. + +Run a fresh-delta discovery review weekly after the retained corpus is +refreshed, using `corpus_gap_scan.py --since` with an overlap from the preceding +run. Replay the complete historical corpus before releases that change a +detection, metadata, container, registry, invisible-routing, or removal seam, +after every detector registration or recalibration, and monthly as a backstop. +The historical pass is required because newer code can expose an old +serialization and a new detector can crossfire on an old neighbouring cohort. +A stratified sidecar sample is a useful preflight but does not replace a +triggered full replay. + +Removal-parity sweeps are change-triggered rather than weekly: run the matching +metadata, visible, video, or invisible audit whenever either its detector or +cleaner moves. External provider oracles are claim-triggered, never +calendar-triggered. Keep private inputs and reports untracked, and implement an +actionable detector or cleaner gap in this library rather than in a downstream +consumer. + ## Standing gap None of this is in `maintain.sh`, and it should not all be -- the sweeps take hours. But diff --git a/scripts/README.md b/scripts/README.md index 6dad77d..2357bcc 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -9,7 +9,7 @@ explicitly names a tracked canonical result. | Script | Purpose | | --- | --- | -| `corpus_gap_scan.py` | Compare a local image corpus with the library's `identify` results. | +| `corpus_gap_scan.py` | Inventory `identify`, metadata markers, unattributed outcomes, errors, and gap candidates over a full or date-bounded corpus. | | `detection_timing.py` | Record per-method metadata and verdict timings. | | `detection_timing_report.py` | Aggregate timing records by method and segment. | | `fidelity_metrics.py` | Compute objective image-fidelity metrics for paired outputs. | diff --git a/scripts/corpus_gap_scan.py b/scripts/corpus_gap_scan.py index f15d3df..219d6d2 100644 --- a/scripts/corpus_gap_scan.py +++ b/scripts/corpus_gap_scan.py @@ -3,7 +3,8 @@ Two jobs in one pass: 1. **Report** -- run ``identify`` over every image and write one CSV row per file - (verdict, platform, confidence, watermarks, signals, integrity clashes). + (verdict, platform, confidence, watermarks, signals, raw metadata markers, + candidate classes, integrity clashes, and errors). 2. **Gap audit** -- for every ``unknown``-verdict file, scan only its *metadata region* (PNG text/eXIf chunks, JPEG APPn segments before SOS, or the file head for other containers) for known provenance markers. A marker found there @@ -20,6 +21,8 @@ Usage: uv run python scripts/corpus_gap_scan.py --corpus .local-eval/originals uv run python scripts/corpus_gap_scan.py --corpus .local-eval/originals \\ --report .local-eval/detector-report.csv + uv run python scripts/corpus_gap_scan.py --corpus .local-eval/originals \\ + --since 2026-09-01 --report .local-eval/detector-report-weekly.csv """ from __future__ import annotations @@ -27,13 +30,17 @@ from __future__ import annotations import csv import logging from collections import Counter +from datetime import date +from importlib.metadata import version from pathlib import Path +from typing import Any import click from _plain_console import Console, Table +from remove_ai_watermarks.identify import _metadata_region as identify_metadata_region from remove_ai_watermarks.identify import identify -from remove_ai_watermarks.metadata import _png_late_metadata +from remove_ai_watermarks.metadata import scan_head log = logging.getLogger(__name__) console = Console() @@ -80,35 +87,31 @@ MARKERS: tuple[bytes, ...] = ( b"Samsung Galaxy", ) - -def _metadata_region(path: Path) -> bytes: - """Return only the bytes where provenance metadata can live, never the - compressed pixel stream (which produces random short-token collisions).""" - try: - head = path.read_bytes() - except OSError: - return b"" - if head[:8] == b"\x89PNG\r\n\x1a\n": - # All ancillary metadata chunks (window=0), via the library's own walker. - return _png_late_metadata(path, 0) - if head[:2] == b"\xff\xd8": # JPEG: APPn segments up to Start-Of-Scan - out = bytearray() - p = 2 - n = len(head) - while p + 4 <= n and head[p] == 0xFF: - marker = head[p + 1] - if marker == 0xDA: # SOS -> compressed scan data follows - break - seg_len = (head[p + 2] << 8) | head[p + 3] - out += head[p + 4 : p + 2 + seg_len] - p += 2 + seg_len - return bytes(out) - return head[:65536] # webp/avif/heif/jxl: metadata sits near the head +REPORT_FIELDS: tuple[str, ...] = ( + "path", + "suffix", + "lib_version", + "is_ai", + "platform", + "confidence", + "watermarks", + "signals", + "integrity_clashes", + "markers", + "candidate_classes", + "error", +) -def _row(rep) -> dict[str, str]: # noqa: ANN001 (ProvenanceReport) +def _base_row(path: str, suffix: str, lib_version: str) -> dict[str, str]: + row = dict.fromkeys(REPORT_FIELDS, "") + row.update(path=path, suffix=suffix, lib_version=lib_version) + return row + + +def _row(rep, *, path: str, suffix: str, lib_version: str) -> dict[str, str]: # noqa: ANN001 return { - "path": "", # filled by caller (relative) + **_base_row(path, suffix, lib_version), "is_ai": str(rep.is_ai_generated), "platform": rep.platform or "", "confidence": rep.confidence, @@ -118,6 +121,55 @@ def _row(rep) -> dict[str, str]: # noqa: ANN001 (ProvenanceReport) } +def _marker_hits(region: bytes) -> list[str]: + """Return marker labels found case-insensitively in one metadata region.""" + folded = region.lower() + return sorted({marker.decode("latin-1", "replace") for marker in MARKERS if marker.lower() in folded}) + + +def _marker_hits_for_path(path: Path) -> list[str]: + """Return bounded metadata hits without turning an unreadable row into a fatal scan.""" + try: + return _marker_hits(identify_metadata_region(scan_head(path))) + except OSError: + return [] + + +def _candidate_classes(rep: Any, hits: list[str]) -> list[str]: + """Classify review-worthy outcomes without treating them as proven bugs.""" + classes: list[str] = [] + if hits and not rep.is_ai_generated and not rep.signals: + classes.append("blind_marker") + if rep.is_ai_generated and not rep.platform: + classes.append("unattributed_ai") + elif rep.signals and not rep.platform: + classes.append("unattributed_signal") + if rep.integrity_clashes: + classes.append("integrity_clash") + return classes + + +def _day_of(path: Path, corpus: Path) -> date | None: + """Read the leading YYYY-MM-DD corpus segment, if this layout has one.""" + try: + segment = path.relative_to(corpus).parts[0] + return date.fromisoformat(segment) + except (ValueError, IndexError): + return None + + +def _files(corpus: Path, since: date | None) -> list[Path]: + """Return the deterministic corpus walk, optionally bounded by its date segment.""" + if since is None: + return sorted(path for path in corpus.rglob("*") if path.is_file()) + roots = [ + path + for path in corpus.iterdir() + if path.is_dir() and (day := _day_of(path, corpus)) is not None and day >= since + ] + return sorted(path for root in roots for path in root.rglob("*") if path.is_file()) + + @click.command() @click.option( "--corpus", @@ -132,82 +184,93 @@ def _row(rep) -> dict[str, str]: # noqa: ANN001 (ProvenanceReport) default=None, help="Write the per-file CSV here (default: /../detector_report.csv).", ) +@click.option( + "--since", + type=click.DateTime(formats=["%Y-%m-%d"]), + default=None, + help="Only scan files under YYYY-MM-DD corpus directories on or after this date.", +) @click.option("--limit", type=int, default=0, help="Scan at most N files (0 = all).") -def main(corpus: Path, report: Path | None, limit: int) -> None: +def main(corpus: Path, report: Path | None, since, limit: int) -> None: # noqa: ANN001 logging.basicConfig(level=logging.WARNING, format="%(message)s") report = report or corpus.parent / "detector_report.csv" - files = sorted(p for p in corpus.rglob("*") if p.is_file()) + since_date = since.date() if since is not None else None + files = _files(corpus, since_date) if limit: files = files[:limit] console.print(f"Scanning [bold]{len(files)}[/bold] files under {corpus} ...") verdicts: Counter[str] = Counter() platforms: Counter[str] = Counter() - gap_tokens: Counter[str] = Counter() - gaps: list[tuple[str, list[str]]] = [] + shapes: Counter[str] = Counter() rows: list[dict[str, str]] = [] errors = 0 + lib_version = version("remove-ai-watermarks") with click.progressbar(files, label="identify") as bar: for p in bar: rel = str(p.relative_to(corpus)) + shapes[p.suffix.lower() or "(none)"] += 1 try: rep = identify(p) except Exception as exc: log.warning("identify failed on %s: %s", rel, exc) errors += 1 + row = _base_row(rel, p.suffix.lower(), lib_version) + row["candidate_classes"] = "identify_error" + row["error"] = f"{type(exc).__name__}: {exc}"[:300] + hits = _marker_hits_for_path(p) + row["markers"] = "|".join(hits) + rows.append(row) continue - row = _row(rep) - row["path"] = rel + row = _row(rep, path=rel, suffix=p.suffix.lower(), lib_version=lib_version) + # identify() has already populated scan_head's bounded cache, so this + # reuses the exact metadata region rather than reading the file again. + hits = _marker_hits_for_path(p) + classes = _candidate_classes(rep, hits) + row["markers"] = "|".join(hits) + row["candidate_classes"] = "|".join(classes) rows.append(row) if rep.is_ai_generated: verdicts["ai"] += 1 platforms[rep.platform or "?"] += 1 continue verdicts["unknown"] += 1 - # A gap candidate is a file identify is *blind* to (no signal at all) - # yet whose metadata carries a known marker. A file that produced a - # signal but no AI verdict (e.g. an ASUS Gallery C2PA signer, which we - # attribute but do not call AI) is handled correctly -- not a gap. - if rep.signals: - continue - region = _metadata_region(p) - hits = sorted({m.decode("latin-1", "replace") for m in MARKERS if m in region}) - if hits: - gaps.append((rel, hits)) - gap_tokens.update(hits) + report.parent.mkdir(parents=True, exist_ok=True) with report.open("w", newline="") as f: - writer = csv.DictWriter( - f, - fieldnames=["path", "is_ai", "platform", "confidence", "watermarks", "signals", "integrity_clashes"], - ) + writer = csv.DictWriter(f, fieldnames=REPORT_FIELDS) writer.writeheader() writer.writerows(rows) console.print(f"\nWrote [bold]{len(rows)}[/bold] rows -> {report}") console.print(f"\n[bold]Verdicts:[/bold] AI {verdicts['ai']} | unknown {verdicts['unknown']} | errors {errors}") + console.print("[bold]Input shapes:[/bold] " + " | ".join(f"{name} {n}" for name, n in shapes.most_common())) plat = Table(title="AI platforms", show_header=False) for name, n in platforms.most_common(): plat.add_row(str(n), name) console.print(plat) - if gaps: + candidates = [row for row in rows if row["candidate_classes"]] + if candidates: + candidate_counts = Counter(candidate for row in candidates for candidate in row["candidate_classes"].split("|")) + gap_tokens = Counter(marker for row in rows for marker in row["markers"].split("|") if marker) console.print( - f"\n[bold red]Gap candidates[/bold red]: {len(gaps)} unknown files carry a known " - f"marker in their metadata region (potential undetected serialization/generator):" + f"\n[bold red]Review candidates[/bold red]: {len(candidates)} file(s) " + f"({', '.join(f'{name}={n}' for name, n in candidate_counts.most_common())})" ) - tok = Table(title="markers seen in unknown files") + tok = Table(title="metadata markers seen across the run") tok.add_column("count", justify="right") tok.add_column("marker") for name, n in gap_tokens.most_common(): tok.add_row(str(n), name) console.print(tok) - for rel, hits in gaps: - console.print(f" {rel} -> {', '.join(hits)}") + for row in candidates: + detail = f"; markers={row['markers'].replace('|', ', ')}" if row["markers"] else "" + console.print(f" {row['path']} -> {row['candidate_classes'].replace('|', ', ')}{detail}") else: - console.print("\n[green]No gap candidates: every unknown file is metadata-free.[/green]") + console.print("\n[green]No review candidates in this run.[/green]") if __name__ == "__main__": diff --git a/scripts/sidecar_regression.py b/scripts/sidecar_regression.py index ee6bc1c..2fc7984 100644 --- a/scripts/sidecar_regression.py +++ b/scripts/sidecar_regression.py @@ -59,7 +59,7 @@ OUT = REPO / ".local-eval" / "sidecar-regression.jsonl" # Map a watermark description to a stable behavior family. Order matters: the first # matching pattern wins, so put the specific tokens above the generic ones. -_FAMILIES: tuple[tuple[str, tuple[str, ...]], ...] = ( +_LEGACY_FAMILIES: tuple[tuple[str, tuple[str, ...]], ...] = ( ("synthid", ("synthid",)), ("visible_sparkle", ("sparkle",)), ("visible_doubao", ("豆包", "doubao")), @@ -71,27 +71,61 @@ _FAMILIES: tuple[tuple[str, tuple[str, ...]], ...] = ( ("trustmark", ("trustmark",)), ("open_invisible", ("open invisible", "dwt", "stable diffusion xl")), ("xai_signature", ("xai", "grok signature")), + ("samsung_genai", ("genaitype", "photoeditor_re_edit")), + ("soft_binding", ("soft binding", "soft-binding", "invismark")), + ("c2pa_cloud", ("cloud manifest",)), + ("hf_job", ("hf-job", "hugging face job")), ("exif_generator", ("exif", "software tag", "png text")), ) +def _registered_visible_family(description: str) -> str | None: + """Map current mark labels from the registry instead of copying its keys here.""" + from remove_ai_watermarks.watermark_registry import known_marks + + low = description.lower() + matches = [ + (mark.label.lower() in low, len(mark.label), mark) + for mark in known_marks() + if mark.key != "gemini" and (mark.label.lower() in low or mark.key.replace("_", " ") in low) + ] + if not matches: + return None + _, _, mark = max(matches, key=lambda candidate: candidate[:2]) + return f"visible_{mark.key}" + + def family_of(description: str) -> str: low = description.lower() - for name, tokens in _FAMILIES: + if family := _registered_visible_family(description): + return family + for name, tokens in _LEGACY_FAMILIES: if any(t in low for t in tokens): return name return "other" def families(descriptions: list[str]) -> set[str]: - return {family_of(d) for d in descriptions or []} + return {family for description in descriptions or [] if (family := family_of(description)) != "other"} + + +def sidecar_families(sidecar: dict) -> set[str]: + """Combine stable names with legacy-only families such as SynthID.""" + signals = sidecar.get("signals") or [] + return {str(signal) for signal in signals} | families(sidecar.get("watermarks") or []) + + +def report_families(report: object) -> set[str]: + """Combine current Signal.name values with watermark-only families.""" + signal_names = {str(signal.name) for signal in getattr(report, "signals", []) or []} + return signal_names | families(list(getattr(report, "watermarks", []) or [])) def compare(sidecar: dict, report: object) -> dict: """Classify the difference between a recorded verdict and a fresh one.""" old_ai, new_ai = sidecar.get("is_ai_generated"), getattr(report, "is_ai_generated", None) - old_fam = families(sidecar.get("watermarks") or []) - new_fam = families(list(getattr(report, "watermarks", []) or [])) + old_fam = sidecar_families(sidecar) + new_fam = report_families(report) old_plat, new_plat = sidecar.get("platform"), getattr(report, "platform", None) old_conf, new_conf = sidecar.get("confidence"), getattr(report, "confidence", None) diff --git a/tests/test_signal_discovery_scripts.py b/tests/test_signal_discovery_scripts.py new file mode 100644 index 0000000..b2b9e11 --- /dev/null +++ b/tests/test_signal_discovery_scripts.py @@ -0,0 +1,76 @@ +"""Contracts for the signal-discovery corpus harnesses.""" + +from __future__ import annotations + +import sys +from datetime import date +from pathlib import Path +from types import SimpleNamespace + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +import corpus_gap_scan as gap +import sidecar_regression as regression + + +def _report(**overrides): + values = { + "is_ai_generated": None, + "platform": None, + "signals": [], + "integrity_clashes": [], + } + values.update(overrides) + return SimpleNamespace(**values) + + +def test_gap_markers_are_case_insensitive(): + assert "c2pa" in gap._marker_hits(b"UPPERCASE C2PA MANIFEST") + + +def test_gap_candidates_cover_more_than_blind_unknowns(): + signal = SimpleNamespace(name="iptc") + assert gap._candidate_classes(_report(signals=[signal]), []) == ["unattributed_signal"] + assert gap._candidate_classes(_report(is_ai_generated=True), []) == ["unattributed_ai"] + assert gap._candidate_classes(_report(), ["SynthID"]) == ["blind_marker"] + + +def test_since_filter_uses_corpus_date_directories(tmp_path): + old = tmp_path / "2026-09-01" / "old.png" + current = tmp_path / "2026-09-10" / "current.png" + undated = tmp_path / "loose.png" + for path in (old, current, undated): + path.parent.mkdir(parents=True, exist_ok=True) + path.touch() + assert gap._files(tmp_path, date(2026, 9, 10)) == [current] + + +def test_sidecar_regression_prefers_stable_signal_names(): + sidecar = {"signals": ["visible_qwen"], "watermarks": ["wording may change"]} + assert regression.sidecar_families(sidecar) == {"visible_qwen"} + + +def test_current_report_uses_signal_names_not_watermark_prose(): + report = SimpleNamespace( + signals=[SimpleNamespace(name="visible_liblib")], + watermarks=["new wording not known to the legacy mapper"], + ) + assert regression.report_families(report) == {"visible_liblib"} + + +def test_current_report_keeps_watermark_only_synthid_family(): + report = SimpleNamespace(signals=[], watermarks=["SynthID watermark (Google)"]) + assert regression.report_families(report) == {"synthid"} + + +def test_registry_drives_current_visible_families(): + from remove_ai_watermarks.watermark_registry import known_marks + + for mark in known_marks(): + expected = "visible_sparkle" if mark.key == "gemini" else f"visible_{mark.key}" + assert regression.family_of(mark.label) == expected + + +def test_registry_keys_cover_legacy_visible_wording(): + assert regression.family_of("Qwen 千问 AI label") == "visible_qwen" + assert regression.family_of("Microsoft AI badge") == "visible_microsoft"