Refactor watermark detection and provenance handling

src/remove_ai_watermarks/api.py

@@ -16,6 +16,7 @@ Imports stay lazy (inside the functions), so ``import remove_ai_watermarks`` is
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -25,6 +26,16 @@ if TYPE_CHECKING:
     from remove_ai_watermarks.watermark_registry import Backend, Sensitivity
 
 
+@dataclass(frozen=True)
+class _VisibleInput:
+    """Normalized visible-removal input with its file-only context."""
+
+    bgr: NDArray[Any]
+    alpha: NDArray[Any] | None = None
+    path: Path | None = None
+    provenance: frozenset[str] = frozenset()
+
+
 def visible_provenance(source: str | Path) -> frozenset[str]:
     """Vendor keys that the file's local metadata confirms, the evidence that drives
     the ``auto`` sensitivity (relaxing a corroborated mark's detection trust gate).
@@ -36,19 +47,70 @@ def visible_provenance(source: str | Path) -> frozenset[str]:
     """
     import contextlib
 
-    keys: set[str] = set()
+    path = Path(source)
     with contextlib.suppress(Exception):
-        from remove_ai_watermarks import identify, metadata
+        from remove_ai_watermarks import identify
 
-        rep = identify.identify(Path(source), check_visible=False, check_invisible=False)
+        rep = identify.identify(path, check_visible=False, check_invisible=False)
+        signal_names = {signal.name for signal in rep.signals}
+        keys: set[str] = set()
         platform = (rep.platform or "").lower()
         if "google" in platform or "gemini" in platform:
             keys.add("gemini")
-        if metadata.aigc_label(Path(source)):
+        if "aigc" in signal_names:
             keys |= {"doubao", "jimeng"}
-        if metadata.samsung_genai(Path(source)):
+        if "samsung_genai" in signal_names:
             keys.add("samsung")
-    return frozenset(keys)
+        return frozenset(keys)
+    return frozenset()
+
+
+def _load_visible_input(source: str | Path | NDArray[Any]) -> _VisibleInput:
+    """Normalize a path/array source without making the public operation stateful."""
+    if not isinstance(source, (str, Path)):
+        return _VisibleInput(source)
+
+    from remove_ai_watermarks import image_io
+
+    path = Path(source)
+    bgr, alpha = image_io.read_bgr_and_alpha(path)
+    if bgr is None:
+        raise ValueError(f"Could not read image: {source}")
+    return _VisibleInput(bgr=bgr, alpha=alpha, path=path, provenance=visible_provenance(path))
+
+
+def _write_visible_result(
+    loaded: _VisibleInput,
+    result: NDArray[Any],
+    removed: list[str],
+    output: str | Path,
+    *,
+    strip_metadata: bool,
+    write_noop: bool,
+) -> None:
+    """Write one visible-removal result while preserving a true no-op losslessly."""
+    if not removed and not write_noop:
+        return
+
+    from remove_ai_watermarks import image_io
+
+    out_path = Path(output)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    source_path = loaded.path
+    if not removed and source_path is not None and source_path.suffix.lower() == out_path.suffix.lower():
+        # Copy the ORIGINAL bytes instead of lossily re-encoding a no-op. An in-place
+        # call needs no copy and would otherwise raise shutil.SameFileError.
+        if source_path.resolve() != out_path.resolve():
+            import shutil
+
+            shutil.copyfile(source_path, out_path)
+    else:
+        image_io.write_bgr_with_alpha(out_path, result, loaded.alpha)
+
+    if strip_metadata:
+        from remove_ai_watermarks import metadata
+
+        metadata.remove_ai_metadata(out_path, out_path)
 
 
 def remove_visible(
@@ -88,40 +150,22 @@ def remove_visible(
     output path untouched, so a caller that treats "no mark" as "produce nothing" (the CLI
     ``visible`` no-mark contract) does not clobber a pre-existing file at that path.
     """
-    from remove_ai_watermarks import image_io, watermark_registry
-
-    alpha: NDArray[Any] | None = None
-    provenance: frozenset[str] = frozenset()
-    if isinstance(source, (str, Path)):
-        path = Path(source)
-        bgr, alpha = image_io.read_bgr_and_alpha(path)
-        if bgr is None:
-            raise ValueError(f"Could not read image: {source}")
-        provenance = visible_provenance(path)
-    else:
-        bgr = source
+    from remove_ai_watermarks import watermark_registry
 
+    loaded = _load_visible_input(source)
     result, removed = watermark_registry.remove_auto_marks(
-        bgr, sensitivity=sensitivity, provenance=provenance, backend=backend
+        loaded.bgr,
+        sensitivity=sensitivity,
+        provenance=loaded.provenance,
+        backend=backend,
     )
-    if output is not None and (removed or write_noop):
-        out_path = Path(output)
-        out_path.parent.mkdir(parents=True, exist_ok=True)
-        same_format = isinstance(source, (str, Path)) and Path(source).suffix.lower() == out_path.suffix.lower()
-        if not removed and same_format:
-            # Nothing was removed: copy the ORIGINAL bytes verbatim instead of a lossy
-            # re-encode of its decode, so the pixels stay bit-identical (the metadata
-            # strip below is lossless, so it does not disturb them either). Skip the copy
-            # for an in-place call (output == source): the bytes are already there, and
-            # shutil.copyfile would raise SameFileError.
-            if Path(source).resolve() != out_path.resolve():  # type: ignore[arg-type]
-                import shutil
-
-                shutil.copyfile(source, out_path)  # type: ignore[arg-type]
-        else:
-            image_io.write_bgr_with_alpha(out_path, result, alpha)
-        if strip_metadata:
-            from remove_ai_watermarks import metadata
-
-            metadata.remove_ai_metadata(out_path, out_path)
+    if output is not None:
+        _write_visible_result(
+            loaded,
+            result,
+            removed,
+            output,
+            strip_metadata=strip_metadata,
+            write_noop=write_noop,
+        )
     return result, removed

src/remove_ai_watermarks/cli.py

@@ -12,6 +12,7 @@ import contextlib
 import json
 import logging
 import time
+from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, NoReturn
 
@@ -311,8 +312,9 @@ _visible_sensitivity_option = click.option(
     help="How hard to trust a borderline mark. auto: relax a mark only when metadata "
     "or a same-product sibling mark corroborates it (safe; clean images untouched). "
     "strict: high-precision visual gate only, never relaxed. assume-ai: treat the "
-    "image as AI and relax every mark (best recall on metadata-stripped screenshots, "
-    "at the cost of a small fill on some clean corners).",
+    "image as AI and relax every mark, keeping a confidence floor where the vendor is "
+    "unconfirmed (best recall on metadata-stripped screenshots; a clean image is still "
+    "left untouched).",
 )
 
 
@@ -525,6 +527,113 @@ def main(ctx: click.Context, verbose: bool) -> None:
 
 
 # ── Visible (Gemini) watermark removal ──
+def _run_visible_auto(
+    source: Path,
+    output: Path,
+    *,
+    backend: watermark_registry.Backend,
+    sensitivity: watermark_registry.Sensitivity,
+    strip_metadata: bool,
+) -> None:
+    """Run the registry-wide visible pass and render its CLI result."""
+    from remove_ai_watermarks import api
+
+    t0 = time.monotonic()
+    try:
+        with console.status("Detecting & removing visible marks..."):
+            result, removed = api.remove_visible(
+                str(source),
+                str(output),
+                sensitivity=sensitivity,
+                backend=backend,
+                strip_metadata=strip_metadata,
+                write_noop=False,
+            )
+    except RuntimeError as e:  # selected migan/lama backend whose extra is absent
+        console.print(f"  Error: {e}")
+        raise SystemExit(1) from e
+    except (ValueError, OSError) as e:  # unreadable / truncated / non-image input
+        console.print(f"  Error: cannot read image {source.name}: {e}")
+        raise SystemExit(1) from e
+
+    elapsed = time.monotonic() - t0
+    h, w = result.shape[:2]
+    console.print(f"  Input:  {source.name}  ({w}x{h})")
+    if not removed:
+        # write_noop=False means nothing was written, so a pre-existing output is intact.
+        console.print("  No known visible mark detected (gemini / doubao / jimeng / jimeng-pill / samsung).")
+        _no_visible_mark_exit(source, sensitivity=sensitivity)
+    console.print(f"  Removed: {', '.join(removed)}")
+    size_kb = output.stat().st_size / 1024
+    console.print(f"  Saved: {output}  ({size_kb:.0f} KB, {elapsed:.2f}s)")
+
+
+def _run_visible_explicit(
+    ctx: click.Context,
+    source: Path,
+    output: Path,
+    *,
+    detect: bool,
+    mark: str,
+    backend: watermark_registry.Backend,
+    sensitivity: watermark_registry.Sensitivity,
+    resolved_backend: str,
+    strip_metadata: bool,
+) -> None:
+    """Run one explicitly selected visible-mark detector/remover."""
+    image, alpha = image_io.read_bgr_and_alpha(source)
+    if image is None:
+        console.print(f"Error: Failed to read image: {source}")
+        raise SystemExit(1)
+    h, w = image.shape[:2]
+    console.print(f"  Input:  {source.name}  ({w}x{h})")
+
+    provenance = _visible_provenance(source)
+    target = "gemini" if mark == "auto" else mark  # --no-detect auto: gemini fallback
+    chosen = watermark_registry.get_mark(target)
+    # A single explicit mark has no sibling corroboration. Keep its trust resolution
+    # aligned with the registry arbiter, including the assumption-only floor.
+    trust = watermark_registry.resolve_trust(
+        chosen.key,
+        sensitivity=sensitivity,
+        provenance=provenance,
+        strict_keys=set(),
+    )
+    relax = trust != "strict"
+    detection = chosen.detect(image, provenance=relax)
+    if trust == "assumed" and not watermark_registry.assumed_floor_ok(chosen.key, detection.confidence):
+        relax = False
+        detection = chosen.detect(image, provenance=False)
+    if detect and not detection.detected:
+        console.print(f"  {chosen.label} not detected  (conf {detection.confidence:.2f}). Use --no-detect to force.")
+        _no_visible_mark_exit(source, sensitivity=sensitivity)
+    if detection.detected:
+        console.print(f"  {chosen.label} detected  ({chosen.location}, conf {detection.confidence:.2f})")
+
+    t0 = time.monotonic()
+    try:
+        with console.status(f"Removing {chosen.label}... ({resolved_backend})"):
+            result, _ = chosen.remove(image, backend=backend, provenance=relax, force=not detect)
+    except RuntimeError as e:  # selected migan/lama backend whose extra is absent
+        console.print(f"  Error: {e}")
+        raise SystemExit(1) from e
+    elapsed = time.monotonic() - t0
+
+    output.parent.mkdir(parents=True, exist_ok=True)
+    image_io.write_bgr_with_alpha(output, result, alpha)
+    if strip_metadata:
+        try:
+            from remove_ai_watermarks.metadata import remove_ai_metadata
+
+            remove_ai_metadata(output, output)
+        except Exception as e:
+            if ctx.obj.get("verbose"):
+                console.print(f"  Warning: Failed to strip metadata: {e}")
+
+    size_kb = output.stat().st_size / 1024
+    console.print(f"  Saved: {output}  ({size_kb:.0f} KB, {elapsed:.2f}s)")
+
+
 @main.command("visible")
 @click.argument("source", type=click.Path(exists=True, path_type=Path))
 @click.option(
@@ -560,18 +669,16 @@ def cmd_visible(
     MI-GAN > cv2). ``--mark auto`` removes every detected mark in one
     pass. For arbitrary logos/objects, use ``erase``.
     """
-    from remove_ai_watermarks import watermark_registry as registry
-
     _banner()
     source = _validate_image(source)
 
     if output is None:
         output = source.with_stem(source.stem + "_clean")
 
-    bk: registry.Backend = backend  # type: ignore[assignment]
+    bk: watermark_registry.Backend = backend  # type: ignore[assignment]
     sens = _parse_sensitivity(sensitivity)
-    resolved_backend = registry.resolve_backend(bk)
-    if resolved_backend == "cv2" and not registry.inpaint_model_available():
+    resolved_backend = watermark_registry.resolve_backend(bk)
+    if resolved_backend == "cv2" and not watermark_registry.inpaint_model_available():
         console.print("  Note: using cv2 fill (install the 'migan' extra for a lightweight ONNX model).")
 
     # ``auto`` removes EVERY detected in_auto mark in one pass (a Jimeng-basic image
@@ -579,82 +686,20 @@ def cmd_visible(
     # read -> provenance -> localize/fill -> write -> metadata-strip to the library
     # entry point, so the CLI and the library go through ONE path (no drift).
     if mark == "auto" and detect:
-        from remove_ai_watermarks import api
-
-        t0 = time.monotonic()
-        try:
-            with console.status("Detecting & removing visible marks..."):
-                result, removed = api.remove_visible(
-                    str(source),
-                    str(output),
-                    sensitivity=sens,
-                    backend=bk,
-                    strip_metadata=strip_metadata,
-                    write_noop=False,
-                )
-        except RuntimeError as e:  # e.g. a selected migan/lama backend whose extra is absent
-            console.print(f"  Error: {e}")
-            raise SystemExit(1) from e
-        except (ValueError, OSError) as e:  # unreadable / truncated / non-image input
-            console.print(f"  Error: cannot read image {source.name}: {e}")
-            raise SystemExit(1) from e
-        elapsed = time.monotonic() - t0
-        h, w = result.shape[:2]
-        console.print(f"  Input:  {source.name}  ({w}x{h})")
-        if not removed:
-            # write_noop=False means nothing was written, so a pre-existing file at the
-            # output path is left intact (the no-mark contract writes nothing).
-            console.print("  No known visible mark detected (gemini / doubao / jimeng / jimeng-pill / samsung).")
-            _no_visible_mark_exit(source, sensitivity=sens)
-        console.print(f"  Removed: {', '.join(removed)}")
-        size_kb = output.stat().st_size / 1024
-        console.print(f"  Saved: {output}  ({size_kb:.0f} KB, {elapsed:.2f}s)")
+        _run_visible_auto(source, output, backend=bk, sensitivity=sens, strip_metadata=strip_metadata)
         return
 
-    # Explicit single mark (or --no-detect): needs the decoded array + the per-mark gate,
-    # so it keeps its own read/remove/write (still through the shared io + registry).
-    image, alpha = image_io.read_bgr_and_alpha(source)
-    if image is None:
-        console.print(f"Error: Failed to read image: {source}")
-        raise SystemExit(1)
-    h, w = image.shape[:2]
-    console.print(f"  Input:  {source.name}  ({w}x{h})")
-    provenance = _visible_provenance(source)
-    target = "gemini" if mark == "auto" else mark  # --no-detect auto: gemini fallback
-    chosen = registry.get_mark(target)
-    # A single explicit mark has no cross-mark pass (no sibling corroboration), so use the
-    # canonical arbiter policy with an empty strict-sibling set instead of re-deriving it
-    # inline (keeps this in lockstep with `decide`).
-    prov = registry.resolve_relax(chosen.key, sensitivity=sens, provenance=provenance, strict_keys=set())
-    det = chosen.detect(image, provenance=prov)
-    if detect and not det.detected:
-        console.print(f"  {chosen.label} not detected  (conf {det.confidence:.2f}). Use --no-detect to force.")
-        _no_visible_mark_exit(source, sensitivity=sens)
-    if det.detected:
-        console.print(f"  {chosen.label} detected  ({chosen.location}, conf {det.confidence:.2f})")
-    t0 = time.monotonic()
-    try:
-        with console.status(f"Removing {chosen.label}... ({resolved_backend})"):
-            result, _ = chosen.remove(image, backend=bk, provenance=prov, force=not detect)
-    except RuntimeError as e:  # e.g. a selected migan/lama backend whose extra is absent
-        console.print(f"  Error: {e}")
-        raise SystemExit(1) from e
-    elapsed = time.monotonic() - t0
-
-    # Save (rejoins the original alpha plane unchanged) + strip metadata.
-    output.parent.mkdir(parents=True, exist_ok=True)
-    image_io.write_bgr_with_alpha(output, result, alpha)
-    if strip_metadata:
-        try:
-            from remove_ai_watermarks.metadata import remove_ai_metadata
-
-            remove_ai_metadata(output, output)
-        except Exception as e:
-            if ctx.obj.get("verbose"):
-                console.print(f"  Warning: Failed to strip metadata: {e}")
-
-    size_kb = output.stat().st_size / 1024
-    console.print(f"  Saved: {output}  ({size_kb:.0f} KB, {elapsed:.2f}s)")
+    _run_visible_explicit(
+        ctx,
+        source,
+        output,
+        detect=detect,
+        mark=mark,
+        backend=bk,
+        sensitivity=sens,
+        resolved_backend=resolved_backend,
+        strip_metadata=strip_metadata,
+    )
 
 
 # ── Universal region eraser ──
@@ -1261,32 +1306,106 @@ def _passthrough_copy(img_path: Path, out_path: Path) -> None:
         image_io.write_bgr_with_alpha(out_path, src_bgr, src_alpha)
 
 
+@dataclass(frozen=True)
+class _BatchOptions:
+    """Validated processing options shared by every image in one batch.
+
+    Click necessarily exposes these as individual command parameters, but the
+    processing core should receive one coherent value instead of a 21-argument
+    call. Keeping the object immutable also makes it safe to reuse while the
+    batch caches model instances in ``ctx.obj``.
+    """
+
+    strength: float | None
+    steps: int
+    pipeline: str
+    device: str
+    seed: int | None
+    hf_token: str | None
+    humanize: float
+    backend: str = "auto"
+    sensitivity: str = "auto"
+    unsharp: float = 0.0
+    max_resolution: int = 0
+    min_resolution: int = 1024
+    controlnet_scale: float = 1.0
+    upscaler: str = "lanczos"
+    model: str | None = None
+    guidance_scale: float | None = None
+    adaptive_polish: bool = False
+    tile: bool = False
+    tile_size: int = 1024
+    tile_overlap: int = 128
+    force: bool = False
+
+
+def _run_batch_invisible(
+    ctx: click.Context,
+    img_path: Path,
+    out_path: Path,
+    mode: str,
+    options: _BatchOptions,
+) -> bool:
+    """Run or safely skip the invisible pass for one batch image.
+
+    Returns ``True`` only when a detectable target could not be processed because
+    the GPU dependencies are missing. The availability probe is intentionally
+    evaluated once so branching cannot observe inconsistent optional-dependency
+    state.
+    """
+    from remove_ai_watermarks.invisible_engine import is_available as invisible_available
+
+    skip_no_signal = _should_skip_invisible_scrub(options.force, img_path)
+    available = invisible_available()
+    if available and not skip_no_signal:
+        from remove_ai_watermarks.invisible_engine import InvisibleEngine
+
+        # Cache the engine in ctx.obj so the batch builds it once (pipeline is a
+        # single CLI value, constant across the run).
+        engines = ctx.obj.setdefault("_inv_engines", {})
+        if options.pipeline not in engines:
+            engines[options.pipeline] = InvisibleEngine(
+                model_id=options.model,
+                device=None if options.device == "auto" else options.device,
+                pipeline=options.pipeline,
+                hf_token=options.hf_token,
+                controlnet_conditioning_scale=options.controlnet_scale,
+            )
+        engines[options.pipeline].remove_watermark(
+            img_path if mode == "invisible" else out_path,
+            out_path,
+            strength=options.strength,
+            num_inference_steps=options.steps,
+            guidance_scale=options.guidance_scale,
+            seed=options.seed,
+            humanize=options.humanize,
+            unsharp=options.unsharp,
+            adaptive_polish=options.adaptive_polish,
+            max_resolution=options.max_resolution,
+            min_resolution=options.min_resolution,
+            upscaler=options.upscaler,
+            tile=options.tile,
+            tile_size=options.tile_size,
+            tile_overlap=options.tile_overlap,
+            # Detect the vendor from the pristine original (`img_path`), not the
+            # visible-processed `out_path` whose C2PA is already gone.
+            vendor=vendor_for_strength(img_path),
+        )
+        return False
+
+    # Invisible-only mode has no preceding visible pass to create ``out_path``.
+    # Preserve a complete output directory while deliberately leaving pixels intact.
+    if mode == "invisible" and not out_path.exists():
+        _passthrough_copy(img_path, out_path)
+    return not available and not skip_no_signal
+
+
 def _process_batch_image(
     ctx: click.Context,
     img_path: Path,
     out_path: Path,
     mode: str,
-    strength: float | None,
-    steps: int,
-    pipeline: str,
-    device: str,
-    seed: int | None,
-    hf_token: str | None,
-    humanize: float,
-    backend: str = "auto",
-    sensitivity: str = "auto",
-    unsharp: float = 0.0,
-    max_resolution: int = 0,
-    min_resolution: int = 1024,
-    controlnet_scale: float = 1.0,
-    upscaler: str = "lanczos",
-    model: str | None = None,
-    guidance_scale: float | None = None,
-    adaptive_polish: bool = False,
-    tile: bool = False,
-    tile_size: int = 1024,
-    tile_overlap: int = 128,
-    force: bool = False,
+    options: _BatchOptions,
 ) -> bool:
     """Process a single image for batch mode.
 
@@ -1312,71 +1431,21 @@ def _process_batch_image(
         if image is None:
             raise ValueError("Failed to read image")
 
-        result, _ = _remove_visible_auto(image, source_path=img_path, backend=backend, sensitivity=sensitivity)
+        result, _ = _remove_visible_auto(
+            image,
+            source_path=img_path,
+            backend=options.backend,
+            sensitivity=options.sensitivity,
+        )
 
         image_io.write_bgr_with_alpha(out_path, result, alpha)
         saved_alpha = alpha
 
     if mode in ("invisible", "all"):
-        from remove_ai_watermarks.invisible_engine import (
-            is_available as invisible_available,
-        )
-
         # Skip the destructive regeneration when no invisible watermark is locally
         # detectable (would only degrade a clean image). Read the pristine `img_path`;
         # `out_path` may already be the visible-processed result. --force overrides.
-        skip_no_signal = _should_skip_invisible_scrub(force, img_path)
-        if invisible_available() and not skip_no_signal:
-            from remove_ai_watermarks.invisible_engine import InvisibleEngine
-
-            # Cache the engine in ctx.obj so the batch builds it once (pipeline is a
-            # single CLI value, constant across the run).
-            engines = ctx.obj.setdefault("_inv_engines", {})
-            if pipeline not in engines:
-                engines[pipeline] = InvisibleEngine(
-                    model_id=model,
-                    device=None if device == "auto" else device,
-                    pipeline=pipeline,
-                    hf_token=hf_token,
-                    controlnet_conditioning_scale=controlnet_scale,
-                )
-            engine_inv = engines[pipeline]
-            engine_inv.remove_watermark(
-                img_path if mode == "invisible" else out_path,
-                out_path,
-                strength=strength,
-                num_inference_steps=steps,
-                guidance_scale=guidance_scale,
-                seed=seed,
-                humanize=humanize,
-                unsharp=unsharp,
-                adaptive_polish=adaptive_polish,
-                max_resolution=max_resolution,
-                min_resolution=min_resolution,
-                upscaler=upscaler,
-                tile=tile,
-                tile_size=tile_size,
-                tile_overlap=tile_overlap,
-                # Detect the vendor from the pristine original (`img_path`), not the
-                # visible-processed `out_path` whose C2PA is already gone.
-                vendor=vendor_for_strength(img_path),
-            )
-        elif not invisible_available() and not skip_no_signal:
-            # An invisible signal IS present but the GPU deps are missing, so the
-            # SynthID scrub cannot run. Mirror the single `all` command's loud skip:
-            # flag it for a batch-level warning + non-zero exit (a silently retained
-            # SynthID watermark is the #1 "it didn't work" report). For invisible mode
-            # nothing wrote out_path yet -> copy the input through so the output dir is
-            # complete with the pixels deliberately left intact (without this, a
-            # signal-bearing image in a GPU-less --mode invisible run got NO output).
-            synthid_skipped = True
-            if mode == "invisible" and not out_path.exists():
-                _passthrough_copy(img_path, out_path)
-        elif skip_no_signal and mode == "invisible" and not out_path.exists():
-            # No invisible target and the visible/all pass did not write out_path
-            # (invisible mode): copy the input through so the output dir is complete
-            # with the pixels deliberately left intact.
-            _passthrough_copy(img_path, out_path)
+        synthid_skipped = _run_batch_invisible(ctx, img_path, out_path, mode, options)
 
     if mode in ("metadata", "all"):
         from remove_ai_watermarks.metadata import remove_ai_metadata
@@ -1485,6 +1554,29 @@ def cmd_batch(
     if mode in ("invisible", "all"):
         _warn_if_esrgan_unavailable(upscaler)
     adaptive_polish = _resolve_auto_polish(auto, adaptive_polish)
+    options = _BatchOptions(
+        strength=strength,
+        steps=steps,
+        pipeline=pipeline,
+        device=device,
+        seed=seed,
+        hf_token=hf_token,
+        humanize=humanize,
+        backend=backend,
+        sensitivity=sensitivity,
+        unsharp=unsharp,
+        max_resolution=max_resolution,
+        min_resolution=min_resolution,
+        controlnet_scale=controlnet_scale,
+        upscaler=upscaler,
+        model=model,
+        guidance_scale=guidance_scale,
+        adaptive_polish=adaptive_polish,
+        tile=tile,
+        tile_size=tile_size,
+        tile_overlap=tile_overlap,
+        force=force,
+    )
 
     processed = 0
     errors = 0
@@ -1510,27 +1602,7 @@ def cmd_batch(
                     img_path=img_path,
                     out_path=out_path,
                     mode=mode,
-                    strength=strength,
-                    steps=steps,
-                    pipeline=pipeline,
-                    device=device,
-                    seed=seed,
-                    hf_token=hf_token,
-                    humanize=humanize,
-                    backend=backend,
-                    sensitivity=sensitivity,
-                    unsharp=unsharp,
-                    max_resolution=max_resolution,
-                    min_resolution=min_resolution,
-                    controlnet_scale=controlnet_scale,
-                    upscaler=upscaler,
-                    model=model,
-                    guidance_scale=guidance_scale,
-                    adaptive_polish=adaptive_polish,
-                    tile=tile,
-                    tile_size=tile_size,
-                    tile_overlap=tile_overlap,
-                    force=force,
+                    options=options,
                 ):
                     synthid_skipped_count += 1
                 processed += 1

src/remove_ai_watermarks/identify.py

@@ -505,6 +505,42 @@ def _trustmark(image_path: Path) -> str | None:
     return detect_trustmark(image_path)
 
 
+def _collect_visible_signals(
+    image_path: Path,
+    signals: list[Signal],
+    watermarks: list[str],
+    platform: str | None,
+) -> str | None:
+    """Decode once, append every trusted visible-mark signal, and return platform.
+
+    Keeping this stage separate from metadata aggregation makes the optional cv2
+    boundary explicit and guarantees that all visible detectors share one decoded
+    BGR array. A decode failure preserves the detectors' historical fallback/no-op
+    behavior.
+    """
+    image: NDArray[Any] | None = None
+    try:
+        from remove_ai_watermarks.image_io import imread
+
+        image = imread(image_path)
+    except Exception as exc:  # cv2 missing - detectors fall back / no-op
+        logger.debug("visible-mark decode unavailable: %s", exc)
+
+    sparkle_conf = _visible_sparkle(image_path, image=image)
+    if sparkle_conf is not None and sparkle_conf >= _SPARKLE_THRESHOLD:
+        signals.append(Signal("visible_sparkle", f"NCC confidence {sparkle_conf:.2f}", "medium"))
+        watermarks.append(f"Visible Gemini sparkle (confidence {sparkle_conf:.2f})")
+        if platform is None:
+            platform = "Google Gemini family (visible sparkle detected)"
+
+    for detection in _visible_text_marks(image_path, image=image):
+        signals.append(Signal(f"visible_{detection.key}", f"NCC confidence {detection.confidence:.2f}", "medium"))
+        watermarks.append(f"Visible {detection.label} (confidence {detection.confidence:.2f})")
+        if platform is None:
+            platform = _VISIBLE_MARK_PLATFORM[detection.key]
+    return platform
+
+
 def identify(image_path: Path, *, check_visible: bool = True, check_invisible: bool = True) -> ProvenanceReport:
     """Identify an image's origin platform and watermark inventory.
 
@@ -755,36 +791,8 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
         or xai_sig
     )
 
-    # Decode the file ONCE for every visible-mark detector. The sparkle and the
-    # text-mark detectors both consume a BGR array; letting each re-read the file
-    # was two full cv2 decodes of the same bitmap, which spikes memory on a small
-    # worker. None (cv2 missing / unreadable container) makes each detector fall
-    # back to its own read, preserving the old behavior.
-    vis_image: NDArray[Any] | None = None
     if check_visible:
-        try:
-            from remove_ai_watermarks.image_io import imread
-
-            vis_image = imread(image_path)
-        except Exception as exc:  # cv2 missing - detectors fall back / no-op
-            logger.debug("visible-mark decode unavailable: %s", exc)
-
-    # ── Visible Gemini sparkle (fallback for stripped-metadata case) ─
-    sparkle_conf = _visible_sparkle(image_path, image=vis_image) if check_visible else None
-    if sparkle_conf is not None and sparkle_conf >= _SPARKLE_THRESHOLD:
-        signals.append(Signal("visible_sparkle", f"NCC confidence {sparkle_conf:.2f}", "medium"))
-        watermarks.append(f"Visible Gemini sparkle (confidence {sparkle_conf:.2f})")
-        if platform is None:
-            platform = "Google Gemini family (visible sparkle detected)"
-
-    # ── Visible Doubao / Jimeng text marks (registry; same stripped-metadata
-    #    fallback role as the Gemini sparkle above) ─
-    if check_visible:
-        for det in _visible_text_marks(image_path, image=vis_image):
-            signals.append(Signal(f"visible_{det.key}", f"NCC confidence {det.confidence:.2f}", "medium"))
-            watermarks.append(f"Visible {det.label} (confidence {det.confidence:.2f})")
-            if platform is None:
-                platform = _VISIBLE_MARK_PLATFORM[det.key]
+        platform = _collect_visible_signals(image_path, signals, watermarks, platform)
 
     visible_only = any(s.name.startswith("visible_") for s in signals) and not ai_from_metadata
     hf_only = bool(hf_job) and not ai_from_metadata

src/remove_ai_watermarks/metadata.py

@@ -347,7 +347,7 @@ def has_ai_metadata(image_path: Path) -> bool:
         return True
     # China TC260 AIGC label as a PNG text chunk (the byte scan above catches
     # only the XMP form; the raw-JSON tEXt chunk needs the PIL-based parse).
-    if aigc_label(image_path):
+    if aigc_label(image_path) is not None:
         return True
     # HuggingFace-hosted job marker (hf-job-id PNG text chunk).
     if huggingface_job(image_path):
@@ -887,7 +887,7 @@ def get_ai_metadata(image_path: Path) -> dict[str, str]:
             result["soft_binding"] = ", ".join(vendors)
 
     # China TC260 AI-content label (Doubao and other China-served generators).
-    if aigc := aigc_label(image_path):
+    if (aigc := aigc_label(image_path)) is not None:
         producer = aigc.get("ContentProducer", "")
         result["aigc_label"] = f"China AIGC label (TC260){f'; producer {producer}' if producer else ''}"
 

src/remove_ai_watermarks/watermark_registry.py

@@ -50,16 +50,24 @@ Backend = Literal["auto", "cv2", "migan", "lama"]
 #     a clean corner). Lowest recall on faint/moved marks.
 #   * ``auto`` (default): relax a mark's gate ONLY when the image carries same-product
 #     evidence the mark is there -- metadata provenance for that vendor, or a confidently
-#     detected sibling mark of the same product (see ``resolve_relax``). No evidence ->
+#     detected sibling mark of the same product (see ``resolve_trust``). No evidence ->
 #     stays strict. Safe: it only escalates where the mark is corroborated.
 #   * ``assume_ai``: relax every mark's gate regardless of evidence -- the caller asserts
 #     the image is AI and wants the mark gone (e.g. a metadata-stripped screenshot uploaded
-#     to a watermark remover). Recovers the faint/moved marks the strict gate demotes
-#     (~49% -> ~89% Gemini recall, corpus-measured), at the cost of a harmless small fill
-#     on some clean corners. The library CANNOT infer this from a stripped image -- only the
-#     caller's out-of-band context (the user uploaded to remove a mark) justifies it.
+#     to a watermark remover). Recovers the faint/moved marks the strict gate demotes. The
+#     library CANNOT infer this from a stripped image -- only the caller's out-of-band
+#     context (the user uploaded to remove a mark) justifies it. An assertion that the
+#     image is AI is NOT evidence of WHICH vendor made it, so a mark relaxed on assumption
+#     alone must still clear ``_ASSUMED_CONF_FLOOR``; see that constant for why.
 Sensitivity = Literal["auto", "strict", "assume_ai"]
 
+# The trust level a mark's detection gate is resolved to (see ``resolve_trust``). The
+# split between ``assumed`` and ``confirmed`` is load-bearing: both bypass the engine's
+# false-positive gate, but only ``confirmed`` has evidence naming THIS vendor, which is
+# exactly what that bypass is documented to require (see GeminiEngine.detect_watermark's
+# ``trust_provenance`` contract). ``assumed`` therefore carries a confidence floor.
+Trust = Literal["strict", "assumed", "confirmed"]
+
 # Product family per mark, for the ``auto`` cross-mark corroboration: a confidently
 # detected mark relaxes only OTHER marks of the SAME product (different corners, one
 # product -- the Jimeng wordmark + the Jimeng pill). Doubao and Jimeng are BOTH ByteDance
@@ -120,6 +128,8 @@ class Candidate:
     Carries the mark's verdict at BOTH trust levels (``detected_strict`` = the
     conservative gate, ``detected_relaxed`` = the gate the engine relaxes to under
     provenance/assume), so the arbiter can pick per mark without re-running detection.
+    ``relaxed_confidence`` is the gate-bypassed detection's confidence, which the arbiter
+    needs to apply :func:`assumed_floor_ok` when a mark is relaxed on assumption alone.
     ``features`` is a generic bag of physical measurements a mark's gate may need (the
     mark owns which it reports via ``KnownMark._features``); e.g. the pill supplies
     ``footprint_flat`` (0/1). Empty for marks whose gate needs no extra evidence."""
@@ -128,6 +138,7 @@ class Candidate:
     label: str
     detected_strict: bool
     detected_relaxed: bool
+    relaxed_confidence: float
     features: dict[str, float]  # generic; both construction sites always supply it (empty when none)
 
 
@@ -446,28 +457,59 @@ def detect_marks(
     return [m.detect(image, provenance=m.key in provenance) for m in _REGISTRY if include_explicit or m.in_auto]
 
 
-def resolve_relax(
+# Minimum gate-bypassed confidence a mark must reach when it is relaxed on ASSUMPTION
+# (``assume_ai``) rather than on evidence naming its vendor. Relaxing bypasses the
+# engine's false-positive gate entirely, which is justified by vendor CONFIRMATION; an
+# assumption that the image is AI says nothing about WHICH vendor, so the bypassed
+# detector needs its own floor or it fires on ordinary content.
+#
+# Corpus-measured 2026-07-16 (256 genuine camera captures -- Make/Model/exposure/aperture
+# present and no AI token, so a Gemini sparkle cannot be there -- vs 697 Google-C2PA
+# positives, metadata used only as the label, never fed to the detector):
+#
+#     bypassed threshold   recall   false-fire on clean photos
+#              0.35         82.6%            59.8%     <- the bare detector gate
+#              0.45         66.6%            12.5%
+#              0.50         59.4%             0.0%     <- chosen
+#     strict gate          56.4%             0.0%
+#
+# So 0.35 sat on a cliff: it bought +26pp recall over strict by filling a corner on ~6
+# of every 10 CLEAN photos. At 0.50 the flag is honest -- it still beats strict, for free.
+# Marks absent from this dict relax identically at both levels; their bypassed false-fire
+# on the same negatives is under 1% (doubao 0.8%, jimeng 0.4%, samsung 0.4%).
+_ASSUMED_CONF_FLOOR: dict[str, float] = {"gemini": 0.50}
+
+
+def assumed_floor_ok(key: str, confidence: float) -> bool:
+    """Whether an ``assumed``-trust detection of ``key`` at ``confidence`` is trustworthy
+    enough to act on (see :data:`_ASSUMED_CONF_FLOOR`). Marks with no floor always pass."""
+    floor = _ASSUMED_CONF_FLOOR.get(key)
+    return floor is None or confidence >= floor
+
+
+def resolve_trust(
     key: str,
     *,
     sensitivity: Sensitivity,
     provenance: frozenset[str],
     strict_keys: set[str],
-) -> bool:
-    """Whether mark ``key``'s detection gate is relaxed (strict -> assume level).
+) -> Trust:
+    """The trust level mark ``key``'s detection gate is resolved to.
 
     The single place that turns the ``sensitivity`` policy + evidence into a per-mark
-    boolean (which the engines consume): ``strict`` never relaxes, ``assume_ai`` always
-    relaxes, and ``auto`` relaxes only on same-product evidence -- the vendor confirmed
-    by metadata (``key in provenance``) or a confidently strict-detected sibling of the
-    same product (``_PRODUCT_OF``)."""
+    level (which the engines consume as ``provenance = level != "strict"``). ``strict``
+    never relaxes. A mark is ``confirmed`` only on same-product evidence -- the vendor
+    confirmed by metadata (``key in provenance``) or a confidently strict-detected
+    sibling of the same product (``_PRODUCT_OF``). Without that evidence, ``assume_ai``
+    yields ``assumed`` (relaxed, but subject to :func:`assumed_floor_ok`) and ``auto``
+    stays ``strict``."""
     if sensitivity == "strict":
-        return False
-    if sensitivity == "assume_ai":
-        return True
-    if key in provenance:
-        return True
+        return "strict"
     product = _PRODUCT_OF[key]
-    return any(_PRODUCT_OF[k] == product for k in strict_keys if k != key)
+    confirmed = key in provenance or any(_PRODUCT_OF[k] == product for k in strict_keys if k != key)
+    if confirmed:
+        return "confirmed"
+    return "assumed" if sensitivity == "assume_ai" else "strict"
 
 
 def _keep_pill(keys: set[str], *, provenance: frozenset[str], sensitivity: Sensitivity, footprint_flat: bool) -> bool:
@@ -515,7 +557,7 @@ def _build_candidates(image: NDArray[Any]) -> list[Candidate]:
         strict = m.detect(image, provenance=False)
         relaxed = m.detect(image, provenance=True)
         feats = m.features(image) if (strict.detected or relaxed.detected) else {}
-        cands.append(Candidate(m.key, m.label, strict.detected, relaxed.detected, feats))
+        cands.append(Candidate(m.key, m.label, strict.detected, relaxed.detected, relaxed.confidence, feats))
     return cands
 
 
@@ -523,17 +565,25 @@ def decide(candidates: list[Candidate], context: Context) -> list[Decision]:
     """The removal ARBITER: a pure function turning perception + context into the
     ordered list of marks to remove (and the trust level each was accepted at).
 
-    All policy lives here, in one place: per-mark relaxation (:func:`resolve_relax`,
-    which needs the strict-detected siblings for ``auto`` cross-mark corroboration) and
-    the capture-less pill gate (:func:`_keep_pill`). No image, no I/O -- so it is
-    unit-testable in isolation and the same decision drives every caller."""
+    All policy lives here, in one place: per-mark trust resolution (:func:`resolve_trust`,
+    which needs the strict-detected siblings for ``auto`` cross-mark corroboration), the
+    assumed-trust confidence floor (:func:`assumed_floor_ok`) and the capture-less pill
+    gate (:func:`_keep_pill`). No image, no I/O -- so it is unit-testable in isolation and
+    the same decision drives every caller."""
     strict_keys = {c.key for c in candidates if c.detected_strict}
     fired: list[Decision] = []
     for c in candidates:
-        relax = resolve_relax(
+        trust = resolve_trust(
             c.key, sensitivity=context.sensitivity, provenance=context.provenance, strict_keys=strict_keys
         )
-        if c.detected_relaxed if relax else c.detected_strict:
+        relax = trust != "strict"
+        ok = c.detected_relaxed if relax else c.detected_strict
+        if trust == "assumed" and not assumed_floor_ok(c.key, c.relaxed_confidence):
+            # Relaxed on assumption alone and too weak to trust: fall back to the strict
+            # verdict rather than dropping the mark, so assume_ai is monotonic -- it only
+            # ever ADDS recall over strict, never removes less than strict would.
+            ok, relax = c.detected_strict, False
+        if ok:
             fired.append(Decision(c, relax))
     keys = {d.candidate.key for d in fired}
     if "jimeng_pill" in keys: