diff --git a/docs/benchmark-kernel.md b/docs/benchmark-kernel.md index 2a652ed..11e79c7 100644 --- a/docs/benchmark-kernel.md +++ b/docs/benchmark-kernel.md @@ -70,6 +70,13 @@ dirty, and SHA-256 digests of the benchmark kernel and adapter source. Detector dependency absence is `unavailable`; an undecodable artifact or adapter exception is `error`. Neither is counted as `not_detected`. +`detection.adapter_elapsed_ms` measures the wall time of the named detector +adapter call only. Image decoding and fidelity calculation are outside that +interval. The value is `null` when the adapter was unavailable or the artifact +could not be decoded, because no detector call occurred. Preserve the first +call separately when aggregating: it can include lazy imports and model loading, +while subsequent rows describe warm validation cost. + Fidelity is calculated only when a reference is present, both decoded images are 8-bit, and their dimensions match. Other bit depths remain explicitly incomparable until their sample-range contract is defined. The result records @@ -82,3 +89,103 @@ Schema v1 deliberately has no aggregate score, confidence interval, video adapter, automatic attack generator, removal runner, or provider oracle. Add those only after the case-level evidence is large enough to define and test the corresponding statistical or media contract. + +## Synthetic cohort v1 + +`scripts/watermark_benchmark_cohort.py` builds a reproducible local smoke cohort +without a provider API or provenance oracle. Its default 65 cases cover three +deterministic 512-pixel carriers, SDXL, FLUX, and Stable Diffusion 1.x DWT-DCT +patterns, TrustMark Variant P with schema 1, and JPEG quality 90, 75% resize +round-trip, and 5% crop-per-edge round-trip attacks. Each adapter also receives +a high-frequency hard negative. TrustMark cases include the package remover's +output as a separate `removed` state. + +```bash +uv run python scripts/watermark_benchmark_cohort.py \ + --output-dir .local-eval/watermark-benchmark-cohort-v1 +uv run python scripts/watermark_benchmark.py \ + .local-eval/watermark-benchmark-cohort-v1/manifest.jsonl \ + --output .local-eval/watermark-benchmark-cohort-v1-results.jsonl +``` + +The builder refuses an existing destination, hashes every generated file, pins +its own recipe digest and dependency versions, and validates the finished +manifest through the benchmark's strict loader. It reads the DWT-DCT patterns +from the production detector so the generated positive cannot silently drift +from the matcher. TrustMark's optional package and model weights must already be +available. + +Attacked and removed rows intentionally use `expected: unresolved`: they are +measurements, not assertions smuggled into fixtures. The cohort is too small +and too synthetic for prevalence, confidence, or release-quality claims. It is +a deterministic first baseline that proves the harness and exposes gross +detector behavior before a licensed, stratified real-image cohort is added. + +### Development baseline, 2026-09-05 + +The first local v1 run used cohort recipe +`sha256:d98fd6402bcc149833005ec03aab77bd0074ee105696f9e1ac5da01ef9c666f7` +and kernel +`sha256:37e552ba53c3d2aff2c5f9e160a33ad47b0d32c877b0a8fbba39c90ca0ab0179`. +Dependencies were invisible-watermark 0.2.0, TrustMark 0.9.1, NumPy 1.26.4, +and Pillow 12.3.0. The tree was based on commit `ad53790` with the cohort and +timing changes still uncommitted, so the content hashes, not that commit alone, +identify the executed code. A second build in a different directory produced a +byte-identical manifest and artifact-hash inventory. + +| Adapter and state | Unique artifacts | Detected | Not detected | +| --- | ---: | ---: | ---: | +| DWT-DCT clean | 4 | 0 | 4 | +| DWT-DCT marked | 9 | 6 | 3 | +| DWT-DCT attacked | 27 | 13 | 14 | +| TrustMark clean | 4 | 0 | 4 | +| TrustMark marked | 3 | 2 | 1 | +| TrustMark attacked | 9 | 4 | 5 | +| TrustMark removed | 3 | 0 | 3 | + +All three plain-gradient DWT-DCT positives were missed, while all texture and +low-detail positives were recognized. DWT-DCT retained 7/9 positives through +JPEG, 6/9 through resize, and 0/9 through crop. TrustMark retained 2/3 through +JPEG and 1/3 through each resize and crop. Its three remover outputs produced no +recognized signal and measured 46.17-49.24 dB PSNR against their clean +carriers. Those three negatives are a post-removal observation, not proof that +the signal is absent beyond this decoder. + +Ten independent sequential benchmark processes reproduced every case status. +Nearest-rank adapter timings were: + +| Adapter call | n | p50 | p90 | max | +| --- | ---: | ---: | ---: | ---: | +| DWT-DCT cold, first call per process | 10 | 14.96 ms | 19.86 ms | 27.88 ms | +| DWT-DCT warm | 450 | 0.65 ms | 1.64 ms | 13.53 ms | +| TrustMark cold, first call per process | 10 | 1,989.47 ms | 2,085.62 ms | 2,086.02 ms | +| TrustMark warm | 180 | 17.96 ms | 34.20 ms | 39.95 ms | +| TrustMark warm, removed state only | 30 | 16.80 ms | 17.69 ms | 18.56 ms | + +These intervals include only the detector adapter call. They exclude image +decoding, fidelity metrics, removal, and process startup. The roughly two-second +TrustMark cold start matters only when validation is the first TrustMark use in +the process; a workflow that already identified the source pays the warm cost. +The sample is deliberately diagnostic rather than statistically representative. + +## Aggregate repeated runs + +`scripts/watermark_benchmark_report.py` turns one or more result JSONL files +into a standalone Markdown report. Treat each input file as one independent +process run: the first measured call for each adapter in that file is cold, and +the remaining measured calls are warm. + +```bash +uv run python scripts/watermark_benchmark_report.py \ + .local-eval/results/run-01.jsonl \ + .local-eval/results/run-02.jsonl \ + --output .local-eval/results/report.md +``` + +The report retains raw observation counts but deduplicates detector summaries +by artifact SHA-256 within each adapter and state. It reports state and +transform breakdowns, post-removal observations, nearest-rank timing, fidelity, +expected-result mismatches, unstable repeated verdicts, errors, and unavailable +adapters as separate evidence. A repeated `case_id` whose artifact, transform, +or other case identity changes is rejected rather than merged. Input file +digests in the final report bind every aggregate to its exact case-level source. diff --git a/scripts/README.md b/scripts/README.md index 805be62..e30f9e0 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -26,6 +26,7 @@ explicitly names a tracked canonical result. | `visible_eval.py` | Benchmark registered visible-mark detectors. | | `visible_removal_audit.py` | Audit visible-removal results over a local corpus. | | `watermark_benchmark.py` | Run hash-pinned image cases while keeping detection, removal observation, and fidelity separate. | +| `watermark_benchmark_report.py` | Aggregate repeated benchmark results with artifact-aware counts and separate cold/warm timing. | ## Calibration and corpus preparation @@ -44,6 +45,7 @@ explicitly names a tracked canonical result. | `visible_positives.py` | List corpus images carrying a registered visible mark. | | `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`. | | `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. | | `verify_engine_selection_fixtures.py` | Verify hashes, dimensions, and prompt pairing in the tracked auto-engine content matrix. | diff --git a/scripts/watermark_benchmark.py b/scripts/watermark_benchmark.py index d23bb9b..153c90f 100644 --- a/scripts/watermark_benchmark.py +++ b/scripts/watermark_benchmark.py @@ -32,6 +32,7 @@ import shutil import subprocess import sys import tempfile +import time from collections import Counter from dataclasses import dataclass from functools import cache, lru_cache @@ -59,6 +60,7 @@ DetectorStatus = Literal["detected", "not_detected", "unavailable", "error"] _ARMS: tuple[Arm, ...] = ("positive", "matched_negative", "wrong_key", "hard_negative") _STATES: tuple[State, ...] = ("clean", "marked", "attacked", "removed") _EXPECTED: tuple[ExpectedDetection, ...] = ("detected", "not_detected", "unresolved") +DETECTOR_STATUSES: tuple[DetectorStatus, ...] = ("detected", "not_detected", "unavailable", "error") _FIELDS = { "schema_version", "case_id", @@ -146,7 +148,8 @@ class FunctionAdapter: return DetectorOutcome(status="detected" if label is not None else "not_detected", label=label) -def _sha256(path: Path) -> str: +def sha256_file(path: Path) -> str: + """Return the lowercase SHA-256 digest of one file.""" digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): @@ -156,7 +159,7 @@ def _sha256(path: Path) -> str: @cache def _code_sha256(path: Path) -> str: - return _sha256(path) + return sha256_file(path) def _reject_json_constant(value: str) -> None: @@ -172,14 +175,25 @@ def _unique_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: return result -def _nonempty_string(value: object, *, field: str, location: str) -> str: +def parse_strict_json(value: str) -> object: + """Parse JSON while rejecting duplicate fields and non-finite numbers.""" + return json.loads( + value, + object_pairs_hook=_unique_json_object, + parse_constant=_reject_json_constant, + ) + + +def require_nonempty_string(value: object, *, field: str, location: str) -> str: + """Return a validated non-empty string or raise with its source location.""" if not isinstance(value, str) or not value.strip(): raise ValueError(f"{location}: {field} must be a non-empty string") return value -def _digest(value: object, *, field: str, location: str) -> str: - digest = _nonempty_string(value, field=field, location=location) +def require_sha256(value: object, *, field: str, location: str) -> str: + """Return a validated lowercase SHA-256 digest.""" + digest = require_nonempty_string(value, field=field, location=location) if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest): raise ValueError(f"{location}: {field} must be a lowercase sha256") return digest @@ -194,8 +208,8 @@ def _artifact( location: str, digest_cache: dict[Path, str], ) -> tuple[Path, str]: - raw_path = _nonempty_string(path_value, field=f"{prefix}path", location=location) - expected_digest = _digest(digest_value, field=f"{prefix}sha256", location=location) + raw_path = require_nonempty_string(path_value, field=f"{prefix}path", location=location) + expected_digest = require_sha256(digest_value, field=f"{prefix}sha256", location=location) path = Path(raw_path) if not path.is_absolute(): path = manifest.parent / path @@ -204,7 +218,7 @@ def _artifact( raise ValueError(f"{location}: {prefix}path is not a file: {path}") actual_digest = digest_cache.get(path) if actual_digest is None: - actual_digest = _sha256(path) + actual_digest = sha256_file(path) digest_cache[path] = actual_digest if actual_digest != expected_digest: raise ValueError( @@ -219,8 +233,8 @@ def _transform(value: object, *, location: str) -> Transform: transform = cast("dict[str, object]", value) if set(transform) != _TRANSFORM_FIELDS: raise ValueError(f"{location}: transform must contain exactly name, revision, and parameters") - name = _nonempty_string(transform["name"], field="transform.name", location=location) - revision = _nonempty_string(transform["revision"], field="transform.revision", location=location) + name = require_nonempty_string(transform["name"], field="transform.name", location=location) + revision = require_nonempty_string(transform["revision"], field="transform.revision", location=location) parameters = transform["parameters"] if not isinstance(parameters, dict): raise ValueError(f"{location}: transform.parameters must be an object") @@ -290,17 +304,17 @@ def _case( ) return BenchmarkCase( - case_id=_nonempty_string(values["case_id"], field="case_id", location=location), - pair_id=_nonempty_string(values["pair_id"], field="pair_id", location=location), + case_id=require_nonempty_string(values["case_id"], field="case_id", location=location), + pair_id=require_nonempty_string(values["pair_id"], field="pair_id", location=location), media_type="image", - adapter=_nonempty_string(values["adapter"], field="adapter", location=location), + adapter=require_nonempty_string(values["adapter"], field="adapter", location=location), arm=arm, state=state, path=path, sha256=sha256, reference_path=reference_path, reference_sha256=reference_sha256, - source_revision=_nonempty_string(values["source_revision"], field="source_revision", location=location), + source_revision=require_nonempty_string(values["source_revision"], field="source_revision", location=location), transform=_transform(values["transform"], location=location), seed=seed, expected=expected, @@ -317,11 +331,7 @@ def load_manifest(path: Path) -> list[BenchmarkCase]: if not line.strip(): continue try: - raw = json.loads( - line, - object_pairs_hook=_unique_json_object, - parse_constant=_reject_json_constant, - ) + raw = parse_strict_json(line) except (json.JSONDecodeError, ValueError) as exc: detail = exc.msg if isinstance(exc, json.JSONDecodeError) else str(exc) raise ValueError(f"{path}:{line_number}: invalid JSON: {detail}") from exc @@ -449,15 +459,17 @@ def _image_fidelity( def _detect(adapter: DetectorAdapter, path: Path, image: NDArray[Any]) -> DetectorOutcome: try: - if not adapter.available: - return DetectorOutcome(status="unavailable", label=None) return adapter.detect(path, image) except Exception as exc: log.warning("Benchmark adapter %s failed for %s: %s", adapter.name, path, exc) return DetectorOutcome(status="error", label=None, error=f"{type(exc).__name__}: {exc}") -def _detection_record(outcome: DetectorOutcome, expected: ExpectedDetection) -> dict[str, Any]: +def _detection_record( + outcome: DetectorOutcome, + expected: ExpectedDetection, + adapter_elapsed_ms: float | None, +) -> dict[str, Any]: comparable = outcome.status in ("detected", "not_detected") and expected != "unresolved" record: dict[str, Any] = { "status": outcome.status, @@ -465,6 +477,7 @@ def _detection_record(outcome: DetectorOutcome, expected: ExpectedDetection) -> "expected": expected, "matches_expected": outcome.status == expected if comparable else None, "positive_evidence": outcome.status == "detected", + "adapter_elapsed_ms": adapter_elapsed_ms, } if outcome.error is not None: record["error"] = outcome.error @@ -493,6 +506,7 @@ def evaluate_case( adapters: Mapping[str, DetectorAdapter], repository: Mapping[str, str | bool], reference_decoder: Callable[[Path], NDArray[Any] | None] = _decode_image, + clock_ns: Callable[[], int] = time.perf_counter_ns, ) -> dict[str, Any]: """Evaluate one case while keeping detector and fidelity claims distinct.""" try: @@ -500,11 +514,15 @@ def evaluate_case( except KeyError as exc: raise ValueError(f"unknown adapter {case.adapter!r}") from exc artifact = _decode_image(case.path) - outcome = ( - _detect(adapter, case.path, artifact) - if artifact is not None - else DetectorOutcome(status="error", label=None, error="artifact could not be decoded") - ) + adapter_elapsed_ms: float | None = None + if artifact is None: + outcome = DetectorOutcome(status="error", label=None, error="artifact could not be decoded") + elif not adapter.available: + outcome = DetectorOutcome(status="unavailable", label=None) + else: + started_ns = clock_ns() + outcome = _detect(adapter, case.path, artifact) + adapter_elapsed_ms = (clock_ns() - started_ns) / 1e6 return { "schema_version": SCHEMA_VERSION, "case_id": case.case_id, @@ -529,7 +547,7 @@ def evaluate_case( "kernel_source_sha256": _code_sha256(Path(__file__)), "adapter_source_sha256": _code_sha256(adapter.source_file), }, - "detection": _detection_record(outcome, case.expected), + "detection": _detection_record(outcome, case.expected, adapter_elapsed_ms), "removal": _removal_record(case.state, outcome), "fidelity": _image_fidelity(case, artifact, reference_decoder), } diff --git a/scripts/watermark_benchmark_cohort.py b/scripts/watermark_benchmark_cohort.py new file mode 100755 index 0000000..21d73b3 --- /dev/null +++ b/scripts/watermark_benchmark_cohort.py @@ -0,0 +1,584 @@ +#!/usr/bin/env python3 +"""Build a local, synthetic image-watermark benchmark cohort. + +The builder creates deterministic clean carriers, matched DWT-DCT and TrustMark +embeds, fixed attack transforms, TrustMark remover outputs, and hard negatives. +It writes every generated artifact once under a new output directory and emits +the strict JSONL manifest consumed by ``watermark_benchmark.py``. + +Generated artifacts and reports belong under ``.local-eval/`` and are never +committed. No provider oracle or API is used. + + uv run python scripts/watermark_benchmark_cohort.py \ + --output-dir .local-eval/watermark-benchmark-cohort-v1 +""" + +from __future__ import annotations + +import argparse +import hashlib +import io +import logging +import sys +from dataclasses import dataclass +from functools import cache +from importlib import import_module +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal, cast + +from PIL import Image + +if TYPE_CHECKING: + from collections.abc import Sequence + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO / "src")) +sys.path.insert(0, str(REPO / "scripts")) + +from watermark_benchmark import SCHEMA_VERSION, load_manifest, sha256_file, write_jsonl # noqa: E402 + +log = logging.getLogger(__name__) + +RECIPE_VERSION = "watermark-benchmark-cohort-v1" +AdapterName = Literal["dwt-dct", "trustmark"] +AttackName = Literal["jpeg-q90", "resize-75", "crop-5"] + +DEFAULT_CARRIERS = ( + # The content spread is deliberate: the legacy DWT-DCT decoder is strongly + # carrier-dependent even before an attack. + # Keep seeds explicit so adding a carrier never changes an existing one. + ("texture", 7), + ("gradient", 11), + ("low-detail", 19), +) +DEFAULT_DWT_SCHEMES = ("sdxl", "flux", "sd1") +DEFAULT_ATTACKS: tuple[AttackName, ...] = ("jpeg-q90", "resize-75", "crop-5") +_ADAPTERS: tuple[AdapterName, ...] = ("dwt-dct", "trustmark") +_DWT_SCHEMES = frozenset(DEFAULT_DWT_SCHEMES) +_ATTACKS = frozenset(DEFAULT_ATTACKS) + + +@dataclass(frozen=True) +class CarrierSpec: + """Identity of one deterministic synthetic carrier.""" + + name: str + seed: int + + +def trustmark_available() -> bool: + """Return whether the optional TrustMark package can be imported.""" + from remove_ai_watermarks.trustmark_detector import is_available + + return is_available() + + +@cache +def _artifact_sha256(path: Path) -> str: + """Cache hashes for immutable artifacts created by this builder.""" + return sha256_file(path) + + +def _package_version(name: str) -> str: + try: + return version(name) + except PackageNotFoundError: + return "unavailable" + + +def _dependencies() -> dict[str, str]: + return { + "numpy": _package_version("numpy"), + "pillow": _package_version("pillow"), + "invisible-watermark": _package_version("invisible-watermark"), + "trustmark": _package_version("trustmark"), + } + + +def _source_revision() -> str: + return f"{RECIPE_VERSION}@sha256:{_artifact_sha256(Path(__file__))}" + + +def _carrier(spec: CarrierSpec, *, size: int = 512) -> Image.Image: + import numpy as np + from PIL import ImageDraw + + if spec.name == "texture": + y, x = np.mgrid[:size, :size] + rng = np.random.default_rng(spec.seed) + pixels = np.stack( + ( + x * 255 // (size - 1), + y * 255 // (size - 1), + (x + y) * 127 // (size - 1), + ), + axis=2, + ).astype(np.int16) + pixels += cast("Any", rng).integers(-20, 21, pixels.shape, dtype=np.int16) + return Image.fromarray(np.clip(pixels, 0, 255).astype(np.uint8), "RGB") + + if spec.name == "gradient": + y, x = np.mgrid[:size, :size] + pixels = np.stack( + ( + 32 + x * 190 // (size - 1), + 48 + y * 160 // (size - 1), + 64 + (x + y) * 140 // (2 * size - 2), + ), + axis=2, + ).astype(np.uint8) + image = Image.fromarray(pixels, "RGB") + draw = ImageDraw.Draw(image) + draw.ellipse((96, 112, 304, 320), fill=(205, 96, 72), outline=(245, 230, 210), width=9) + draw.rectangle((330, 72, 455, 430), fill=(58, 121, 184), outline=(230, 240, 250), width=7) + return image + + if spec.name == "low-detail": + rng = np.random.default_rng(spec.seed) + bands = cast("Any", rng).integers(0, 9, (16, 16, 3), dtype=np.uint8) + pixels = np.repeat(np.repeat(bands, size // 16, axis=0), size // 16, axis=1) + return Image.fromarray(np.clip(pixels.astype(np.int16) + 220, 0, 255).astype(np.uint8), "RGB") + + if spec.name == "hard-checker": + y, x = np.mgrid[:size, :size] + checker = ((x // 4 + y // 4 + spec.seed) % 2) * 255 + pixels = np.stack((checker, np.roll(checker, 2, axis=0), np.roll(checker, 2, axis=1)), axis=2) + return Image.fromarray(pixels.astype(np.uint8), "RGB") + + raise ValueError(f"unknown carrier {spec.name!r}") + + +def _save(image: Image.Image, path: Path) -> Path: + image.convert("RGB").save(path, format="PNG") + return path + + +def _attack(image: Image.Image, attack: AttackName) -> Image.Image: + image = image.convert("RGB") + if attack == "jpeg-q90": + buffer = io.BytesIO() + image.save(buffer, format="JPEG", quality=90, subsampling=0) + buffer.seek(0) + with Image.open(buffer) as decoded: + return decoded.convert("RGB").copy() + if attack == "resize-75": + reduced = image.resize((384, 384), Image.Resampling.LANCZOS) + return reduced.resize(image.size, Image.Resampling.LANCZOS) + if attack == "crop-5": + margin_x = round(image.width * 0.05) + margin_y = round(image.height * 0.05) + cropped = image.crop((margin_x, margin_y, image.width - margin_x, image.height - margin_y)) + return cropped.resize(image.size, Image.Resampling.LANCZOS) + raise ValueError(f"unknown attack {attack!r}") + + +def _embed_dwt_dct(image: Image.Image, scheme: str) -> Image.Image: + import cv2 + import numpy as np + + watermark_patterns = import_module("remove_ai_watermarks.invisible_watermark") + patterns = vars(watermark_patterns) + bits_48 = cast("dict[str, int]", patterns["_BITS_48"]) + sd1_string = cast("bytes", patterns["_SD1_STRING"]) + WatermarkEncoder = import_module("imwatermark").WatermarkEncoder + encoder = WatermarkEncoder() + if scheme == "sdxl": + encoder.set_watermark("bits", [int(bit) for bit in format(bits_48["Stable Diffusion XL"], "048b")]) + elif scheme == "flux": + encoder.set_watermark( + "bits", + [int(bit) for bit in format(bits_48["FLUX.2 (Black Forest Labs)"], "048b")], + ) + elif scheme == "sd1": + encoder.set_watermark("bytes", sd1_string) + else: + raise ValueError(f"unknown DWT-DCT scheme {scheme!r}") + + rgb = np.asarray(image.convert("RGB")) + marked_bgr = np.asarray(encoder.encode(cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR), "dwtDct"), dtype=np.uint8) + return Image.fromarray(cv2.cvtColor(marked_bgr, cv2.COLOR_BGR2RGB), "RGB") + + +def _trustmark_runtime() -> Any: + if not trustmark_available(): + raise RuntimeError("TrustMark cohort requested but the trustmark extra is unavailable") + + TrustMark = import_module("trustmark").TrustMark + return TrustMark( + verbose=False, + model_type="P", + encoding_type=TrustMark.Encoding.BCH_5, + loadRemover=True, + ) + + +def _relative(path: Path, root: Path) -> str: + return path.relative_to(root).as_posix() + + +def _row( + *, + root: Path, + case_id: str, + pair_id: str, + adapter: AdapterName, + arm: str, + state: str, + path: Path, + reference: Path | None, + source_revision: str, + transform_name: str, + transform_revision: str, + parameters: dict[str, Any], + seed: int, + expected: str, +) -> dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "case_id": case_id, + "pair_id": pair_id, + "media_type": "image", + "adapter": adapter, + "arm": arm, + "state": state, + "path": _relative(path, root), + "sha256": _artifact_sha256(path), + "reference_path": _relative(reference, root) if reference is not None else None, + "reference_sha256": _artifact_sha256(reference) if reference is not None else None, + "source_revision": source_revision, + "transform": { + "name": transform_name, + "revision": transform_revision, + "parameters": parameters, + }, + "seed": seed, + "expected": expected, + } + + +def _attack_parameters(attack: AttackName) -> dict[str, Any]: + if attack == "jpeg-q90": + values: dict[str, Any] = {"quality": 90, "subsampling": 0} + elif attack == "resize-75": + values = {"scale": 0.75, "restore_size": True, "resampling": "lanczos"} + else: + values = {"crop_fraction_per_edge": 0.05, "restore_size": True, "resampling": "lanczos"} + return values + + +def _append_hard_negative( + rows: list[dict[str, Any]], + *, + root: Path, + artifacts: Path, + adapter: AdapterName, + source_revision: str, + dependencies: dict[str, str], +) -> None: + spec = CarrierSpec("hard-checker", 101) + path = artifacts / "hard-checker.png" + if not path.exists(): + _save(_carrier(spec), path) + rows.append( + _row( + root=root, + case_id=f"{adapter}--hard-checker--clean", + pair_id=f"{adapter}--hard-checker", + adapter=adapter, + arm="hard_negative", + state="clean", + path=path, + reference=None, + source_revision=source_revision, + transform_name="synthetic-carrier", + transform_revision=RECIPE_VERSION, + parameters={"carrier": spec.name, "dependencies": dependencies}, + seed=spec.seed, + expected="not_detected", + ) + ) + + +def _append_dwt_rows( + rows: list[dict[str, Any]], + *, + root: Path, + artifacts: Path, + carriers: Sequence[tuple[CarrierSpec, Image.Image]], + schemes: Sequence[str], + attacks: Sequence[AttackName], + source_revision: str, + dependencies: dict[str, str], +) -> None: + for spec, clean in carriers: + clean_path = _save(clean, artifacts / f"{spec.name}--clean.png") + for scheme in schemes: + pair_id = f"dwt-dct--{scheme}--{spec.name}" + base_parameters = {"scheme": scheme, "carrier": spec.name, "dependencies": dependencies} + rows.append( + _row( + root=root, + case_id=f"{pair_id}--clean", + pair_id=pair_id, + adapter="dwt-dct", + arm="matched_negative", + state="clean", + path=clean_path, + reference=clean_path, + source_revision=source_revision, + transform_name="synthetic-carrier", + transform_revision=RECIPE_VERSION, + parameters=base_parameters, + seed=spec.seed, + expected="not_detected", + ) + ) + marked = _embed_dwt_dct(clean, scheme) + marked_path = _save(marked, artifacts / f"{spec.name}--dwt-{scheme}--marked.png") + rows.append( + _row( + root=root, + case_id=f"{pair_id}--marked", + pair_id=pair_id, + adapter="dwt-dct", + arm="positive", + state="marked", + path=marked_path, + reference=clean_path, + source_revision=source_revision, + transform_name="dwt-dct-embed", + transform_revision=f"invisible-watermark@{dependencies['invisible-watermark']}", + parameters=base_parameters, + seed=spec.seed, + expected="detected", + ) + ) + for attack in attacks: + attacked_path = _save( + _attack(marked, attack), + artifacts / f"{spec.name}--dwt-{scheme}--{attack}.png", + ) + rows.append( + _row( + root=root, + case_id=f"{pair_id}--{attack}", + pair_id=pair_id, + adapter="dwt-dct", + arm="positive", + state="attacked", + path=attacked_path, + reference=clean_path, + source_revision=source_revision, + transform_name=f"dwt-dct-embed-then-{attack}", + transform_revision=RECIPE_VERSION, + parameters=base_parameters | _attack_parameters(attack), + seed=spec.seed, + expected="unresolved", + ) + ) + + +def _append_trustmark_rows( + rows: list[dict[str, Any]], + *, + root: Path, + artifacts: Path, + carriers: Sequence[tuple[CarrierSpec, Image.Image]], + attacks: Sequence[AttackName], + source_revision: str, + dependencies: dict[str, str], +) -> None: + runtime = _trustmark_runtime() + for spec, clean in carriers: + clean_path = artifacts / f"{spec.name}--clean.png" + if not clean_path.exists(): + _save(clean, clean_path) + pair_id = f"trustmark--p-schema1--{spec.name}" + base_parameters = { + "variant": "P", + "schema": 1, + "carrier": spec.name, + "dependencies": dependencies, + } + rows.append( + _row( + root=root, + case_id=f"{pair_id}--clean", + pair_id=pair_id, + adapter="trustmark", + arm="matched_negative", + state="clean", + path=clean_path, + reference=clean_path, + source_revision=source_revision, + transform_name="synthetic-carrier", + transform_revision=RECIPE_VERSION, + parameters=base_parameters, + seed=spec.seed, + expected="not_detected", + ) + ) + payload = format(spec.seed, "061b") + marked = runtime.encode(clean, payload, MODE="binary") + marked_path = _save(marked, artifacts / f"{spec.name}--trustmark-p--marked.png") + rows.append( + _row( + root=root, + case_id=f"{pair_id}--marked", + pair_id=pair_id, + adapter="trustmark", + arm="positive", + state="marked", + path=marked_path, + reference=clean_path, + source_revision=source_revision, + transform_name="trustmark-embed", + transform_revision=f"trustmark@{dependencies['trustmark']}", + parameters=base_parameters | {"payload_sha256": hashlib.sha256(payload.encode()).hexdigest()}, + seed=spec.seed, + expected="detected", + ) + ) + for attack in attacks: + attacked_path = _save( + _attack(marked, attack), + artifacts / f"{spec.name}--trustmark-p--{attack}.png", + ) + rows.append( + _row( + root=root, + case_id=f"{pair_id}--{attack}", + pair_id=pair_id, + adapter="trustmark", + arm="positive", + state="attacked", + path=attacked_path, + reference=clean_path, + source_revision=source_revision, + transform_name=f"trustmark-embed-then-{attack}", + transform_revision=RECIPE_VERSION, + parameters=base_parameters | _attack_parameters(attack), + seed=spec.seed, + expected="unresolved", + ) + ) + removed_path = _save( + runtime.remove_watermark(marked), + artifacts / f"{spec.name}--trustmark-p--removed.png", + ) + rows.append( + _row( + root=root, + case_id=f"{pair_id}--removed", + pair_id=pair_id, + adapter="trustmark", + arm="positive", + state="removed", + path=removed_path, + reference=clean_path, + source_revision=source_revision, + transform_name="trustmark-remove", + transform_revision=f"trustmark@{dependencies['trustmark']}", + parameters=base_parameters | {"input": "matched-marked"}, + seed=spec.seed, + expected="unresolved", + ) + ) + + +def build_cohort( + output_dir: Path, + *, + adapters: Sequence[AdapterName] = _ADAPTERS, + carriers: Sequence[CarrierSpec] = tuple(CarrierSpec(name, seed) for name, seed in DEFAULT_CARRIERS), + dwt_schemes: Sequence[str] = DEFAULT_DWT_SCHEMES, + attacks: Sequence[AttackName] = DEFAULT_ATTACKS, +) -> Path: + """Build and validate a new local cohort directory, returning its manifest.""" + if output_dir.exists(): + raise FileExistsError(f"output directory already exists: {output_dir}") + unknown_adapters = sorted(set(adapters) - set(_ADAPTERS)) + if not adapters or unknown_adapters: + detail = ", ".join(unknown_adapters) if unknown_adapters else "none selected" + raise ValueError(f"invalid adapters: {detail}") + unknown_schemes = sorted(set(dwt_schemes) - _DWT_SCHEMES) + if unknown_schemes: + raise ValueError(f"unknown DWT-DCT schemes: {', '.join(unknown_schemes)}") + unknown_attacks = sorted(set(attacks) - _ATTACKS) + if unknown_attacks: + raise ValueError(f"unknown attacks: {', '.join(unknown_attacks)}") + + output_dir.mkdir(parents=True) + artifacts = output_dir / "artifacts" + artifacts.mkdir() + rows: list[dict[str, Any]] = [] + source_revision = _source_revision() + dependencies = _dependencies() + carrier_images = tuple((spec, _carrier(spec)) for spec in carriers) + + if "dwt-dct" in adapters: + _append_dwt_rows( + rows, + root=output_dir, + artifacts=artifacts, + carriers=carrier_images, + schemes=dwt_schemes, + attacks=attacks, + source_revision=source_revision, + dependencies=dependencies, + ) + _append_hard_negative( + rows, + root=output_dir, + artifacts=artifacts, + adapter="dwt-dct", + source_revision=source_revision, + dependencies=dependencies, + ) + if "trustmark" in adapters: + _append_trustmark_rows( + rows, + root=output_dir, + artifacts=artifacts, + carriers=carrier_images, + attacks=attacks, + source_revision=source_revision, + dependencies=dependencies, + ) + _append_hard_negative( + rows, + root=output_dir, + artifacts=artifacts, + adapter="trustmark", + source_revision=source_revision, + dependencies=dependencies, + ) + + manifest = output_dir / "manifest.jsonl" + write_jsonl(manifest, rows) + load_manifest(manifest) + return manifest + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output-dir", type=Path, required=True, help="new local cohort directory") + parser.add_argument( + "--adapter", + action="append", + choices=_ADAPTERS, + dest="adapters", + help="adapter to include; repeat to select both (default: both)", + ) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + try: + manifest = build_cohort(args.output_dir, adapters=tuple(args.adapters or _ADAPTERS)) + except (FileExistsError, ImportError, OSError, RuntimeError, ValueError) as exc: + parser.error(str(exc)) + log.info("Wrote validated benchmark cohort to %s", manifest) + + +if __name__ == "__main__": + main() diff --git a/scripts/watermark_benchmark_report.py b/scripts/watermark_benchmark_report.py new file mode 100755 index 0000000..8586e13 --- /dev/null +++ b/scripts/watermark_benchmark_report.py @@ -0,0 +1,627 @@ +#!/usr/bin/env python3 +"""Aggregate one or more watermark benchmark JSONL result files. + +The report keeps detector observations, post-removal observations, fidelity, +and adapter timing separate. Repeated result files are independent runs. +Repeated pixels are deduplicated by artifact SHA-256 before case-level counts, +while raw observation counts remain visible. The first measured adapter call in +each input file is classified as cold; later calls to that adapter are warm. + +This is a development-only reader. It calls no detector, model, oracle, or +provider API and refuses to overwrite an existing report. + + uv run python scripts/watermark_benchmark_report.py results/*.jsonl \ + --output .local-eval/watermark-benchmark-report.md +""" + +from __future__ import annotations + +import argparse +import json +import logging +import math +import sys +from collections import Counter, defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal, cast + +if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + +SCRIPT_DIR = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR)) + +from watermark_benchmark import ( # noqa: E402 + DETECTOR_STATUSES, + SCHEMA_VERSION, + parse_strict_json, + require_nonempty_string, + require_sha256, + sha256_file, +) + +log = logging.getLogger(__name__) + +TimingPhase = Literal["cold", "warm"] +_TOP_LEVEL_FIELDS = { + "schema_version", + "case_id", + "pair_id", + "media_type", + "adapter", + "arm", + "state", + "artifact", + "reference", + "transform", + "run", + "detection", + "removal", + "fidelity", +} + + +@dataclass(frozen=True) +class ResultRecord: + """One minimally validated result row plus its run identity.""" + + source: Path + line_number: int + data: dict[str, Any] + + +def _object(value: object, *, field: str, location: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ValueError(f"{location}: {field} must be an object") + return cast("dict[str, Any]", value) + + +def _validate_result(raw: object, *, location: str) -> dict[str, Any]: + values = _object(raw, field="result", location=location) + if set(values) != _TOP_LEVEL_FIELDS: + raise ValueError(f"{location}: result fields do not match benchmark schema v{SCHEMA_VERSION}") + if values["schema_version"] != SCHEMA_VERSION: + raise ValueError(f"{location}: unsupported benchmark schema {values['schema_version']!r}") + for field in ("case_id", "pair_id", "adapter", "arm", "state"): + require_nonempty_string(values[field], field=field, location=location) + if values["media_type"] != "image": + raise ValueError(f"{location}: media_type must be image") + + artifact = _object(values["artifact"], field="artifact", location=location) + require_sha256(artifact.get("sha256"), field="artifact.sha256", location=location) + require_nonempty_string(artifact.get("source_revision"), field="artifact.source_revision", location=location) + reference = values["reference"] + if reference is not None: + require_sha256( + _object(reference, field="reference", location=location).get("sha256"), + field="reference.sha256", + location=location, + ) + + transform = _object(values["transform"], field="transform", location=location) + require_nonempty_string(transform.get("name"), field="transform.name", location=location) + require_nonempty_string(transform.get("revision"), field="transform.revision", location=location) + _object(transform.get("parameters"), field="transform.parameters", location=location) + + run = _object(values["run"], field="run", location=location) + _object(run.get("repository"), field="run.repository", location=location) + require_sha256(run.get("kernel_source_sha256"), field="run.kernel_source_sha256", location=location) + require_sha256(run.get("adapter_source_sha256"), field="run.adapter_source_sha256", location=location) + + detection = _object(values["detection"], field="detection", location=location) + status = detection.get("status") + if status not in DETECTOR_STATUSES: + raise ValueError(f"{location}: unknown detection status {status!r}") + elapsed = detection.get("adapter_elapsed_ms") + invalid_elapsed = ( + not isinstance(elapsed, (int, float)) or isinstance(elapsed, bool) or not math.isfinite(elapsed) or elapsed < 0 + ) + if elapsed is not None and invalid_elapsed: + raise ValueError(f"{location}: detection.adapter_elapsed_ms must be a finite non-negative number or null") + _object(values["removal"], field="removal", location=location) + _object(values["fidelity"], field="fidelity", location=location) + return values + + +def load_results(paths: Sequence[Path]) -> list[ResultRecord]: + """Read benchmark results in input and line order.""" + if not paths: + raise ValueError("at least one result file is required") + resolved = [path.resolve() for path in paths] + if len(set(resolved)) != len(resolved): + raise ValueError("duplicate result file") + + records: list[ResultRecord] = [] + for path in resolved: + with path.open(encoding="utf-8") as stream: + for line_number, line in enumerate(stream, 1): + if not line.strip(): + continue + location = f"{path}:{line_number}" + try: + raw = parse_strict_json(line) + except (json.JSONDecodeError, ValueError) as exc: + detail = exc.msg if isinstance(exc, json.JSONDecodeError) else str(exc) + raise ValueError(f"{location}: invalid JSON: {detail}") from exc + records.append( + ResultRecord( + source=path, + line_number=line_number, + data=_validate_result(raw, location=location), + ) + ) + if not records: + raise ValueError("result files contain no records") + return records + + +def _canonical(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False) + + +def _case_identity(record: ResultRecord) -> tuple[object, ...]: + row = record.data + artifact = cast("dict[str, Any]", row["artifact"]) + reference = cast("dict[str, Any] | None", row["reference"]) + detection = cast("dict[str, Any]", row["detection"]) + return ( + row["pair_id"], + row["media_type"], + row["adapter"], + row["arm"], + row["state"], + artifact["sha256"], + artifact["source_revision"], + reference["sha256"] if reference is not None else None, + _canonical(row["transform"]), + detection.get("expected"), + ) + + +def _nearest_rank(ordered: Sequence[float], quantile: float) -> float: + if not ordered: + raise ValueError("cannot calculate a percentile without measurements") + return ordered[max(0, math.ceil(quantile * len(ordered)) - 1)] + + +def _timing_stats(adapter: str, phase: TimingPhase, values: list[float]) -> dict[str, Any]: + ordered = sorted(values) + return { + "adapter": adapter, + "phase": phase, + "n": len(ordered), + "p50_ms": _nearest_rank(ordered, 0.50), + "p90_ms": _nearest_rank(ordered, 0.90), + "max_ms": ordered[-1], + } + + +def _detection_groups(records: Sequence[ResultRecord], *, include_transform: bool) -> list[dict[str, Any]]: + grouped: dict[tuple[str, ...], list[ResultRecord]] = defaultdict(list) + for record in records: + row = record.data + identity = (str(row["adapter"]), str(row["state"])) + if include_transform: + transform = cast("dict[str, Any]", row["transform"]) + identity += (str(transform["name"]),) + grouped[identity].append(record) + + summaries: list[dict[str, Any]] = [] + for identity, group in sorted(grouped.items()): + artifacts: dict[str, set[str]] = defaultdict(set) + for record in group: + row = record.data + artifact = cast("dict[str, Any]", row["artifact"]) + detection = cast("dict[str, Any]", row["detection"]) + artifacts[str(artifact["sha256"])].add(str(detection["status"])) + counts: dict[str, int] = dict.fromkeys(DETECTOR_STATUSES, 0) + unstable = 0 + for statuses in artifacts.values(): + if len(statuses) != 1: + unstable += 1 + continue + counts[next(iter(statuses))] += 1 + summary: dict[str, Any] = { + "adapter": identity[0], + "state": identity[1], + } + if include_transform: + summary["transform"] = identity[2] + summary.update( + { + "unique_artifacts": len(artifacts), + "observations": len(group), + **counts, + "unstable": unstable, + } + ) + summaries.append(summary) + return summaries + + +def _removal_groups(records: Sequence[ResultRecord]) -> list[dict[str, Any]]: + grouped: dict[str, list[ResultRecord]] = defaultdict(list) + for record in records: + if record.data["state"] == "removed": + grouped[str(record.data["adapter"])].append(record) + + summaries: list[dict[str, Any]] = [] + for adapter, group in sorted(grouped.items()): + artifacts: dict[str, set[str]] = defaultdict(set) + for record in group: + row = record.data + removal = cast("dict[str, Any]", row["removal"]) + artifact = cast("dict[str, Any]", row["artifact"]) + artifacts[str(artifact["sha256"])].add(str(removal.get("status"))) + statuses = ( + "signal_detected_after_removal", + "no_recognized_signal_after_removal", + "unmeasured", + ) + counts: dict[str, int] = dict.fromkeys(statuses, 0) + unstable = 0 + for observed in artifacts.values(): + if len(observed) != 1 or next(iter(observed)) not in counts: + unstable += 1 + else: + counts[next(iter(observed))] += 1 + summaries.append( + { + "adapter": adapter, + "unique_artifacts": len(artifacts), + "observations": len(group), + **counts, + "unstable": unstable, + } + ) + return summaries + + +def _fidelity_groups(records: Sequence[ResultRecord]) -> list[dict[str, Any]]: + grouped: dict[tuple[str, str], list[ResultRecord]] = defaultdict(list) + for record in records: + grouped[(str(record.data["adapter"]), str(record.data["state"]))].append(record) + + summaries: list[dict[str, Any]] = [] + for (adapter, state), group in sorted(grouped.items()): + pairs: dict[tuple[str, str | None], tuple[str, dict[str, Any]]] = {} + inconsistent_pairs: set[tuple[str, str | None]] = set() + for record in group: + row = record.data + artifact = cast("dict[str, Any]", row["artifact"]) + reference = cast("dict[str, Any] | None", row["reference"]) + key = (str(artifact["sha256"]), str(reference["sha256"]) if reference is not None else None) + fidelity = cast("dict[str, Any]", row["fidelity"]) + canonical = _canonical(fidelity) + previous = pairs.setdefault(key, (canonical, fidelity)) + if previous[0] != canonical: + inconsistent_pairs.add(key) + + counts = {"measured": 0, "incomparable": 0, "not_measured": 0, "inconsistent": 0} + identical = 0 + psnr_values: list[float] = [] + for key, (_, value) in pairs.items(): + if key in inconsistent_pairs: + counts["inconsistent"] += 1 + continue + status = str(value.get("status")) + if status not in ("measured", "incomparable", "not_measured"): + counts["inconsistent"] += 1 + continue + counts[status] += 1 + if status == "measured": + identical += int(value.get("identical") is True) + psnr = value.get("psnr_db") + if isinstance(psnr, (int, float)) and not isinstance(psnr, bool) and math.isfinite(psnr): + psnr_values.append(float(psnr)) + summaries.append( + { + "adapter": adapter, + "state": state, + "unique_pairs": len(pairs), + **counts, + "identical": identical, + "finite_psnr": len(psnr_values), + "p50_psnr_db": _nearest_rank(sorted(psnr_values), 0.50) if psnr_values else None, + } + ) + return summaries + + +def summarize(records: Sequence[ResultRecord]) -> dict[str, Any]: + """Build a deterministic, artifact-aware summary of benchmark records.""" + if not records: + raise ValueError("cannot summarize an empty result set") + + identities: dict[str, tuple[object, ...]] = {} + case_statuses: dict[tuple[str, str], set[str]] = defaultdict(set) + mismatches: set[tuple[str, str, str]] = set() + non_results: set[tuple[str, str, str, str | None]] = set() + timing_values: dict[tuple[str, TimingPhase], list[float]] = defaultdict(list) + timed_adapters: set[tuple[Path, str]] = set() + + for record in records: + row = record.data + removal = cast("dict[str, Any]", row["removal"]) + if removal.get("certifies_erasure") is not False: + raise ValueError(f"{record.source}:{record.line_number}: removal.certifies_erasure must be false") + case_id = str(row["case_id"]) + identity = _case_identity(record) + previous = identities.setdefault(case_id, identity) + if previous != identity: + raise ValueError(f"case identity drift for {case_id!r} at {record.source}:{record.line_number}") + + adapter = str(row["adapter"]) + detection = cast("dict[str, Any]", row["detection"]) + status = str(detection["status"]) + case_statuses[(case_id, adapter)].add(status) + if detection.get("matches_expected") is False: + mismatches.add((case_id, adapter, status)) + if status in ("error", "unavailable"): + error = detection.get("error") + non_results.add((case_id, adapter, status, str(error) if error is not None else None)) + + elapsed = detection.get("adapter_elapsed_ms") + if isinstance(elapsed, (int, float)) and not isinstance(elapsed, bool): + timing_key = (record.source, adapter) + phase: TimingPhase = "warm" if timing_key in timed_adapters else "cold" + timed_adapters.add(timing_key) + timing_values[(adapter, phase)].append(float(elapsed)) + + unstable_cases = [ + {"case_id": case_id, "adapter": adapter, "statuses": sorted(statuses)} + for (case_id, adapter), statuses in sorted(case_statuses.items()) + if len(statuses) > 1 + ] + timing = [_timing_stats(adapter, phase, values) for (adapter, phase), values in sorted(timing_values.items())] + source_counts = Counter(record.source for record in records) + inputs = [ + { + "path": str(source), + "sha256": sha256_file(source), + "observations": source_counts[source], + } + for source in sorted(source_counts) + ] + + return { + "runs": len(source_counts), + "observations": len(records), + "unique_cases": len(identities), + "inputs": inputs, + "detection": _detection_groups(records, include_transform=False), + "detection_by_transform": _detection_groups(records, include_transform=True), + "removal": _removal_groups(records), + "fidelity": _fidelity_groups(records), + "timing": timing, + "unstable_cases": unstable_cases, + "expected_mismatches": [ + {"case_id": case_id, "adapter": adapter, "status": status} + for case_id, adapter, status in sorted(mismatches) + ], + "non_results": [ + {"case_id": case_id, "adapter": adapter, "status": status, "error": error} + for case_id, adapter, status, error in sorted(non_results) + ], + } + + +def _cell(value: object) -> str: + if value is None: + return "-" + if isinstance(value, float): + return f"{value:.2f}" + return str(value).replace("|", "\\|").replace("\n", " ") + + +def _table(columns: Sequence[tuple[str, str]], rows: Sequence[Mapping[str, Any]]) -> list[str]: + header = "| " + " | ".join(label for _, label in columns) + " |" + separator = "| " + " | ".join("---" for _ in columns) + " |" + body = ["| " + " | ".join(_cell(row.get(key)) for key, _ in columns) + " |" for row in rows] + return [header, separator, *body] + + +def render_markdown(summary: Mapping[str, Any], *, title: str = "Watermark benchmark report") -> str: + """Render a standalone report without merging distinct evidence questions.""" + lines = [ + f"# {title}", + "", + f"Runs: {summary['runs']}. Observations: {summary['observations']}. Unique cases: {summary['unique_cases']}.", + "", + ( + "> `not_detected` means only that the selected adapter did not recognize its signal. " + "It is not evidence that a watermark is absent. Post-removal rows never certify erasure." + ), + "", + "## Detection by state", + "", + *_table( + ( + ("adapter", "Adapter"), + ("state", "State"), + ("unique_artifacts", "Unique artifacts"), + ("observations", "Observations"), + ("detected", "Detected"), + ("not_detected", "Not detected"), + ("unavailable", "Unavailable"), + ("error", "Error"), + ("unstable", "Unstable"), + ), + cast("list[dict[str, Any]]", summary["detection"]), + ), + "", + "## Detection by transform", + "", + *_table( + ( + ("adapter", "Adapter"), + ("state", "State"), + ("transform", "Transform"), + ("unique_artifacts", "Unique artifacts"), + ("detected", "Detected"), + ("not_detected", "Not detected"), + ("unavailable", "Unavailable"), + ("error", "Error"), + ("unstable", "Unstable"), + ), + cast("list[dict[str, Any]]", summary["detection_by_transform"]), + ), + "", + "## Post-removal observation", + "", + ] + removal = cast("list[dict[str, Any]]", summary["removal"]) + if removal: + lines.extend( + _table( + ( + ("adapter", "Adapter"), + ("unique_artifacts", "Unique artifacts"), + ("signal_detected_after_removal", "Signal remains"), + ("no_recognized_signal_after_removal", "No recognized signal"), + ("unmeasured", "Unmeasured"), + ("unstable", "Unstable"), + ), + removal, + ) + ) + else: + lines.append("No removed-state artifacts were present.") + + lines.extend(["", "## Adapter timing", ""]) + timing = cast("list[dict[str, Any]]", summary["timing"]) + if timing: + lines.extend( + _table( + ( + ("adapter", "Adapter"), + ("phase", "Phase"), + ("n", "n"), + ("p50_ms", "p50 ms"), + ("p90_ms", "p90 ms"), + ("max_ms", "max ms"), + ), + timing, + ) + ) + else: + lines.append("No adapter calls carried timing measurements.") + + lines.extend( + [ + "", + ( + "The first measured call for each adapter in each input file is cold; subsequent measured calls are " + "warm. Image decode, fidelity calculation, and process startup are outside the interval." + ), + "", + "## Fidelity", + "", + *_table( + ( + ("adapter", "Adapter"), + ("state", "State"), + ("unique_pairs", "Unique pairs"), + ("measured", "Measured"), + ("identical", "Identical"), + ("incomparable", "Incomparable"), + ("not_measured", "Not measured"), + ("inconsistent", "Inconsistent"), + ("finite_psnr", "Finite PSNR"), + ("p50_psnr_db", "p50 PSNR dB"), + ), + cast("list[dict[str, Any]]", summary["fidelity"]), + ), + "", + "## Stability and non-results", + "", + ] + ) + unstable = cast("list[dict[str, Any]]", summary["unstable_cases"]) + mismatches = cast("list[dict[str, Any]]", summary["expected_mismatches"]) + non_results = cast("list[dict[str, Any]]", summary["non_results"]) + lines.append( + f"Unstable cases: {len(unstable)}. Expected-result mismatches: {len(mismatches)}. " + f"Errors or unavailable observations: {len(non_results)}." + ) + if unstable: + lines.extend( + [ + "", + *_table( + (("case_id", "Unstable case"), ("adapter", "Adapter"), ("statuses", "Statuses")), + [row | {"statuses": ", ".join(row["statuses"])} for row in unstable], + ), + ] + ) + if mismatches: + lines.extend( + [ + "", + *_table( + (("case_id", "Expected-result mismatch"), ("adapter", "Adapter"), ("status", "Observed")), + mismatches, + ), + ] + ) + if non_results: + lines.extend( + [ + "", + *_table( + ( + ("case_id", "Non-result case"), + ("adapter", "Adapter"), + ("status", "Status"), + ("error", "Error"), + ), + non_results, + ), + ] + ) + + lines.extend( + [ + "", + "## Inputs", + "", + *_table( + (("path", "Result file"), ("sha256", "SHA-256"), ("observations", "Observations")), + cast("list[dict[str, Any]]", summary["inputs"]), + ), + "", + ] + ) + return "\n".join(lines) + + +def write_report(path: Path, report: str) -> None: + """Write a complete report once, refusing an existing destination.""" + path.parent.mkdir(parents=True, exist_ok=True) + try: + with path.open("x", encoding="utf-8") as stream: + stream.write(report) + except FileExistsError as exc: + raise FileExistsError(f"refusing to overwrite existing report: {path}") from exc + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("results", type=Path, nargs="+", help="one or more benchmark result JSONL files") + parser.add_argument("--output", type=Path, required=True, help="new Markdown report path; never overwritten") + parser.add_argument("--title", default="Watermark benchmark report") + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + try: + records = load_results(args.results) + summary = summarize(records) + write_report(args.output, render_markdown(summary, title=args.title)) + except (FileNotFoundError, OSError, ValueError) as exc: + parser.error(str(exc)) + log.info("Wrote benchmark report for %s observations to %s", summary["observations"], args.output) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_script_help.py b/tests/test_script_help.py index a8e9e69..a0989ed 100644 --- a/tests/test_script_help.py +++ b/tests/test_script_help.py @@ -20,6 +20,8 @@ ROOT = Path(__file__).resolve().parents[1] "visible_sheets.py", "registered_mark_calibrate.py", "watermark_benchmark.py", + "watermark_benchmark_cohort.py", + "watermark_benchmark_report.py", ], ) def test_script_help_exits_cleanly(script: str) -> None: diff --git a/tests/test_watermark_benchmark.py b/tests/test_watermark_benchmark.py index 8fd857c..692066e 100644 --- a/tests/test_watermark_benchmark.py +++ b/tests/test_watermark_benchmark.py @@ -193,11 +193,13 @@ def test_evaluation_keeps_detection_removal_and_fidelity_separate(tmp_path: Path ) case = MODULE.load_manifest(manifest)[0] adapter = _FakeAdapter(status="not_detected", label=None) + times = iter((100, 500)) record = MODULE.evaluate_case( case, adapters={"fake": adapter}, repository={"commit": "abc123", "dirty": False}, + clock_ns=lambda: next(times), ) assert adapter.calls == [removed.resolve()] @@ -208,6 +210,7 @@ def test_evaluation_keeps_detection_removal_and_fidelity_separate(tmp_path: Path "expected": "not_detected", "matches_expected": True, "positive_evidence": False, + "adapter_elapsed_ms": 0.0004, } assert record["removal"] == { "attempted": True, @@ -272,6 +275,7 @@ def test_unavailable_adapter_is_not_reported_as_negative(tmp_path: Path) -> None assert record["detection"]["status"] == "unavailable" assert record["detection"]["matches_expected"] is None + assert record["detection"]["adapter_elapsed_ms"] is None assert adapter.calls == [] @@ -290,6 +294,7 @@ def test_undecodable_artifact_is_error_not_negative(tmp_path: Path) -> None: assert record["detection"]["status"] == "error" assert record["detection"]["matches_expected"] is None assert record["detection"]["error"] == "artifact could not be decoded" + assert record["detection"]["adapter_elapsed_ms"] is None assert adapter.calls == [] diff --git a/tests/test_watermark_benchmark_cohort.py b/tests/test_watermark_benchmark_cohort.py new file mode 100644 index 0000000..af8c23f --- /dev/null +++ b/tests/test_watermark_benchmark_cohort.py @@ -0,0 +1,75 @@ +"""Contracts for the synthetic watermark benchmark cohort builder.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "watermark_benchmark_cohort.py" +SPEC = importlib.util.spec_from_file_location("watermark_benchmark_cohort", SCRIPT) +assert SPEC is not None +assert SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = MODULE +SPEC.loader.exec_module(MODULE) + + +def test_dwt_cohort_has_matched_positive_attack_and_hard_negative_arms(tmp_path: Path) -> None: + output = tmp_path / "cohort" + + manifest = MODULE.build_cohort( + output, + adapters=("dwt-dct",), + carriers=(MODULE.CarrierSpec("texture", 7),), + dwt_schemes=("sdxl",), + attacks=("jpeg-q90",), + ) + rows = MODULE.load_manifest(manifest) + + assert len(rows) == 4 + assert {(row.arm, row.state, row.expected) for row in rows} == { + ("matched_negative", "clean", "not_detected"), + ("positive", "marked", "detected"), + ("positive", "attacked", "unresolved"), + ("hard_negative", "clean", "not_detected"), + } + assert all(row.source_revision.startswith("watermark-benchmark-cohort-v1@sha256:") for row in rows) + assert all(row.path.is_relative_to(output) for row in rows) + attacked = next(row for row in rows if row.state == "attacked") + assert attacked.transform.parameters["carrier"] == "texture" + assert attacked.transform.parameters["scheme"] == "sdxl" + + +def test_cohort_refuses_an_existing_output_directory(tmp_path: Path) -> None: + output = tmp_path / "cohort" + output.mkdir() + + with pytest.raises(FileExistsError, match="output directory already exists"): + MODULE.build_cohort(output, adapters=("dwt-dct",)) + + +def test_trustmark_cohort_includes_removed_observation(tmp_path: Path) -> None: + if not MODULE.trustmark_available(): + pytest.skip("trustmark not installed") + output = tmp_path / "cohort" + + manifest = MODULE.build_cohort( + output, + adapters=("trustmark",), + carriers=(MODULE.CarrierSpec("gradient", 11),), + attacks=(), + ) + rows = MODULE.load_manifest(manifest) + + assert {(row.arm, row.state, row.expected) for row in rows} == { + ("matched_negative", "clean", "not_detected"), + ("positive", "marked", "detected"), + ("positive", "removed", "unresolved"), + ("hard_negative", "clean", "not_detected"), + } + removed = next(row for row in rows if row.state == "removed") + assert removed.reference_path is not None + assert removed.transform.name == "trustmark-remove" diff --git a/tests/test_watermark_benchmark_report.py b/tests/test_watermark_benchmark_report.py new file mode 100644 index 0000000..2acd439 --- /dev/null +++ b/tests/test_watermark_benchmark_report.py @@ -0,0 +1,236 @@ +"""Contracts for the watermark benchmark report builder.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import numpy as np +import pytest +from PIL import Image + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "watermark_benchmark_report.py" +SPEC = importlib.util.spec_from_file_location("watermark_benchmark_report", SCRIPT) +assert SPEC is not None +assert SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = MODULE +SPEC.loader.exec_module(MODULE) +BENCHMARK = sys.modules["watermark_benchmark"] + + +class _FakeAdapter: + name = "dwt-dct" + source_file = SCRIPT + + def __init__(self, status: str) -> None: + self.status = status + + @property + def available(self) -> bool: + return self.status != "unavailable" + + def detect(self, _path: Path, _image: Any) -> Any: + if self.status == "error": + raise RuntimeError("failed") + return BENCHMARK.DetectorOutcome( + status=self.status, + label="mark" if self.status == "detected" else None, + ) + + +def _image(path: Path, value: int) -> Path: + Image.fromarray(np.full((16, 20, 3), value, dtype=np.uint8), "RGB").save(path) + return path + + +def _row( + artifact: Path, + case_id: str, + *, + state: str = "marked", + status: str = "detected", + elapsed_ms: float | None = 2.0, + transform: str = "embed", + expected: str = "detected", +) -> dict[str, Any]: + digest = BENCHMARK.sha256_file(artifact) + case = BENCHMARK.BenchmarkCase( + case_id=case_id, + pair_id=case_id.split("--")[0], + media_type="image", + adapter="dwt-dct", + arm="positive" if state != "clean" else "matched_negative", + state=state, + path=artifact, + sha256=digest, + reference_path=artifact, + reference_sha256=digest, + source_revision="cohort@abc", + transform=BENCHMARK.Transform(name=transform, revision="recipe-v1", parameters={}), + seed=7, + expected=expected, + ) + times = iter((0, round((elapsed_ms or 0.0) * 1e6))) + + def clock_ns() -> int: + return next(times) + + return BENCHMARK.evaluate_case( + case, + adapters={"dwt-dct": _FakeAdapter(status)}, + repository={"commit": "abc123", "dirty": False}, + clock_ns=clock_ns, + ) + + +def _write(path: Path, rows: list[dict[str, Any]]) -> Path: + path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8") + return path + + +def _group(summary: dict[str, Any], section: str, **identity: str) -> dict[str, Any]: + return next(row for row in summary[section] if all(row[key] == value for key, value in identity.items())) + + +def test_summary_deduplicates_artifacts_and_separates_cold_from_warm(tmp_path: Path) -> None: + clean_path = _image(tmp_path / "clean.png", 10) + marked_path = _image(tmp_path / "marked.png", 20) + removed_path = _image(tmp_path / "removed.png", 30) + clean = _row( + clean_path, + "clean-sdxl", + state="clean", + status="not_detected", + elapsed_ms=10.0, + transform="synthetic-carrier", + expected="not_detected", + ) + repeated_clean = _row( + clean_path, + "clean-flux", + state="clean", + status="not_detected", + elapsed_ms=1.0, + transform="synthetic-carrier", + expected="not_detected", + ) + marked = _row(marked_path, "marked", elapsed_ms=2.0) + removed = _row( + removed_path, + "removed", + state="removed", + status="not_detected", + elapsed_ms=3.0, + transform="remove", + expected="unresolved", + ) + first = _write(tmp_path / "run-1.jsonl", [clean, repeated_clean, marked, removed]) + second_rows = json.loads(json.dumps([clean, repeated_clean, marked, removed])) + for row, elapsed in zip(second_rows, (20.0, 1.5, 2.5, 3.5), strict=True): + row["detection"]["adapter_elapsed_ms"] = elapsed + second = _write(tmp_path / "run-2.jsonl", second_rows) + + summary = MODULE.summarize(MODULE.load_results([first, second])) + + clean_group = _group(summary, "detection", adapter="dwt-dct", state="clean") + assert clean_group == { + "adapter": "dwt-dct", + "state": "clean", + "unique_artifacts": 1, + "observations": 4, + "detected": 0, + "not_detected": 1, + "unavailable": 0, + "error": 0, + "unstable": 0, + } + cold = _group(summary, "timing", adapter="dwt-dct", phase="cold") + assert cold == {"adapter": "dwt-dct", "phase": "cold", "n": 2, "p50_ms": 10.0, "p90_ms": 20.0, "max_ms": 20.0} + warm = _group(summary, "timing", adapter="dwt-dct", phase="warm") + assert warm == {"adapter": "dwt-dct", "phase": "warm", "n": 6, "p50_ms": 2.0, "p90_ms": 3.5, "max_ms": 3.5} + removal = _group(summary, "removal", adapter="dwt-dct") + assert removal["unique_artifacts"] == 1 + assert removal["no_recognized_signal_after_removal"] == 1 + assert summary["runs"] == 2 + assert summary["observations"] == 8 + + +def test_summary_reports_unstable_errors_unavailable_and_mismatches(tmp_path: Path) -> None: + changing_path = _image(tmp_path / "changing.png", 10) + error_path = _image(tmp_path / "error.png", 20) + unavailable_path = _image(tmp_path / "unavailable.png", 30) + first_case = _row(changing_path, "changing", status="detected") + second_case = _row( + changing_path, + "changing", + status="not_detected", + ) + error = _row(error_path, "error", status="error", elapsed_ms=4.0) + unavailable = _row(unavailable_path, "unavailable", status="unavailable", elapsed_ms=None) + first = _write(tmp_path / "run-1.jsonl", [first_case, error, unavailable]) + second = _write(tmp_path / "run-2.jsonl", [second_case]) + + summary = MODULE.summarize(MODULE.load_results([first, second])) + + marked = _group(summary, "detection", adapter="dwt-dct", state="marked") + assert marked["unstable"] == 1 + assert marked["error"] == 1 + assert marked["unavailable"] == 1 + assert summary["unstable_cases"] == [ + {"case_id": "changing", "adapter": "dwt-dct", "statuses": ["detected", "not_detected"]} + ] + assert summary["expected_mismatches"] == [{"case_id": "changing", "adapter": "dwt-dct", "status": "not_detected"}] + + +def test_summary_rejects_case_identity_drift_between_runs(tmp_path: Path) -> None: + first_artifact = _image(tmp_path / "first.png", 10) + second_artifact = _image(tmp_path / "second.png", 20) + first = _write(tmp_path / "run-1.jsonl", [_row(first_artifact, "same-case")]) + second = _write(tmp_path / "run-2.jsonl", [_row(second_artifact, "same-case")]) + + with pytest.raises(ValueError, match="case identity drift"): + MODULE.summarize(MODULE.load_results([first, second])) + + +def test_loader_names_invalid_json_location(tmp_path: Path) -> None: + path = tmp_path / "broken.jsonl" + artifact = _image(tmp_path / "valid.png", 10) + path.write_text(json.dumps(_row(artifact, "valid")) + "\n{broken}\n", encoding="utf-8") + + with pytest.raises(ValueError, match=r"broken\.jsonl:2: invalid JSON"): + MODULE.load_results([path]) + + +def test_summary_rejects_a_false_erasure_certificate(tmp_path: Path) -> None: + artifact = _image(tmp_path / "removed.png", 10) + row = _row( + artifact, + "removed", + state="removed", + status="not_detected", + expected="unresolved", + ) + row["removal"]["certifies_erasure"] = True + path = _write(tmp_path / "run.jsonl", [row]) + + with pytest.raises(ValueError, match="certifies_erasure must be false"): + MODULE.summarize(MODULE.load_results([path])) + + +def test_report_refuses_overwrite_and_preserves_interpretation_caveat(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 10) + source = _write(tmp_path / "run.jsonl", [_row(artifact, "marked")]) + summary = MODULE.summarize(MODULE.load_results([source])) + report = MODULE.render_markdown(summary, title="Test report") + output = tmp_path / "report.md" + + MODULE.write_report(output, report) + + assert output.read_text(encoding="utf-8").startswith("# Test report\n") + assert "not evidence that a watermark is absent" in report + with pytest.raises(FileExistsError, match="refusing to overwrite"): + MODULE.write_report(output, report)