fix(review): anchor files on their saver in the scan and Needs review (#390) (#414)

* fix(review): anchor files on their saver in the scan and Needs review (#390)

The scan and Apply now resolve with the file's name, as the workspace
inspector does since #383, so the one saver whose filename_prefix names
the file anchors its branch. A file whose branch that anchor resolves is
never queued (from the inspection profile it offers no choice; a
default-ambiguous file asks saver_anchored) and its pending, undecided
items leave on the next scan. REVIEW_RULE_VERSION is folded into
inspection item versions so old items are re-evaluated. No match, several
matches or an unfinished walk anchor nothing. Apply refuses a choice the
anchor contradicts.

* fix(review): refuse a default choice the saver anchor contradicts (#390)

Apply re-resolves a default item with the file's name: when its saver
anchors the branch, no candidate is offered and the choice is refused.
The scan's anchor check skips bytes carrying a decision and remembers
unanchored file states per process, so a rescan does not inspect an
unchanged file again.

* fix(review): queue an anchored file with its saver's pair as the choice (#390)

Maintainer ruling: a file never falls through both the link and the
queue. When only the saver anchor resolves a file the default profile and
the parser floor leave unlinked, the scan queues it with the anchor's pair
as its one candidate (evidence anchor: saver), never linking it. Files
already linked or resolved by the default path are unchanged. Apply
accepts the anchored pair when the bytes now resolve to it, and an
anchored item is a choice the untraced settle keeps.

* fix(review): upgrade old default items of anchored files, tag anchored results (#390)

A pending default item recorded before the saver anchor no longer
shortcuts the rescan when only the anchor resolves the file: the
anchored item supersedes it. Apply accepts a resolved re-resolution only
when it is tagged as the saver anchor's, the anchored pair must carry a
positive text to choose, and limited anchored evidence still counts as a
choice. TESTING.md describes the ruling.

TESTING.md

@@ -4623,7 +4623,23 @@ files, oldest forgotten first, keyed by 16-byte digests: 8.6 MiB measured at
 the cap). A file whose inspection raises is logged, counted in
 `review_errors` and reported `unresolved`; the scan goes on. An older version's
 pending, undecided item of bytes now untraced leaves the queue
-(`settle_stale_review_items`), so an item never stays stuck at 409. The 16-candidate limited
+(`settle_stale_review_items`), so an item never stays stuck at 409.
+Saver anchor (#390, `tests/test_review_saver_anchor.py`): the scan and Apply
+resolve with the file's name as the workspace inspector does, so the one saver
+whose `filename_prefix` names the file anchors its branch. A file is never both
+unlinked and unqueued (maintainer ruling): when only that anchor resolves it
+(the named inspection run resolves, the unnamed one does not) and neither the
+default profile nor the parser floor links it, the scan queues it with the
+anchor's pair as its one candidate (evidence `anchor: "saver"`; the page
+preselects it), never linking it. Such an item counts as a choice for the
+untraced settle. Files already linked or resolved by the default path are
+unchanged. No match, several matches or an unfinished walk anchor nothing, and
+the file is queued as before. `REVIEW_RULE_VERSION` is folded into inspection
+item versions, so older inspection items are recorded again once. Apply accepts
+the anchored pair and refuses a choice the anchor contradicts. Measured on the
+maintainer's 40,890 outputs: the queue is unchanged (249 files); the 242
+`gpu_unlock_bench` files the inspector resolves are linked by the parser
+floor. The 16-candidate limited
 rule and the 8 MiB bound apply unchanged. The queue heals itself: after each batch, the scan removes the pending
 items of every listed file (read or skipped as known) that a saved row now
 links, and the monitor does the same for each ambiguous output it links

database/review_schema.py

@@ -229,13 +229,15 @@ def evidence_offers_choice(evidence: Any) -> bool:
     """
     if not isinstance(evidence, dict):
         return False
+    # The saver anchor's one pair (#390) is a suggestion to confirm.
+    needed = 1 if evidence.get("anchor") == "saver" else MIN_CHOOSABLE
     if evidence.get("limited"):
         total = evidence.get("candidate_total")
-        return type(total) is int and total >= MIN_CHOOSABLE
+        return type(total) is int and total >= needed
     candidates = evidence.get("candidates")
     if not isinstance(candidates, list):
         return False
-    return sum(1 for c in candidates if choosable(c)) >= MIN_CHOOSABLE
+    return sum(1 for c in candidates if choosable(c)) >= needed
 
 
 def evidence_digest(text: str) -> str:

py/api/admin.py

@@ -1630,7 +1630,14 @@ class AdminRoutesMixin:
                 if self._review_file_linked(media_file, roots):
                     outcome = "unchanged"  # already linked: not a review case
                 else:
-                    self._record_scanned_review(media_file, roots, review)
+                    # The records name the item versions that count; the
+                    # saver anchor (#390) is checked on the captured bytes.
+                    self._record_scanned_review(
+                        media_file,
+                        roots,
+                        review,
+                        scanned=(records.get("canonical"), None),
+                    )
             elif untraced and outcome == "unresolved":
                 # Untraced by the default profile: the inspection profile may
                 # still offer a choice (queued for review, never linked).
@@ -1691,9 +1698,9 @@ class AdminRoutesMixin:
         ``"ambiguous"`` (nothing linked it) is a Needs review case. An
         ambiguous result with nothing to choose (``reviewable`` is False: fewer
         than two candidates carry a prompt) that nothing links returns
-        ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``;
-        its parsed records go to ``records["canonical"]`` when ``records`` is
-        a dict.
+        ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``.
+        The parsed records of either ambiguous file go to
+        ``records["canonical"]`` when ``records`` is a dict.
         """
         if not metadata:
             # No prompt metadata (--disable-metadata, bare savers): only the
@@ -1723,6 +1730,8 @@ class AdminRoutesMixin:
                 records["canonical"] = canonical
         elif resolution["state"] == "ambiguous" and ambiguous is not None:
             ambiguous.append(media_file)
+            if records is not None:
+                records["canonical"] = canonical
         if resolution["state"] == "resolved":
             result = ingest_resolved_pair(
                 self.db, str(media_file), canonical, resolution

py/api/review.py

@@ -677,7 +677,14 @@ def _verify_and_read(store, capture, item, root, need_metadata: bool):
     facts = store.verify(root, relative)
     captured = None
     if need_metadata:
-        captured = captured_review(store, capture, item_profile(item["evidence"]))
+        # The file's own name, so an inspection item resolves anchored on its
+        # saver as it was recorded (#390).
+        captured = captured_review(
+            store,
+            capture,
+            item_profile(item["evidence"]),
+            relative.rsplit("/", 1)[-1],
+        )
     return relative, facts, captured
 
 
@@ -723,6 +730,9 @@ def _candidate_still_offered(resolution: dict, choice: str) -> bool:
     that rewords the top-level reason or reorders candidates leaves a
     pending item applicable.
     """
+    if isinstance(resolution, dict) and resolution.get("anchor") == "saver":
+        # Only the saver anchor resolves the bytes (#390): its pair alone.
+        resolution = {"state": "ambiguous", "candidates": [resolution]}
     if not isinstance(resolution, dict) or resolution.get("state") != "ambiguous":
         return False
     evidence, _count, limited = review_evidence(resolution)

py/api/review_recording.py

@@ -88,7 +88,8 @@ class ReviewRecordingMixin:
     def _record_scanned_review(self, media_file, roots, review=None, scanned=None):
         """Record one ambiguous, unlinked scanned file (logged and counted).
 
-        ``scanned`` is the scan's ``(metadata, inspection_resolution)``.
+        ``scanned`` is the scan's ``(metadata, inspection_resolution)``; the
+        resolution is None when the scan did not run the inspection profile.
         """
         try:
             outcome = self._record_review_sync(
@@ -199,7 +200,7 @@ class ReviewRecordingMixin:
         key = _memory_key(locator)
         if state is not None and memory.get(key) == state:
             return None
-        resolution = review_capture.inspection_choice(canonical)
+        resolution = review_capture.inspection_choice(canonical, media_file.name)
         if resolution is not None:
             memory.pop(key, None)
             return canonical, resolution

tests/test_api_review.py

@@ -1040,8 +1040,8 @@ class TestApplyRefusals(ReviewRoutesTestCase):
         _path, item, detail, observed, cat = await self._opened()
         real = review_capture_module.resolve_prompt_pair
 
-        def reworded(metadata):
-            resolution = real(metadata)
+        def reworded(metadata, **kwargs):
+            resolution = real(metadata, **kwargs)
             return {
                 **resolution,
                 "reason": "a reworded reason from a newer resolver",
@@ -1065,8 +1065,8 @@ class TestApplyRefusals(ReviewRoutesTestCase):
         _path, item, detail, observed, cat = await self._opened()
         real = review_capture_module.resolve_prompt_pair
 
-        def without_cat(metadata):
-            resolution = real(metadata)
+        def without_cat(metadata, **kwargs):
+            resolution = real(metadata, **kwargs)
             kept = [
                 c
                 for c in resolution["candidates"]

tests/test_review_inspection.py

@@ -298,19 +298,21 @@ class TestQwenEditBatchIsQueued(InspectionTestCase):
 
 class TestDefaultProfileStillDecides(InspectionTestCase):
     async def test_ambiguous_file_with_texts_keeps_the_default_profile(self):
-        """A default-ambiguous file with a choice never runs the inspection
-        profile, even when that profile would offer more candidates."""
+        """A default-ambiguous file with a choice is recorded from the
+        default profile, even when the inspection profile would offer more
+        candidates. That profile only checks the saver anchor (#390): no
+        saver names this file, so it anchors nothing."""
         self._roots()
         graph = {
             **_prefixed(qwen_edit_graph("make it night"), "q"),
             **unlinkable_graph("cat", "dog"),
         }
-        self._write(graph=graph)
+        self._write("unsaved_00001_.png", graph=graph)
 
         with patch.object(
             review_capture,
-            "resolve_inspection",
-            side_effect=AssertionError("inspection profile ran"),
+            "inspection_choice",
+            side_effect=AssertionError("inspection choices were asked"),
         ):
             await self._scan()
 
@@ -321,17 +323,24 @@ class TestDefaultProfileStillDecides(InspectionTestCase):
         self.assertEqual(item["resolver_version"], review_capture.RESOLVER_VERSION)
         self.assertEqual(_texts(evidence), ["cat", "dog"])
 
-    async def test_apply_of_a_default_item_never_runs_the_inspection_profile(self):
+    async def test_apply_of_a_default_item_inspects_only_for_the_anchor(
+        self,
+    ):
         self._roots()
         write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
         await self._scan()
-        with patch.object(
-            review_capture,
-            "resolve_inspection",
-            side_effect=AssertionError("inspection profile ran"),
+        with (
+            _counted(review_capture, "resolve_inspection") as calls,
+            patch.object(
+                review_capture,
+                "inspection_choice",
+                side_effect=AssertionError("inspection choices were asked"),
+            ),
         ):
             _item, (status, data) = await self._apply("cat")
         self.assertEqual(status, 200, data)
+        # The saver anchor check (#390): with the file's name, nothing matches.
+        self.assertEqual([c[1] for c in calls], ["two.png"])
         self.assertEqual(self._linked_text(self.output_dir / "two.png"), "cat")
 
 
@@ -369,15 +378,16 @@ class TestApplyInspectionCandidate(InspectionTestCase):
         calls = []
         real = review_capture.resolve_inspection
 
-        def spy(metadata):
-            calls.append(metadata)
-            return real(metadata)
+        def spy(metadata, file_name=None):
+            calls.append((metadata, file_name))
+            return real(metadata, file_name)
 
         with patch.object(review_capture, "resolve_inspection", side_effect=spy):
             _item, (status, data) = await self._apply("make it night")
 
         self.assertEqual(status, 200, data)
         self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0][1], NAME)  # anchored on its saver (#390)
 
     async def test_table_change_keeps_the_item_applicable(self):
         """A table change for a class the graph uses moves the item's
@@ -853,8 +863,8 @@ class TestInspectionResolverVersion(InspectionTestCase):
         version = _version()
         real = review_capture.resolve_inspection
 
-        def newer(metadata):
-            resolution = real(metadata)
+        def newer(metadata, file_name=None):
+            resolution = real(metadata, file_name)
             return {
                 **resolution,
                 "candidates": [
@@ -899,10 +909,10 @@ class TestScanContainment(InspectionTestCase):
         self._write("b.png", graph=qwen_edit_graph("snow", "rain"))
         real = review_capture.resolve_inspection
 
-        def fails_for_a(metadata):
+        def fails_for_a(metadata, file_name=None):
             if "make it night" in json.dumps(metadata):
                 raise RuntimeError("inspection blew up")
-            return real(metadata)
+            return real(metadata, file_name)
 
         with (
             patch.object(review_capture, "resolve_inspection", side_effect=fails_for_a),

tests/test_review_saver_anchor.py

@@ -0,0 +1,333 @@
+"""The scan and Needs review anchor a file on the saver that wrote it (#390).
+
+The workspace inspector (since #383) passes the file's name to the resolver,
+which anchors the file on the one saver whose ``filename_prefix`` names it
+(``_GraphResolver.saver_stages``). The scan and Needs review use the same
+rule. Maintainer ruling: a file never falls through both the link and the
+queue, so when only that anchor resolves a file the default profile and the
+parser floor do not link, it is queued with the anchor's pair as its one
+choice (never linked automatically). No match, several matches or an
+unfinished walk anchor nothing, so such a file is queued as before. Apply
+re-resolves with the file's name, so a choice the anchor contradicts is
+refused as stale evidence.
+"""
+
+import json
+from unittest.mock import patch
+
+try:
+    from tests.test_review import unlinkable_graph, write_png
+    from tests.test_review_inspection import (
+        PROMPTS,
+        _counted,
+        InspectionTestCase,
+        qwen_edit_graph,
+    )
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_review import unlinkable_graph, write_png
+    from test_review_inspection import (
+        PROMPTS,
+        _counted,
+        InspectionTestCase,
+        qwen_edit_graph,
+    )
+
+from utils import review_capture
+
+
+def branch_graph(*prompts, prefixes=None):
+    """A Qwen-edit batch whose every SaveImage has its own literal prefix."""
+    graph = qwen_edit_graph(*prompts)
+    savers = [n for n in graph.values() if n["class_type"] == "SaveImage"]
+    names = prefixes or [f"branch_{index}" for index in range(len(savers))]
+    for saver, prefix in zip(savers, names):
+        saver["inputs"]["filename_prefix"] = prefix
+    return graph
+
+
+def saved_unlinkable_graph(*positives):
+    """``unlinkable_graph`` whose branches save under their own prefix.
+
+    The default profile finds it ambiguous with a choice and the scan's
+    browser-parser floor links nothing, so it is queued from the default
+    profile unless the saver that wrote the file anchors its branch.
+    """
+    graph = unlinkable_graph(*positives)
+    samplers = [
+        node_id
+        for node_id, node in graph.items()
+        if node["class_type"] == "KSampler" and node_id != "3"
+    ]
+    for index, sampler in enumerate(samplers):
+        graph[f"{sampler}1"] = {
+            "class_type": "VAEDecode",
+            "inputs": {"samples": [sampler, 0], "vae": ["1", 2]},
+        }
+        graph[f"{sampler}2"] = {
+            "class_type": "SaveImage",
+            "inputs": {
+                "images": [f"{sampler}1", 0],
+                "filename_prefix": f"side_{index}",
+            },
+        }
+    return graph
+
+
+def _without_name(real):
+    """``resolve_inspection`` as it was before #390: the file name ignored."""
+
+    def unnamed(metadata, file_name=None):
+        return real(metadata)
+
+    return unnamed
+
+
+def _only_candidate(test, item):
+    """The positive text of an anchored item's one candidate."""
+    evidence = json.loads(item["evidence"])
+    test.assertEqual(evidence.get("anchor"), "saver")
+    test.assertEqual(evidence.get("profile"), "inspection")
+    (candidate,) = evidence["candidates"]
+    return candidate["positive"]["text"]
+
+
+class TestInspectionItemsAnchor(InspectionTestCase):
+    async def test_file_its_saver_names_is_queued_with_the_anchor_choice(self):
+        self._roots()
+        path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+
+        job = await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(_only_candidate(self, item), "add a red hat")
+        self.assertEqual(item["candidate_count"], 1)
+        self.assertEqual(
+            item["resolver_version"],
+            review_capture.inspection_review_version(
+                {"prompt": branch_graph(*PROMPTS)}
+            ),
+        )
+        (row,) = job["outcomes"]["review"]
+        self.assertEqual(row["path"], "branch_1_00001_.png")
+        # Queued, never linked: ingestion stays on the default profile.
+        self.assertIsNone(self._linked_text(path))
+
+    async def test_anchored_item_stays_on_rescan(self):
+        self._roots()
+        self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        await self._scan()
+        (item,) = self._items()
+
+        job = await self._scan("all")
+
+        self.assertEqual([i["id"] for i in self._items()], [item["id"]])
+        self.assertEqual(job["scope"]["review_untraced"], 0)
+
+    async def test_file_no_saver_names_stays_queued(self):
+        self._roots()
+        self._write("elsewhere_00001_.png", graph=branch_graph(*PROMPTS))
+
+        await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(item["candidate_count"], len(PROMPTS))
+        self.assertNotIn("anchor", json.loads(item["evidence"]))
+
+    async def test_file_several_savers_name_stays_queued(self):
+        self._roots()
+        graph = branch_graph(*PROMPTS, prefixes=["same", "same", "other"])
+        self._write("same_00001_.png", graph=graph)
+
+        await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(item["candidate_count"], len(PROMPTS))
+
+    async def test_old_rule_item_is_replaced_by_the_anchor_choice(self):
+        self._roots()
+        self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        real = review_capture.resolve_inspection
+        with (
+            patch.object(review_capture, "REVIEW_RULE_VERSION", 0),
+            patch.object(
+                review_capture, "resolve_inspection", side_effect=_without_name(real)
+            ),
+        ):
+            await self._scan()
+        (old,) = self._items()  # queued under the old rule with every branch
+        self.assertEqual(old["candidate_count"], len(PROMPTS))
+
+        await self._scan("all")
+
+        (item,) = self._items()
+        self.assertNotEqual(item["id"], old["id"])
+        self.assertEqual(_only_candidate(self, item), "add a red hat")
+
+    async def test_apply_links_the_anchor_choice(self):
+        self._require_routes()
+        self._roots()
+        path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        await self._scan()
+
+        _item, (status, data) = await self._apply("add a red hat")
+
+        self.assertEqual(status, 200, data)
+        self.assertEqual(self._linked_text(path), "add a red hat")
+
+    async def test_apply_refuses_an_old_choice_the_anchor_contradicts(self):
+        self._require_routes()
+        self._roots()
+        path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        real = review_capture.resolve_inspection
+        with (
+            patch.object(review_capture, "REVIEW_RULE_VERSION", 0),
+            patch.object(
+                review_capture, "resolve_inspection", side_effect=_without_name(real)
+            ),
+        ):
+            await self._scan()
+        item, detail, observed = await self._open()
+        # branch_1 is the saver of "add a red hat"; choose another branch.
+        wrong = self._candidate(detail, "make it night")
+
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, wrong["hash"])
+        )
+
+        self.assertEqual(status, 409, data)
+        self.assertEqual(data["code"], "review-conflict", data)
+        self.assertIsNone(self._linked_text(path))
+
+
+class TestDefaultItemsAnchor(InspectionTestCase):
+    async def test_anchor_resolves_and_nothing_links_so_it_is_queued(self):
+        """Maintainer ruling: never unlinked and unqueued. The anchor resolves
+        the file, the default profile and the parser floor link nothing, so
+        it is queued with the anchor's pair as its choice."""
+        self._roots()
+        path = self.output_dir / "side_0_00001_.png"
+        graph = saved_unlinkable_graph("cat", "dog")
+        write_png(path, prompt=graph)
+        self.assertEqual(
+            review_capture.resolve_prompt_pair({"prompt": graph})["state"], "ambiguous"
+        )
+
+        job = await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(_only_candidate(self, item), "cat")
+        (row,) = job["outcomes"]["review"]
+        self.assertIn("more than one prompt could match", row["detail"])
+        self.assertIsNone(self._linked_text(path))
+
+    async def test_default_ambiguous_file_no_saver_names_stays_queued(self):
+        self._roots()
+        write_png(
+            self.output_dir / "unsaved_00001_.png",
+            prompt=saved_unlinkable_graph("cat", "dog"),
+        )
+
+        await self._scan()
+
+        (item,) = self._items()
+        evidence = json.loads(item["evidence"])
+        self.assertNotIn("profile", evidence)
+        self.assertNotIn("anchor", evidence)
+        self.assertEqual(item["resolver_version"], review_capture.RESOLVER_VERSION)
+
+    async def test_file_the_parser_floor_links_is_not_queued(self):
+        """Already linked by the default path: unchanged by the anchor."""
+        self._roots()
+        graph = saved_unlinkable_graph("cat", "dog")
+        del graph["3"]  # the parser floor now reads a real sampler
+        path = self.output_dir / "side_0_00001_.png"
+        write_png(path, prompt=graph)
+
+        await self._scan()
+
+        self.assertEqual(self._items(), [])
+        self.assertIsNotNone(self._linked_text(path))
+
+    async def test_apply_refuses_an_old_default_choice_the_anchor_contradicts(self):
+        self._require_routes()
+        self._roots()
+        path = self.output_dir / "side_0_00001_.png"
+        write_png(path, prompt=saved_unlinkable_graph("cat", "dog"))
+        with patch.object(review_capture, "_captured_inspection", return_value=None):
+            await self._scan()  # queued from the default profile, as before #390
+        item, detail, observed = await self._open()
+        # side_0 is the saver of "cat"; choose the other branch.
+        wrong = self._candidate(detail, "dog")
+
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, wrong["hash"])
+        )
+
+        self.assertEqual(status, 409, data)
+        self.assertEqual(data["code"], "review-conflict", data)
+        self.assertIsNone(self._linked_text(path))
+
+    async def test_old_default_item_is_replaced_by_the_anchor_choice(self):
+        self._roots()
+        write_png(
+            self.output_dir / "side_0_00001_.png",
+            prompt=saved_unlinkable_graph("cat", "dog"),
+        )
+        with patch.object(review_capture, "_captured_inspection", return_value=None):
+            await self._scan()  # queued from the default profile, as before #390
+        (old,) = self._items()
+        self.assertEqual(old["resolver_version"], review_capture.RESOLVER_VERSION)
+
+        await self._scan("all")
+
+        (item,) = self._items()
+        self.assertNotEqual(item["id"], old["id"])
+        self.assertEqual(_only_candidate(self, item), "cat")
+
+        with _counted(review_capture, "resolve_inspection") as calls:
+            await self._scan("all")  # now queued under its version: no re-read
+        self.assertEqual(calls, [])
+        self.assertEqual([i["id"] for i in self._items()], [item["id"]])
+
+    async def test_decided_anchored_item_is_kept(self):
+        self._require_routes()
+        self._roots()
+        write_png(
+            self.output_dir / "side_0_00001_.png",
+            prompt=saved_unlinkable_graph("cat", "dog"),
+        )
+        await self._scan()
+        item, detail, observed = await self._open()
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, "unresolved")
+        )
+        self.assertEqual(status, 200, data)
+        before = (self._items(), self._decisions())
+
+        await self._scan("all")
+
+        self.assertEqual((self._items(), self._decisions()), before)
+
+    async def test_apply_accepts_an_old_default_choice_the_anchor_names(self):
+        self._require_routes()
+        self._roots()
+        path = self.output_dir / "side_0_00001_.png"
+        write_png(path, prompt=saved_unlinkable_graph("cat", "dog"))
+        with patch.object(review_capture, "_captured_inspection", return_value=None):
+            await self._scan()
+
+        _item, (status, data) = await self._apply("cat")
+
+        self.assertEqual(status, 200, data)
+        self.assertEqual(self._linked_text(path), "cat")
+
+
+class TestSaverAnchored(InspectionTestCase):
+    def test_only_the_anchor_counts(self):
+        anchored = {"prompt": saved_unlinkable_graph("cat", "dog")}
+        self.assertTrue(review_capture.saver_anchored(anchored, "side_1_00001_.png"))
+        self.assertFalse(review_capture.saver_anchored(anchored, "nope_00001_.png"))
+        self.assertFalse(review_capture.saver_anchored(anchored, None))
+        # One sampler resolves without any name: the anchor decides nothing.
+        single = {"prompt": branch_graph("make it night")}
+        self.assertFalse(review_capture.saver_anchored(single, "branch_0_00001_.png"))

utils/review_capture.py

@@ -40,6 +40,19 @@ while it is pending and undecided (never beside a decision, see
 profile the item recorded (``captured_review(..., item_profile(evidence))``).
 The inspection profile only ever produces review choices; ingestion's own
 links stay on the default profile, which never loads the table (D1).
+
+The saver anchor (#390): the inspection profile resolves a file with its
+name, as the workspace inspector does since #383, so the one saver whose
+``filename_prefix`` names the file anchors its branch. When that anchor alone
+resolves a file the default profile left ambiguous and nothing links
+(``saver_anchored``), the scan queues it with the anchor's pair as its one
+candidate (evidence ``anchor: "saver"``), never linking it: a file never
+falls through both the link and the queue (maintainer ruling). No match,
+several matches or an unfinished walk anchor nothing and the file is queued
+as before. Apply re-resolves with the file's name, so only the anchored
+branch can be applied. ``REVIEW_RULE_VERSION`` is folded into inspection
+item versions, so inspection items recorded before the anchor are recorded
+again by the next scan.
 """
 
 from __future__ import annotations
@@ -104,10 +117,16 @@ MAX_CANDIDATE_DEPTH = 8
 # The resolver profile an item's evidence was built with (``profile``).
 DEFAULT = "default"
 INSPECTION = "inspection"
+# The ``anchor`` an item recorded from the saver anchor names (#390).
+SAVER_ANCHOR = "saver"
+ANCHOR_REASON = "The saver that wrote this file names its branch"
 # Inspection item versions: above every plausible ``RESOLVER_VERSION`` and,
 # with 40 bits of the folded digest, below 2**53 (exact as a JSON number).
 INSPECTION_VERSION_BASE = 1 << 40
 _INSPECTION_VERSION_HEX = 10
+# How the scan and Apply use the inspection profile, folded into inspection
+# item versions. 1: they resolve with the file's name (the saver anchor, #390).
+REVIEW_RULE_VERSION = 1
 
 
 # Above this many distinct node classes the whole table is the fingerprint.
@@ -154,38 +173,92 @@ def inspection_review_version(metadata: Any = None) -> int | None:
     """The ``resolver_version`` of an item recorded from the inspection profile.
 
     Folds the profile with its budgets, ``RESOLVER_VERSION``,
-    ``INSPECTION_RESOLVER_VERSION`` and ``semantics_version(metadata)``, so
-    a change of either resolver, of the profile's bounds or of the table
-    about the graph's classes moves it. None when the table cannot be loaded.
+    ``INSPECTION_RESOLVER_VERSION``, ``REVIEW_RULE_VERSION`` and
+    ``semantics_version(metadata)``, so a change of either resolver, of how
+    review uses the profile, of the profile's bounds or of the table about
+    the graph's classes moves it. None when the table cannot be loaded.
     """
     table = semantics_version(metadata)
     if table is None:
         return None
     profile = repr(astuple(INSPECTION_PROFILE))
     seed = f"{profile}:{RESOLVER_VERSION}:{INSPECTION_RESOLVER_VERSION}:{table}"
+    if REVIEW_RULE_VERSION:  # 0 is the seed items had before the rule existed
+        seed += f":review-rule-{REVIEW_RULE_VERSION}"
     folded = hashlib.sha256(seed.encode("utf-8")).hexdigest()
     return INSPECTION_VERSION_BASE + int(folded[:_INSPECTION_VERSION_HEX], 16)
 
 
-def resolve_inspection(metadata: dict) -> dict:
-    """``metadata`` resolved with the inspection profile (bounded work)."""
-    return resolve_prompt_pair(metadata, profile=INSPECTION_PROFILE)
+def resolve_inspection(metadata: dict, file_name: str | None = None) -> dict:
+    """``metadata`` resolved with the inspection profile (bounded work).
+
+    ``file_name`` (the file's last path part) anchors the file on the one
+    saver that wrote it, as in the workspace inspector (#390).
+    """
+    return resolve_prompt_pair(
+        metadata, profile=INSPECTION_PROFILE, file_name=file_name
+    )
 
 
-def inspection_choice(metadata: dict) -> dict | None:
-    """The inspection profile's ambiguous result when it offers a choice.
+def inspection_choice(metadata: dict, file_name: str | None = None) -> dict | None:
+    """The inspection profile's result when the file is a review case.
 
     The scan asks only for a file the default profile found ambiguous
-    without a choice that nothing links.
+    without a choice that nothing links. The result is ambiguous with a
+    choice, or ``resolved`` when ``file_name`` anchors the file on its saver
+    and only that anchor resolves it (``saver_anchored``): the item then
+    offers the anchor's pair. None otherwise.
     """
     if inspection_review_version(metadata) is None:
         return None
-    resolution = resolve_inspection(metadata)
+    resolution = resolve_inspection(metadata, file_name)
+    if resolution.get("state") == "resolved":
+        return _anchor_of(metadata, file_name, resolution)
     if resolution.get("state") != "ambiguous" or not reviewable(resolution):
         return None
     return resolution
 
 
+def _anchor_of(metadata: Any, file_name: str | None, named: dict) -> dict | None:
+    """``named`` tagged ``anchor: "saver"`` when only the saver anchor
+    resolves the file, to a pair with a positive text to choose; else None.
+
+    ``named`` is ``metadata`` resolved with ``file_name``; the profile runs
+    once more without the name only when it is resolved and choosable.
+    """
+    if not file_name or named.get("state") != "resolved":
+        return None
+    try:
+        offered = choosable(_projection(named))
+    except (KeyError, TypeError):
+        return None
+    if not offered or resolve_inspection(metadata).get("state") == "resolved":
+        return None
+    return {**named, "anchor": SAVER_ANCHOR}
+
+
+def saver_anchored(metadata: Any, file_name: str | None) -> bool:
+    """True when the saver that wrote ``file_name`` resolves its branch (#390).
+
+    The inspection profile resolves ``metadata`` with the file's name, and
+    without it the file does not resolve: the anchor alone decides it. A file
+    the default profile left ambiguous is then queued with the anchor's pair
+    as its one candidate. No name, an unusable table, no matching saver,
+    several, or an unfinished walk anchor nothing (False). The profile
+    without the name runs only when the named run resolved.
+    """
+    return _anchored(metadata, file_name) is not None
+
+
+def _anchored(metadata: Any, file_name: str | None) -> dict | None:
+    """The anchored result ``saver_anchored`` describes, else None."""
+    if not file_name or not isinstance(metadata, dict):
+        return None
+    if inspection_review_version(metadata) is None:
+        return None
+    return _anchor_of(metadata, file_name, resolve_inspection(metadata, file_name))
+
+
 def item_profile(evidence: Any) -> str:
     """The profile an item's stored evidence was built with."""
     if isinstance(evidence, dict) and evidence.get("profile") == INSPECTION:
@@ -246,7 +319,9 @@ def record_ambiguous_file(
     """
     versions = item_versions(scanned[0] if scanned is not None else None)
     spelled_path, queued = _queued_spelling(db, root, relative, versions)
-    if queued is not None:
+    if queued is not None and not (
+        inspection and _default_item_to_anchor(db, queued, scanned, relative)
+    ):
         return _already_queued(db, *queued, versions)
     epoch = db.get_execution_epoch()
     try:
@@ -272,24 +347,24 @@ def record_ambiguous_file(
     if resolution is None or resolution.get("state") != "ambiguous":
         return "skipped"
     profile, version = DEFAULT, RESOLVER_VERSION
-    if not reviewable(resolution):
-        version = inspection_review_version(captured[0]) if inspection else None
-        if version is None:
+    named = _captured_inspection(captured[0], relative, scanned) if inspection else None
+    anchored = named is not None and named[0].get("anchor") == SAVER_ANCHOR
+    if anchored:
+        # Only the saver anchor resolves it (#390): offer that one pair.
+        profile, version = INSPECTION, named[1]
+        resolution = {
+            "state": "ambiguous",
+            "reason": ANCHOR_REASON,
+            "candidates": [named[0]],
+        }
+    elif not reviewable(resolution):
+        if named is None:
             return "untraced"
-        profile = INSPECTION
-        if (
-            scanned is not None
-            and scanned[1] is not None
-            and _same_records(captured[0], scanned[0])
-        ):
-            resolution = scanned[1]  # the same records: no second run
-        else:
-            # Heavier, so only for a file ambiguous with nothing to choose.
-            resolution = resolve_inspection(captured[0])
+        profile, (resolution, version) = INSPECTION, named
         if resolution.get("state") != "ambiguous" or not reviewable(resolution):
             return "untraced"
     evidence, count, limited = review_evidence(
-        resolution, profile, version, captured[0]
+        resolution, profile, version, captured[0], anchored=anchored
     )
     try:
         inserted = db.record_review_item(
@@ -313,6 +388,60 @@ def record_ambiguous_file(
     return "limited" if limited else "recorded"
 
 
+def _default_item_to_anchor(db, queued: tuple, scanned, relative: str) -> bool:
+    """True when the bytes' only pending item is a default item recorded
+    before #390 and the saver anchor alone resolves the scanned records.
+
+    The file is then captured again so the anchored item supersedes that
+    pending, undecided item. Decided bytes and bytes already queued under
+    the current inspection version are never re-read.
+    """
+    if scanned is None or scanned[0] is None:
+        return False
+    version = inspection_review_version(scanned[0])
+    if version is None or db.review_pending_at(*queued, version):
+        return False
+    if not db.review_pending_at(*queued, RESOLVER_VERSION):
+        return False
+    return _anchored(scanned[0], _file_name(relative)) is not None
+
+
+def _captured_inspection(metadata: dict, relative: str, scanned) -> tuple | None:
+    """``(resolution, item_version)`` of the captured records under the
+    inspection profile, anchored on the file's name; None when the table is
+    unusable.
+
+    A result tagged ``anchor: "saver"`` means the saver anchor alone
+    resolves the file (#390); any other resolved result is returned as
+    ``unresolved`` (no review case, as before). The scan's result is reused
+    when the capture holds the same records (``inspection_choice`` applies
+    the same rule).
+    """
+    version = inspection_review_version(metadata)
+    if version is None:
+        return None
+    if (
+        scanned is not None
+        and scanned[1] is not None
+        and _same_records(metadata, scanned[0])
+    ):
+        return scanned[1], version  # the same records: no second run
+    name = _file_name(relative)
+    resolution = resolve_inspection(metadata, name)
+    if resolution.get("state") == "resolved":
+        # Resolved without the anchor too: no review case, as before.
+        resolution = _anchor_of(metadata, name, resolution) or {
+            **resolution,
+            "state": "unresolved",
+        }
+    return resolution, version
+
+
+def _file_name(relative: str) -> str:
+    """The last part of a ``/``-separated relative path (the saved name)."""
+    return relative.rsplit("/", 1)[-1]
+
+
 def _already_queued(db, locator: str, size: int, mtime_ns: int, versions) -> str:
     """``duplicate`` while an existing item is pending, else ``decided``."""
     for version in versions:
@@ -354,14 +483,19 @@ def _queued_spelling(db, root, relative: str, versions) -> tuple[str, tuple | No
     return spelled, None
 
 
-def captured_review(store, capture, profile: str = DEFAULT) -> tuple[dict, dict] | None:
+def captured_review(
+    store, capture, profile: str = DEFAULT, file_name: str | None = None
+) -> tuple[dict, dict] | None:
     """``(metadata, resolution)`` of the capture's own bytes, else None.
 
     ``metadata`` is what the scan hands the ingester: prompt and workflow
     JSON-decoded and ``file_info`` (size, dimensions, format) from the
     capture. ``resolution`` comes from ``profile`` (``DEFAULT`` or
-    ``INSPECTION``). None when the capture is gone or carries no prompt
-    metadata.
+    ``INSPECTION``; the latter anchored on ``file_name``, the file's own
+    name, as the item was recorded). With ``file_name`` (Apply), a default
+    result its saver anchor alone resolves (``saver_anchored``) is that
+    anchored result, so only the anchored branch is still offered. None
+    when the capture is gone or carries no prompt metadata.
     """
     if store.acquire(capture.token) is None:
         return None
@@ -389,8 +523,15 @@ def captured_review(store, capture, profile: str = DEFAULT) -> tuple[dict, dict]
         info["format"] = read.format
     canonical["file_info"] = info
     if profile == INSPECTION:
-        return canonical, resolve_inspection(canonical)
-    return canonical, resolve_prompt_pair(canonical)
+        named = resolve_inspection(canonical, file_name)
+        return canonical, _anchor_of(canonical, file_name, named) or named
+    resolution = resolve_prompt_pair(canonical)
+    if file_name is not None and resolution.get("state") == "ambiguous":
+        anchored = _anchored(canonical, file_name)
+        if anchored is not None:
+            # Its saver names its branch (#390): only that one is offered.
+            return canonical, anchored
+    return canonical, resolution
 
 
 def reviewable(resolution: Any) -> bool:
@@ -416,12 +557,14 @@ def review_evidence(
     profile: str = DEFAULT,
     version: int | None = None,
     metadata: Any = None,
+    anchored: bool = False,
 ) -> tuple[dict, int, bool]:
     """``(evidence, candidate_count, limited)`` for one ambiguous result.
 
     Inspection evidence also names its ``profile`` and ``semantics``, the
     table fingerprint of ``metadata``'s graph it was resolved with;
-    ``version`` is the item's version.
+    ``version`` is the item's version. ``anchored`` evidence (the saver
+    anchor's one pair, #390) also names ``anchor: "saver"``.
     """
     base = {
         "resolver_version": RESOLVER_VERSION if version is None else version,
@@ -430,6 +573,8 @@ def review_evidence(
     if profile == INSPECTION:
         base["profile"] = INSPECTION
         base["semantics"] = semantics_version(metadata)
+    if anchored:
+        base["anchor"] = SAVER_ANCHOR
     candidates: list[dict] = []
     seen: set[str] = set()
     total = 0