nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

Browse Source 
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.
This commit is contained in:
Vito Sansevero
2026-10-27 18:30:51 -07:00
parent 44177974cb
commit c8b1630e1d
7 changed files with 378 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+14 -1
View File
@@ -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
py/api/admin.py
+11 -4
View File
@@ -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
py/api/review.py
+8 -1
View File
@@ -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
py/api/review_recording.py
+41 -1
View File
@@ -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
tests/test_review_inspection.py
+15 -12
View File
@@ -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),
tests/test_review_saver_anchor.py
+224
View File
@@ -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"))
utils/review_capture.py
+65 -14
View File
@@ -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)