Merge remote-tracking branch 'origin/main' into fix/390-review-anchor

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