From c8b1630e1dd3b6ce0d72507351ac7f7947e9352d Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Tue, 27 Oct 2026 18:30:51 -0700 Subject: [PATCH] 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. --- TESTING.md | 15 +- py/api/admin.py | 15 +- py/api/review.py | 9 +- py/api/review_recording.py | 42 +++++- tests/test_review_inspection.py | 27 ++-- tests/test_review_saver_anchor.py | 224 ++++++++++++++++++++++++++++++ utils/review_capture.py | 79 +++++++++-- 7 files changed, 378 insertions(+), 33 deletions(-) create mode 100644 tests/test_review_saver_anchor.py diff --git a/TESTING.md b/TESTING.md index ce4bd6b2..73c4177a 100644 --- a/TESTING.md +++ b/TESTING.md @@ -4592,7 +4592,20 @@ 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 whose branch +that anchor resolves is never queued: from the inspection profile it offers no +choice; a default-ambiguous file with a choice asks `saver_anchored` (the named +run resolves, the unnamed one does not) and is reported `unresolved`, never +linked. Its pending, undecided items leave on the next scan +(`REVIEW_RULE_VERSION` is folded into inspection item versions; a default item +is settled directly). No match, several matches or an unfinished walk anchor +nothing, and the file stays queued as before. Apply 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, so they were never queued. 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 diff --git a/py/api/admin.py b/py/api/admin.py index ec3f6db7..87553e7c 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -1594,7 +1594,8 @@ class AdminRoutesMixin: whose final outcome is ambiguous and unlinked (the resolver found it ambiguous and the fallback did not link it) is recorded for Needs review (best effort, counted in ``review`` when given); one a saved - row already links is reported ``unchanged`` and never queued. An + row already links is reported ``unchanged`` and never queued, and one + whose saver anchors its branch (#390) ``unresolved``. An ambiguous file with nothing to choose is reported ``unresolved`` and its pending items without a choice leave the queue. """ @@ -1629,6 +1630,10 @@ class AdminRoutesMixin: if ambiguous and outcome == "ambiguous": if self._review_file_linked(media_file, roots): outcome = "unchanged" # already linked: not a review case + elif self._saver_anchored_review( + media_file, roots, records.get("canonical"), review + ): + outcome = "unresolved" # its saver names its branch (#390) else: self._record_scanned_review(media_file, roots, review) elif untraced and outcome == "unresolved": @@ -1691,9 +1696,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 +1728,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 diff --git a/py/api/review.py b/py/api/review.py index 6f81e545..dc9f72d6 100644 --- a/py/api/review.py +++ b/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 diff --git a/py/api/review_recording.py b/py/api/review_recording.py index 6de59641..4e30ceea 100644 --- a/py/api/review_recording.py +++ b/py/api/review_recording.py @@ -199,7 +199,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 @@ -210,6 +210,46 @@ class ReviewRecordingMixin: del memory[next(iter(memory))] return None + def _saver_anchored_review(self, media_file, roots, canonical, review=None): + """True when the saver that wrote a default-ambiguous file resolves + its branch (#390), so the file is not queued (blocking). + + Its current bytes' pending, undecided items then leave the queue + (counted in ``review["untraced"]``: nothing to choose). A failure is + logged and counted, and the file is queued as before. + """ + try: + if not review_capture.saver_anchored(canonical, media_file.name): + return False + except Exception: + self.logger.warning( + f"{media_file.name} could not be inspected for Needs review", + exc_info=True, + ) + if review is not None: + review["errors"] += 1 + return False + located = self._review_location(media_file, roots) + if located is None: + return True + try: + facts = os.lstat(located[3]) + # No version is current for these bytes: any pending item goes. + removed = self.db.settle_stale_review_items( + os.path.normcase(located[3]), facts.st_size, facts.st_mtime_ns, () + ) + except Exception: + self.logger.warning( + f"Needs review items of {media_file.name} could not be removed", + exc_info=True, + ) + if review is not None: + review["errors"] += 1 + return True + if review is not None: + review["untraced"] += removed + return True + def _settle_stale_reviews(self, media_file, roots, canonical, review=None): """Remove the pending, undecided items of an untraced file's current bytes recorded under another version (best effort, counted in diff --git a/tests/test_review_inspection.py b/tests/test_review_inspection.py index bec27c2a..3a2485e1 100644 --- a/tests/test_review_inspection.py +++ b/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() @@ -369,15 +371,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 +856,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 +902,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), diff --git a/tests/test_review_saver_anchor.py b/tests/test_review_saver_anchor.py new file mode 100644 index 00000000..3c02f263 --- /dev/null +++ b/tests/test_review_saver_anchor.py @@ -0,0 +1,224 @@ +"""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: a file whose branch that anchor resolves is never queued, and its +pending, undecided items leave the queue on the next scan. No match, several +matches or an unfinished walk anchor nothing, so such a file stays queued as +before. Apply re-resolves an inspection item with the file's name, so a +choice the anchor now 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, + 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, 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 + + +class TestInspectionItemsAnchor(InspectionTestCase): + async def test_file_its_saver_names_is_not_queued(self): + self._roots() + path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS)) + + job = await self._scan() + + self.assertEqual(self._items(), []) + (row,) = job["outcomes"]["review"] + self.assertEqual(row["path"], "branch_1_00001_.png") + self.assertNotIn("more than one prompt could match", row["detail"]) + # Never queued is not linked: ingestion stays on the default profile. + self.assertIsNone(self._linked_text(path)) + + 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)) + + 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_pending_item_of_an_anchored_file_leaves_on_rescan(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() + self.assertEqual(len(self._items()), 1) # queued under the old rule + + job = await self._scan("all") + + self.assertEqual(self._items(), []) + self.assertEqual(job["scope"]["review_untraced"], 1) + + async def test_apply_refuses_a_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_default_ambiguous_file_its_saver_names_is_not_queued(self): + self._roots() + path = self.output_dir / "side_0_00001_.png" + write_png(path, prompt=saved_unlinkable_graph("cat", "dog")) + + job = await self._scan() + + self.assertEqual(self._items(), []) + (row,) = job["outcomes"]["review"] + self.assertNotIn("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.assertEqual(item["resolver_version"], review_capture.RESOLVER_VERSION) + + async def test_pending_default_item_of_an_anchored_file_leaves_on_rescan(self): + self._roots() + write_png( + self.output_dir / "side_0_00001_.png", + prompt=saved_unlinkable_graph("cat", "dog"), + ) + with patch.object(review_capture, "saver_anchored", return_value=False): + await self._scan() + self.assertEqual(len(self._items()), 1) # queued under the old rule + + job = await self._scan("all") + + self.assertEqual(self._items(), []) + self.assertEqual(job["scope"]["review_untraced"], 1) + + async def test_decided_default_item_of_an_anchored_file_is_kept(self): + self._require_routes() + self._roots() + write_png( + self.output_dir / "side_0_00001_.png", + prompt=saved_unlinkable_graph("cat", "dog"), + ) + with patch.object(review_capture, "saver_anchored", return_value=False): + 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) + + +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")) diff --git a/utils/review_capture.py b/utils/review_capture.py index 1859dee7..b635e530 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -40,6 +40,16 @@ 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. A file whose branch +that anchor resolves is never queued, from either profile +(``saver_anchored`` for a default item); no match, several matches or an +unfinished walk anchor nothing and the file stays queued as before. Apply +re-resolves an inspection item with the file's name too. +``REVIEW_RULE_VERSION`` is folded into inspection item versions, so items +recorded before the anchor are re-evaluated by the next scan. """ from __future__ import annotations @@ -108,6 +118,9 @@ INSPECTION = "inspection" # 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 +167,68 @@ 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: +def inspection_choice(metadata: dict, file_name: str | None = None) -> dict | None: """The inspection profile's ambiguous result when it offers a choice. The scan asks only for a file the default profile found ambiguous - without a choice that nothing links. + without a choice that nothing links. ``file_name`` anchors as in + ``resolve_inspection``: a file whose branch its saver resolves offers + no choice. """ if inspection_review_version(metadata) is None: return None - resolution = resolve_inspection(metadata) + resolution = resolve_inspection(metadata, file_name) if resolution.get("state") != "ambiguous" or not reviewable(resolution): return None return resolution +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 found ambiguous with a choice is then not queued. 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. + """ + if not file_name or not isinstance(metadata, dict): + return False + if inspection_review_version(metadata) is None: + return False + if resolve_inspection(metadata, file_name).get("state") != "resolved": + return False + return resolve_inspection(metadata).get("state") != "resolved" + + 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: @@ -285,7 +328,7 @@ def record_ambiguous_file( 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]) + resolution = resolve_inspection(captured[0], _file_name(relative)) if resolution.get("state") != "ambiguous" or not reviewable(resolution): return "untraced" evidence, count, limited = review_evidence( @@ -313,6 +356,11 @@ def record_ambiguous_file( return "limited" if limited else "recorded" +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 +402,17 @@ 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). None when the capture is gone or + carries no prompt metadata. """ if store.acquire(capture.token) is None: return None @@ -389,7 +440,7 @@ 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_inspection(canonical, file_name) return canonical, resolve_prompt_pair(canonical)