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

feat(review): offer inspection-profile choices for untraceable batches (#312)

Browse Source 
* feat(review): offer inspection-profile choices for untraceable batches

Qwen image-edit batches run several KSamplers, each with its own
TextEncodeQwenImageEdit prompt, all saved under one prefix. The scan's
default profile knows neither the encoder nor the string nodes, so every
candidate was unresolved and the files left Needs review as "could not be
traced", although the inspection profile resolves every text.

When the default profile is ambiguous without a choice and nothing links
the file, the scan now resolves it with the inspection profile and, with at
least two choosable candidates, queues the item from that evidence. The
evidence records profile "inspection" and the semantics table version
(SHA-256 of its canonical JSON); the item's resolver_version folds the
profile, RESOLVER_VERSION and the table version, so a table change is
handled like a resolver change (I2). Apply re-resolves the capture with the
recorded profile, re-checks the chosen candidate hash and links through the
compatible-only ingestion. The inspection profile never links on its own,
ingestion keeps the default profile, and the live monitor never queues.

* feat(review): supersede stale items and inspect each file state once

A semantics table or resolver version change no longer leaves two pending
items for one file. A new item supersedes the other items of the same bytes
when every one is pending and undecided (revision 0, no decision row, the
self-heal's safe delete). Beside a decided or advanced item nothing new is
stored, and the rescan's stat shortcut skips those bytes under any version.

The inspection item version now folds a narrower fingerprint: the table
restricted to the graph's node classes plus its class-independent parts, so
an edit about other classes leaves every item alone.

The inspection profile runs once per file state. The scan resolves the
untraced file at batch level, skips files already queued or decided at their
current size and mtime under their version, and remembers files it found
untraced in this process. The capture reuses the scan's result when it holds
the same records.

* feat(review): contain inspection failures, version the inspection resolver

Final-review fixes for the inspection-profile review choices.

- An inspection failure on one file is logged, counted in review_errors and
  reported unresolved; the scan goes on.
- INSPECTION_RESOLVER_VERSION and the inspection profile's budgets are folded
  into the item version, so an inspection-only resolver change re-inspects
  and supersedes the item. An older version's pending, undecided item of
  bytes now untraced leaves the queue, so an item never stays stuck at 409.
- Fingerprint tests: an edit of the encoder's own entry moves the version,
  an unused dict-keyed class leaves it.
- The untraced memory is a plain dict of 16-byte digests (8.6 MiB measured
  at its 65,536-entry cap).
- Only scans add items: the monitor's pass records nothing and only clears
  the items of outputs it links. Tests and the browser review fixture
  record through the scan's recorder.
- record_review_item documents the byte-record refresh it commits when it
  refuses a new item.

* fix(review): an unusable semantics table never retires recorded items
This commit is contained in:
Vito
2026-10-23 18:07:14 -07:00
committed by GitHub
parent 6505d339aa
commit 50703c960a
16 changed files with 1730 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+45 -7
View File
@@ -2725,15 +2725,52 @@ still have made the file, so that text is not certain either. Both report
`unresolved` ("could not be traced"), are never linked, and their pending,
undecided items whose own evidence offers no choice (a limited item counts as
a choice when it holds at least two candidates) leave the queue on the next
scan or monitor pass (`settle_untraced_review_items`). The maintainer's
second live check found 39 items whose 11 candidates were all "unresolved". The queue heals itself: after each batch, the scan removes the pending
scan (`settle_untraced_review_items`). The maintainer's
second live check found 39 items whose 11 candidates were all "unresolved".
Inspection-profile choices (`tests/test_review_inspection.py`, ruling a): when
the default profile is ambiguous without a choice and nothing links the file,
the scan (never the monitor) resolves it again with `INSPECTION_PROFILE`, which
knows `TextEncodeQwenImageEdit`, `StringConstantMultiline`, `JoinStringMulti`
and the rest of `web/js/inspection-semantics.json`. With at least two choosable
candidates there, the file is `ambiguous` and is queued from that profile's
evidence, which adds `profile: "inspection"` and `semantics`: the SHA-256 of
`class_projection`, the table restricted to the graph's node classes plus its
class-independent `schema` and `stage_inputs` (the resolver looks the table up
only by a graph node's class). The item's `resolver_version` is
`inspection_review_version(metadata)`, 2^40 plus 40 bits of SHA-256 over the
profile with its budgets, `RESOLVER_VERSION`, `INSPECTION_RESOLVER_VERSION`
(bump it when the inspection profile would resolve the same metadata otherwise)
and that fingerprint. A table edit about the
graph's classes behaves like a resolver change (I2); an edit about other
classes changes nothing. One file never holds two pending items for the same
bytes: a new version's item supersedes the byte record's other items when all
are pending and undecided (revision 0, no decision row, the self-heal's safe
delete); beside a decided or advanced item nothing new is stored, and the
rescan's stat shortcut then skips those bytes under any version. A pending item
stays applicable while the bytes still offer its chosen candidate.
Apply re-resolves the capture with the profile the item recorded and re-checks
the chosen hash; the link goes through the same compatible-only ingestion. The
inspection profile never links on its own, and ingestion keeps the default
profile (D1). The heavier profile runs only for files already ambiguous without
a choice, and once per file state: the capture reuses the scan's result when
it holds the same records; a rescan skips files already queued or decided at
their current size and `mtime_ns` under their version, and files this process
already found untraced at the same state (an in-memory map of at most 65,536
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
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
(`settle_linked_review_items`; the review index revision moves). Only an
undecided item goes (`pending`, revision 0, no decision row): a decision is
never deleted or changed, so no schema state was added. The monitor never
queues an output it linked; an unlinked file passed to `record_review_file`
follows the scan's rule (a file the fallback would link is `skipped`). The
records an item (only scans add items, maintainer ruling 2026-10-23): an
unlinked file passed to `record_review_file` is `skipped` with no capture,
and only a linked one's pending items are settled
(`test_monitor_never_records_an_unlinked_ambiguous_output`). The
maintainer's live check of #284/#288 found the queue at 10,232 against a review
group of 4,677: every fallback-linked ambiguous file had been queued. It captures
the file (scope `needs-review`), reads its metadata again from the capture,
@@ -2765,9 +2802,10 @@ Apply, and waits up to 2 s for room before counting the file as not added
pass runs after its own link, best effort: a failure is logged and the link
stays. Recording follows the registry's `records`
admission (#289): the scan's recording runs under its own `records` job, and
the monitor's (one file, not a job) is refused while a `records` or `guarded`
job or hold runs, except thumbnail work, which writes no records (S5 M7). It then records nothing, logs that the file waits for the
next scan, and answers `deferred`; the next "new" scan reads the ambiguous file
the monitor's pass (one file, not a job, which only settles) is refused
while a `records` or `guarded` job or hold runs, except thumbnail work, which
writes no records (S5 M7). It then writes nothing, logs that the file waits
for the next scan, and answers `deferred`; the next "new" scan reads the ambiguous file
again (`test_monitor_recording_is_refused_beside_record_changing_jobs`). A new item moves `review_index_revision`. A
rescan does not copy or hash an ambiguous file whose current size and
`mtime_ns` equal the verified fingerprint of bytes an item already cites (a
database/review.py
+89 -2
View File
@@ -90,6 +90,30 @@ _PENDING_AT = (
"WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.resolver_version=? "
"AND i.state='pending' LIMIT 1"
)
# Bytes verified at this stat whose item carries a decision (any version).
_DECIDED_AT = (
f"SELECT 1 FROM {FILES_TABLE} f " # nosec B608
f"JOIN {BYTES_TABLE} b ON b.file_id=f.id "
f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id "
"WHERE f.locator=? AND b.size=? AND b.mtime_ns=? "
"AND (i.state<>'pending' OR i.revision<>0 OR EXISTS (SELECT 1 FROM "
f"{DECISIONS_TABLE} d WHERE d.item_id=i.id)) LIMIT 1"
)
# The other items of one byte record (another version or evidence), with
# whether each may be superseded: pending, revision 0, no decision row.
_OTHER_ITEMS_OF_BYTES = (
"SELECT i.id, (i.state='pending' AND i.revision=0 AND NOT EXISTS " # nosec B608
f"(SELECT 1 FROM {DECISIONS_TABLE} d WHERE d.item_id=i.id)) "
f"FROM {ITEMS_TABLE} i WHERE i.byte_id=? "
"AND NOT (i.resolver_version=? AND i.evidence_hash=?)"
)
# Pending items of the bytes verified at one stat, with their versions.
_PENDING_AT_STAT = (
f"SELECT i.id,i.resolver_version FROM {FILES_TABLE} f " # nosec B608
f"JOIN {BYTES_TABLE} b ON b.file_id=f.id "
f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id "
"WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.state='pending'"
)
# Pending items of one file: the candidates a link settles.
_PENDING_OF_LOCATOR = (
f"SELECT i.id FROM {FILES_TABLE} f " # nosec B608
@@ -274,6 +298,12 @@ class ReviewOperations:
``LibraryCommandError("storage-changed")`` when the storage epoch is
no longer ``epoch`` (a restore since the capture) and ``ValueError``
for invalid facts or evidence.
One file never holds two pending items for the same bytes: the new
item supersedes the byte record's other items (an older resolver or
semantics version) when every one is pending and undecided (the safe
delete of the self-heal); when one carries a decision or a later
revision, it is kept and nothing new is stored (False).
"""
_validate_target(locator, sha256, size, mtime_ns)
_validate_spelling(spelled_path, locator)
@@ -297,6 +327,18 @@ class ReviewOperations:
)
if created or moved:
conn.execute(_BUMP_IMAGE_INDEX)
# The byte record is created or refreshed even when the item is
# refused below (an observe would do the same); that commits
# with the image index bump, and the review queue is unchanged.
others = conn.execute(
_OTHER_ITEMS_OF_BYTES, (byte_id, resolver_version, evidence_hash)
).fetchall()
if any(not removable for _id, removable in others):
return False
superseded = sum(
conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
for item_id, _removable in others
)
cursor = conn.execute(
_INSERT_ITEM,
(
@@ -314,9 +356,9 @@ class ReviewOperations:
)
inserted = cursor.rowcount > 0
refreshed = _refresh_spelling(conn, locator, spelled_path)
if inserted or refreshed:
if inserted or refreshed or superseded:
conn.execute(_BUMP_REVIEW_INDEX)
if inserted or refreshed:
if inserted or refreshed or superseded:
notify_review_change(self.model.db_path)
return inserted
@@ -389,6 +431,19 @@ class ReviewOperations:
)
return row is not None
def review_decided_at(self, locator: str, size: int, mtime_ns: int) -> bool:
"""True when bytes verified at this stat carry a decided item.
Any version: a newer resolver or table never adds an item beside a
decision, so the rescan need not copy those bytes again.
"""
row = (
self.model.get_connection()
.execute(_DECIDED_AT, (locator, size, mtime_ns))
.fetchone()
)
return row is not None
def review_pending_at(
self, locator: str, size: int, mtime_ns: int, resolver_version: int
) -> bool:
@@ -445,6 +500,38 @@ class ReviewOperations:
notify_review_change(self.model.db_path)
return removed
def settle_stale_review_items(
self, locator: str, size: int, mtime_ns: int, versions
) -> int:
"""Remove pending, undecided items of the bytes verified at this stat
whose ``resolver_version`` is not one of ``versions``.
The caller found these bytes untraced under the current versions, so
an older version's item could only refuse Apply. The same safe delete
as the other settles; one indexed read when nothing is pending.
"""
_validate_locator(locator)
current = set(versions)
conn = self.model.get_connection()
stale = [
item_id
for item_id, version in conn.execute(
_PENDING_AT_STAT, (locator, size, mtime_ns)
)
if version not in current
]
if not stale:
return 0
removed = 0
with self._pair_transaction() as conn:
for item_id in stale:
removed += conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
if removed:
conn.execute(_BUMP_REVIEW_INDEX)
if removed:
notify_review_change(self.model.db_path)
return removed
def settle_untraced_review_items(self, locators) -> int:
"""Remove the pending, undecided items that offer nothing to choose.
py/api/admin.py
+19 -4
View File
@@ -1646,7 +1646,7 @@ class AdminRoutesMixin:
if self._scan_job and self._scan_job.cancelled:
break
counts["processed"] += 1
ambiguous, untraced = [], []
ambiguous, untraced, records = [], [], {}
try:
outcome = self._ingest_scanned_file(
media_file,
@@ -1654,6 +1654,7 @@ class AdminRoutesMixin:
detailed=True,
ambiguous=ambiguous,
untraced=untraced,
records=records,
)
except Exception as e:
self.logger.error(f"Error processing {media_file.name}: {e}")
@@ -1666,7 +1667,11 @@ class AdminRoutesMixin:
else:
self._record_scanned_review(media_file, roots, review)
elif untraced and outcome == "unresolved":
self._settle_untraced_reviews(untraced, roots, review)
# Untraced by the default profile: the inspection profile may
# still offer a choice (queued for review, never linked).
outcome = self._untraced_scan_outcome(
media_file, roots, records.get("canonical"), review
)
if outcome is None:
note(media_file, None)
continue
@@ -1701,7 +1706,13 @@ class AdminRoutesMixin:
return positive.strip() or None
def _ingest_scanned_file(
self, media_file, metadata, detailed=False, ambiguous=None, untraced=None
self,
media_file,
metadata,
detailed=False,
ambiguous=None,
untraced=None,
records=None,
):
"""Enrich exact pairs; retain legacy positive-only import for absent pairs.
@@ -1714,7 +1725,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``.
``"unresolved"`` instead, and ``media_file`` goes to ``untraced``;
its parsed records go to ``records["canonical"]`` when ``records`` is
a dict.
"""
if not metadata:
return None
@@ -1735,6 +1748,8 @@ class AdminRoutesMixin:
if unchoosable:
if untraced is not None:
untraced.append(media_file)
if records is not None:
records["canonical"] = canonical
elif resolution["state"] == "ambiguous" and ambiguous is not None:
ambiguous.append(media_file)
if resolution["state"] == "resolved":
py/api/review.py
+11 -4
View File
@@ -38,13 +38,17 @@ try:
from ...database.library_commands import LibraryCommandError
from ...database.review_decisions import ReviewCommandError
from ...utils.media_capture import CaptureError
from ...utils.review_capture import captured_review, review_evidence
from ...utils.review_capture import (
captured_review,
item_profile,
review_evidence,
)
except ImportError:
from database.library import display_title
from database.library_commands import LibraryCommandError
from database.review_decisions import ReviewCommandError
from utils.media_capture import CaptureError
from utils.review_capture import captured_review, review_evidence
from utils.review_capture import captured_review, item_profile, review_evidence
ITEMS_URL = "/prompt_manager/review/items"
DEFAULT_LIMIT = 40
@@ -641,7 +645,8 @@ def _verify_and_read(store, capture, item, root, need_metadata: bool):
"""Worker thread: ``(relative, facts, captured)``, or None off the root.
Formats the item's target under the capture's root (resolving the root's
real path), re-hashes the current file, then reads the capture.
real path), re-hashes the current file, then reads the capture and
resolves it with the profile the item's evidence was recorded with.
"""
found = format_review_target(
item["spelled_path"],
@@ -652,7 +657,9 @@ def _verify_and_read(store, capture, item, root, need_metadata: bool):
return None
relative = found[1]
facts = store.verify(root, relative)
captured = captured_review(store, capture) if need_metadata else None
captured = None
if need_metadata:
captured = captured_review(store, capture, item_profile(item["evidence"]))
return relative, facts, captured
py/api/review_recording.py
+179 -31
View File
@@ -8,21 +8,25 @@ already. A file the fallback links, an exact pair the ingester stored and an
already linked file are never queued. Neither is an ambiguous file with
nothing to choose: fewer than two of its candidates carry a prompt
(``utils.review_capture.reviewable``); the scan reports it unresolved ("could
not be traced").
not be traced") unless the inspection profile finds a choice in it.
Only scans add items (maintainer ruling, 2026-10-23). The monitor's pass
records nothing: it only removes the pending items of an output it linked.
The queue heals itself: a pending item whose file is now linked (by the
fallback, a later resolution or the monitor) is removed by the next scan
that lists the file or by the monitor's next pass over it, and so is a
pending item without a choice whose file is still ambiguous without one
(``settle_untraced_review_items``). Only an undecided item goes
that lists the file or by the monitor's next pass over it. The next scan
also removes a pending item without a choice whose file is still ambiguous
without one (``settle_untraced_review_items``), and an older version's
item of bytes now untraced (``settle_stale_review_items``). Only an undecided item goes
(``pending``, revision 0, no decision row); a decision is never deleted or
changed (``settle_linked_review_items``).
The scan records through ``_record_scanned_review`` (best effort: a failure
is logged and counted); the monitor calls ``record_review_file`` from its
worker thread after its own link. Both go through
``utils.review_capture.record_ambiguous_file`` on the API's capture store.
Nothing here links a file.
is logged and counted), which goes through
``utils.review_capture.record_ambiguous_file`` on the API's capture store;
the monitor calls ``record_review_file`` from its worker thread after its
own link. Nothing here links a file.
Recording changes records, so it follows the registry's ``records``
admission (#289). The scan records under its own ``records`` job. The
@@ -35,6 +39,7 @@ defer the recording (#249 S5 M7): a long rebuild no longer holds Needs review
back for its whole run.
"""
import hashlib
import os
from pathlib import Path
@@ -50,10 +55,29 @@ QUEUED_OUTCOMES = ("recorded", "limited", "duplicate")
NEW_OUTCOMES = ("recorded", "limited")
try:
from ...utils import review_capture
from ...utils.review_capture import record_ambiguous_file
except ImportError:
from utils import review_capture
from utils.review_capture import record_ambiguous_file
# Untraced files remembered per process so a rescan of an unchanged file does
# not run the inspection profile again (oldest forgotten first). A plain dict
# of 16-byte digests (the locator, and its size, mtime_ns and item version):
# measured at 8.6 MiB with 65,536 entries (tracemalloc, CPython 3.14).
MAX_UNTRACED_MEMORY = 65_536
def _digest(text: str) -> bytes:
data = text.encode("utf-8", "surrogatepass")
return hashlib.blake2b(data, digest_size=16).digest()
def _memory_key(locator: str) -> bytes:
"""A 16-byte digest of ``locator`` (the untraced memory's key)."""
return _digest(locator)
# The scan log's reason for items that leave the queue with nothing to choose.
NOTHING_TO_CHOOSE = "nothing to choose"
@@ -61,10 +85,15 @@ NOTHING_TO_CHOOSE = "nothing to choose"
class ReviewRecordingMixin:
"""Record the scan's and the monitor's ambiguous files for Needs review."""
def _record_scanned_review(self, media_file, roots, review=None):
"""Record one ambiguous, unlinked scanned file (logged and counted)."""
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)``.
"""
try:
outcome = self._record_review_sync(media_file, roots, review)
outcome = self._record_review_sync(
media_file, roots, review, inspection=True, scanned=scanned
)
except Exception:
self.logger.warning(
f"{media_file.name} could not be added to Needs review",
@@ -80,9 +109,13 @@ class ReviewRecordingMixin:
if outcome in QUEUED_OUTCOMES:
review["queued"] += 1
def _record_review_sync(self, media_file, roots, review=None):
def _record_review_sync(
self, media_file, roots, review=None, inspection=False, scanned=None
):
"""``record_ambiguous_file`` for a file under one of ``roots`` (blocking).
``inspection`` (the scan only) lets a file without a choice under the
default profile be recorded from the inspection profile's choices.
An ``untraced`` file (its captured bytes offer nothing to choose) also
loses its pending items that offer nothing either.
"""
@@ -91,12 +124,134 @@ class ReviewRecordingMixin:
return "skipped"
root, index, relative, _spelled = located
outcome = record_ambiguous_file(
self.db, self._capture_store(), Path(root), index, relative
self.db,
self._capture_store(),
Path(root),
index,
relative,
inspection=inspection,
scanned=scanned,
)
if outcome == "untraced":
self._settle_untraced_reviews([media_file], roots, review)
return outcome
def _untraced_scan_outcome(self, media_file, roots, canonical, review=None):
"""The scan status of an unlinked file the default profile left
ambiguous without a choice (blocking).
``ambiguous`` when the inspection profile offers a choice: the file is
recorded for review from that profile (never linked), or ``unchanged``
when a saved row links it already. Otherwise ``unresolved``, and its
pending items without a choice leave the queue. The profile runs at
most once per file state: not for bytes already queued (or decided)
at the file's current size and ``mtime_ns`` under the version that
its records depend on, and not again for a state it already found
untraced in this process.
"""
try:
choice = self._inspection_choice(media_file, roots, canonical)
except Exception:
# One bad file never stops the scan: it stays "could not be traced".
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 "unresolved"
if choice is None:
self._settle_untraced_reviews([media_file], roots, review)
self._settle_stale_reviews(media_file, roots, canonical, review)
return "unresolved"
if self._review_file_linked(media_file, roots):
return "unchanged"
self._record_scanned_review(media_file, roots, review, scanned=choice)
return "ambiguous"
def _inspection_choice(self, media_file, roots, canonical):
"""``(canonical, resolution_or_None)`` when the file offers a choice.
The resolution is None when the file's current state is already in
the queue (so the profile did not run). None when it offers none.
"""
if canonical is None:
return None
version = review_capture.inspection_review_version(canonical)
located = self._review_location(media_file, roots)
if version is None or located is None:
return None
locator = os.path.normcase(located[3])
try:
facts = os.lstat(located[3])
except OSError:
facts = None
state = None
if facts is not None:
state = _digest(f"{facts.st_size}:{facts.st_mtime_ns}:{version}")
if self.db.review_decided_at(
locator, facts.st_size, facts.st_mtime_ns
) or self.db.review_fingerprint_queued(
locator, facts.st_size, facts.st_mtime_ns, version
):
return canonical, None
memory = self._untraced_memory()
key = _memory_key(locator)
if state is not None and memory.get(key) == state:
return None
resolution = review_capture.inspection_choice(canonical)
if resolution is not None:
memory.pop(key, None)
return canonical, resolution
if state is not None:
memory.pop(key, None) # re-inserted last: the newest state
memory[key] = state
while len(memory) > MAX_UNTRACED_MEMORY:
del memory[next(iter(memory))]
return None
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
``review["untraced"]``).
The current resolver and table offer nothing to choose in these
bytes, so an item of an older version would only refuse Apply.
"""
located = self._review_location(media_file, roots)
if located is None or canonical is None:
return 0
# An unusable semantics table says nothing about the current
# versions; never let it retire items a working table recorded.
if review_capture.inspection_review_version(canonical) is None:
return 0
try:
facts = os.lstat(located[3])
removed = self.db.settle_stale_review_items(
os.path.normcase(located[3]),
facts.st_size,
facts.st_mtime_ns,
review_capture.item_versions(canonical),
)
except Exception:
self.logger.warning(
f"Stale Needs review items of {media_file.name} could not be "
"removed",
exc_info=True,
)
if review is not None:
review["errors"] += 1
return 0
if review is not None:
review["untraced"] += removed
return removed
def _untraced_memory(self):
memory = getattr(self, "_untraced_inspections", None)
if memory is None:
memory = self._untraced_inspections = {}
return memory
@staticmethod
def _review_location(media_file, roots, real_roots=None):
"""``(root, root_index, relative, spelled)`` of a listed file, or None.
@@ -178,24 +333,19 @@ class ReviewRecordingMixin:
review["untraced"] += removed
return removed
def _fallback_links(self, metadata):
"""True when the scan's legacy fallback would link these records."""
return bool(metadata.get("prompt")) and bool(self._graph_prompt_text(metadata))
def record_review_file(self, image_path):
"""Apply the Needs review rule to one monitor output (its worker thread).
Maps ``image_path`` to the gallery root that lists it; a file under no
current root is ``skipped``. A linked file is never queued: its
pending items are removed and the answer is ``linked``. An unlinked
file the scan's fallback would link is ``skipped`` (the next scan
links it); any other file is recorded when it is ambiguous with a
choice, and is ``untraced`` (its pending items without a choice
removed) when it is ambiguous without one. While a
job runs that a ``records`` job may not run beside, other than
thumbnail work (``PREVIEW_ONLY_KINDS``), nothing is written and the
answer is ``deferred``. Errors propagate: the monitor logs
them and keeps its link.
Only scans add items (maintainer ruling, 2026-10-23): the monitor
only clears the items of outputs it links. Maps ``image_path`` to the
gallery root that lists it; a file under no current root is
``skipped``. A linked file's pending items are removed and the answer
is ``linked``. Any other file is ``skipped``: nothing is captured or
recorded, and the next scan reads it. While a job runs that a
``records`` job may not run beside, other than thumbnail work
(``PREVIEW_ONLY_KINDS``), nothing is written and the answer is
``deferred``. Errors propagate: the monitor logs them and keeps its
link.
"""
busy = self._jobs.blocker(
REVIEW_RECORDING_KIND, admission="records", compatible=PREVIEW_ONLY_KINDS
@@ -215,9 +365,7 @@ class ReviewRecordingMixin:
if self._review_file_linked(located, roots):
self._settle_scanned_reviews([located], roots)
return "linked"
if self._fallback_links(self._extract_comfyui_metadata(str(located))):
return "skipped"
return self._record_review_sync(located, roots)
return "skipped"
@staticmethod
def _log_review_items(job, review):
tests/browser/workspace-review.spec.mjs
+10 -10
View File
@@ -1,6 +1,6 @@
// Needs review (#249 PR F, plan Tasks F1 and F2, Mock inventory M4): the SAVED VIEWS row and its count, the Metadata
// review page over the real review routes, and Activity's Review metadata. Items are recorded by the real recorder
// (tests/browser/workspace_review_fixture.py) through the monitor's entry point; the conflict item's file was linked
// (tests/browser/workspace_review_fixture.py) through the scan's recorder (only scans add items); the conflict item's file was linked
// to another prompt after it was queued. An open page re-reads its queue from the review change stream (F2).
import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
import { test, expect } from './fixtures.mjs';
@@ -318,13 +318,13 @@ test('keyboard focus labels and touch targets', async ({ page, request }) => {
await expect(page).toHaveURL(/#\/images$/);
});
test('an open page shows items the monitor adds and keeps the open choice', async ({ page, request }) => {
test('an open page shows items a scan adds and keeps the open choice', async ({ page, request }) => {
await seed(request, 1);
await open(page);
await expect.poll(async () => (await streams(request)).streams).toBe(1);
await context(page).selectOption({ label: OPTIONS[1] });
await expect(reviewRow(page)).toHaveText('Needs review1');
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect(cards(page)).toHaveCount(2);
await expect(main(page).locator('.notice h2')).toHaveText('2 images need a decision');
await expect(reviewRow(page)).toHaveText('Needs review2');
@@ -381,7 +381,7 @@ test('a visit reads its queue once, and a change reads it once more', async ({ p
const reads = reviewReads(page);
await open(page);
await expect.poll(async () => (await streams(request)).streams).toBe(1);
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect(cards(page)).toHaveCount(2);
await expect(reviewRow(page)).toHaveText('Needs review2');
// The page's first read and the change's re-read; the shell's count read at start, and the row follows the page.
@@ -423,7 +423,7 @@ test('a stream the browser gives up on is opened again, and the page says it is
// The page opens the stream again (after 3 s) and is live once it sends ready.
await expect.poll(async () => (await streams(request)).streams, { timeout: 15000 }).toBe(1);
await expect(live(page)).toBeHidden();
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect(cards(page)).toHaveCount(2);
expect(attempts).toBe(2);
});
@@ -444,8 +444,8 @@ test('a hidden tab closes its stream and a visible one reads only what changed m
await open(page);
await expect.poll(async () => (await streams(request)).streams).toBe(1);
await setVisible(page, false);
// The server drops a closed stream at its next write: this monitor addition is one.
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
// The server drops a closed stream at its next write: this scan addition is one.
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect.poll(async () => (await streams(request)).streams).toBe(0);
await expect(cards(page)).toHaveCount(1);
expect(reads.filter(search => search === '?limit=40')).toHaveLength(1);
@@ -476,7 +476,7 @@ test('a re-read refreshes a card whose media or target changed', async ({ page,
const second = cards(page).nth(1);
await expect(second.getByRole('button', { name: 'Open review_0001.png', exact: true })).not.toHaveAttribute('aria-disabled', 'true');
rewrite = true;
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect(cards(page)).toHaveCount(3);
await expect(cards(page).first().locator('img')).toHaveAttribute('src', /&thumb=1$/);
await expect(second.getByRole('button', { name: 'Open review_0001.png', exact: true })).toHaveAttribute('aria-disabled', 'true');
@@ -522,7 +522,7 @@ test('a live re-read keeps the loaded depth and the scroll position', async ({ p
const before = await scrollTop();
expect(before).toBeGreaterThan(0);
const reads = reviewReads(page);
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect(main(page).locator('.notice h2')).toHaveText('46 images need a decision');
await expect(cards(page)).toHaveCount(46);
expect(await scrollTop()).toBe(before);
@@ -550,7 +550,7 @@ test('teardown releases captures listeners and requests', async ({ page, request
await expect.poll(async () => (await streams(request)).captures).toBe(0);
await route.abort().catch(() => {});
// The server drops a closed stream at its next write: a monitor addition is one.
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
await expect.poll(async () => (await streams(request)).streams).toBe(0);
await expect.poll(() => inFlight.size).toBe(0);
});
tests/browser/workspace_review_fixture.py
+34 -26
View File
@@ -5,8 +5,9 @@ encoders and one shared negative, so the resolver finds it ``ambiguous``. Its
loader node carries ComfyUI's ``Infinity`` stamp, which the legacy pages'
parser (the scan's fallback) cannot read, so nothing links the file and it is
a Needs review case (#284: only ambiguous, unlinked files are queued). It is
recorded through the producers' entry point (``record_review_file``). No path
or SQL comes from a request.
recorded through the scan's recorder (``_record_review_sync`` with
``inspection=True``; only scans add items). No path or SQL comes from a
request.
``POST /__test__/review/reset`` also takes the special items F2 needs, queued
after the plain ones in this order:
@@ -28,10 +29,9 @@ its first negative, the one the concept's panel shows), so the panel's text
wraps as the concept's does.
``POST /__test__/review/add {kind}`` adds one file while a page is open:
``monitor`` records a new ambiguous, unlinked output from a worker thread
through ``record_review_file`` (the monitor's own entry point; the monitor
never queues an output it linked itself), ``file`` only writes one (a
metadata scan records it). ``GET
``scan`` records a new ambiguous, unlinked output from a worker thread
through the scan's recorder (as a running scan's batch does; the monitor
never adds items), ``file`` only writes one (a metadata scan records it). ``GET
/__test__/review/links`` answers each review file's prompt and pair links, and
``GET /__test__/review/streams`` the open change streams and captures.
"""
@@ -139,8 +139,16 @@ def _png(
image.save(path, pnginfo=info)
def _scan_record(api: PromptManagerAPI, path: Path) -> str:
"""Record ``path`` as the scan's recorder does."""
from py.api.admin import _scan_roots
roots = _scan_roots(api._get_all_output_dirs())
return api._record_review_sync(Path(str(path)), roots, inspection=True)
def _record(api: PromptManagerAPI, path: Path, expected: str) -> None:
outcome = api.record_review_file(str(path))
outcome = _scan_record(api, path)
if outcome != expected:
raise RuntimeError(f"{path.name} was {outcome}, not {expected}")
@@ -150,19 +158,19 @@ def _stamp(path: Path, age: float) -> None:
os.utime(path, (stamp, stamp))
class _Monitor:
"""The monitor's recorder over this server's database (one per server)."""
class _ScanRecorder:
"""The scan's recorder over this server's database (one per server)."""
def __init__(self, api: PromptManagerAPI) -> None:
self.api = api
self.lock = threading.Lock()
def new_output(self, path: Path) -> None:
"""The monitor's Needs review pass for a new output, from a worker thread."""
"""The scan's Needs review recording of a new output, from a worker thread."""
outcome = []
with self.lock:
worker = threading.Thread(
target=lambda: outcome.append(self.api.record_review_file(str(path)))
target=lambda: outcome.append(_scan_record(self.api, path))
)
worker.start()
worker.join()
@@ -170,9 +178,9 @@ class _Monitor:
raise RuntimeError(f"{path.name} was {outcome}, not recorded")
def _monitored_output(api: PromptManagerAPI, monitor: _Monitor, path: Path) -> None:
def _scanned_output(api: PromptManagerAPI, recorder: _ScanRecorder, path: Path) -> None:
_png(path, (150, 60, 40))
monitor.new_output(path)
recorder.new_output(path)
def _linked_after_queueing(api: PromptManagerAPI, path: Path) -> None:
@@ -205,7 +213,7 @@ def seed_review(
root: Path,
items: int,
special: tuple = (),
monitor: _Monitor | None = None,
recorder: _ScanRecorder | None = None,
concept: bool = False,
) -> dict:
"""Replace this server's disposable rows and files with ``items`` queued files."""
@@ -258,9 +266,9 @@ def seed_review(
def add_review_file(
api: PromptManagerAPI, root: Path, kind: str, monitor: _Monitor
api: PromptManagerAPI, root: Path, kind: str, recorder: _ScanRecorder
) -> dict:
"""One more ambiguous output: through the monitor, or only written."""
"""One more ambiguous output: recorded as a scan does, or only written."""
output = root / "output"
number = (
len(list((output / "added").glob("*.png")))
@@ -268,8 +276,8 @@ def add_review_file(
else 0
)
relative = f"added/added_{number:04}.png"
if kind == "monitor":
_monitored_output(api, monitor, output / relative)
if kind == "scan":
_scanned_output(api, recorder, output / relative)
else:
_png(output / relative, (200, 200, 40))
api.invalidate_gallery_cache()
@@ -298,12 +306,12 @@ def register_review_commands(
routes: web.RouteTableDef, api: PromptManagerAPI, root: Path
) -> None:
"""Register the review fixture's commands; no path or SQL comes from a request."""
monitors = []
recorders = []
def monitor() -> _Monitor:
if not monitors:
monitors.append(_Monitor(api))
return monitors[0]
def recorder() -> _ScanRecorder:
if not recorders:
recorders.append(_ScanRecorder(api))
return recorders[0]
@routes.post("/__test__/review/reset")
async def reset(request: web.Request) -> web.Response:
@@ -331,7 +339,7 @@ def register_review_commands(
root,
items,
special,
monitor(),
recorder(),
data.get("concept", False),
)
return web.json_response({"success": True, **manifest})
@@ -341,10 +349,10 @@ def register_review_commands(
data = await request.json() if request.can_read_body else None
if request.query or not isinstance(data, dict) or set(data) != {"kind"}:
raise web.HTTPBadRequest()
if data["kind"] not in ("monitor", "file"):
if data["kind"] not in ("scan", "file"):
raise web.HTTPBadRequest()
added = await api._run_in_executor(
add_review_file, api, root, data["kind"], monitor()
add_review_file, api, root, data["kind"], recorder()
)
return web.json_response({"success": True, **added})
tests/test_api_admin.py
+6 -1
View File
@@ -2243,7 +2243,12 @@ class TestActivityJobs(AdminAPITestCase):
(self.output_dir / name).write_bytes(b"not an image")
def fake_ingest(
media_file, metadata, detailed=False, ambiguous=None, untraced=None
media_file,
metadata,
detailed=False,
ambiguous=None,
untraced=None,
records=None,
):
key = Path(media_file).relative_to(self.output_dir).as_posix()
outcome = names[key]
tests/test_api_review_events.py
+9 -3
View File
@@ -12,6 +12,7 @@ import asyncio
import json
import sqlite3
import threading
from pathlib import Path
try:
from tests.test_api_review import ReviewRoutesTestCase
@@ -86,11 +87,16 @@ class ReviewEventsTestCase(ReviewRoutesTestCase):
return path
async def _record_in_thread(self, path):
"""Record as the monitor does: from a worker thread, not the loop."""
"""Record as the scan does: from a worker thread, not the loop."""
from py.api.admin import _scan_roots
self._roots()
roots = _scan_roots(self.api._get_all_output_dirs())
outcomes = []
thread = threading.Thread(
target=lambda: outcomes.append(self.api.record_review_file(str(path)))
target=lambda: outcomes.append(
self.api._record_review_sync(Path(str(path)), roots, inspection=True)
)
)
thread.start()
await asyncio.get_running_loop().run_in_executor(None, thread.join)
@@ -98,7 +104,7 @@ class ReviewEventsTestCase(ReviewRoutesTestCase):
class TestReviewEvents(ReviewEventsTestCase):
async def test_monitor_recording_sends_one_change(self):
async def test_scan_recording_sends_one_change(self):
self._require_events()
stream = await self._stream()
self.assertEqual(await self._record_in_thread(self._queueable()), "recorded")
tests/test_review.py
+29 -18
View File
@@ -171,6 +171,13 @@ class ReviewTestCase(AdminAPITestCase):
def _require_module(self):
self.assertIsNotNone(review_capture, "utils.review_capture is not implemented")
def _scan_record(self, path):
"""Record ``path`` as the scan's recorder does (only scans add items)."""
from py.api.admin import _scan_roots
roots = _scan_roots(self.api._get_all_output_dirs())
return self.api._record_review_sync(Path(str(path)), roots, inspection=True)
class TestScanRecording(ReviewTestCase):
async def test_scan_persists_only_ambiguous_files_with_capture_digest(self):
@@ -502,13 +509,13 @@ class TestNonFiniteGraphs(ReviewTestCase):
["cat NaN", "dog Infinity"],
)
async def test_record_review_file_records_a_nan_stamped_graph(self):
async def test_scan_recorder_records_a_nan_stamped_graph(self):
self._require_module()
self._roots()
path = self.output_dir / "nan.png"
write_png(path, prompt=self._nan_stamped("cat", "dog"))
self.assertEqual(self.api.record_review_file(str(path)), "recorded")
self.assertEqual(self._scan_record(path), "recorded")
(item,) = self._items()
self.assertEqual(item["candidate_count"], 2)
@@ -856,12 +863,13 @@ class TestMonitorRecording(ReviewTestCase):
async def test_monitor_recording_is_refused_beside_record_changing_jobs(self):
"""S1 admission: a ``records`` recording waits for no job; it is refused.
While a ``records`` or ``guarded`` job runs, the monitor's recording
records nothing and answers ``deferred`` (its fallback link stands and
While a ``records`` or ``guarded`` job runs, the monitor's pass
writes nothing and answers ``deferred`` (its fallback link stands and
a later scan reads the ambiguous file again); a ``read`` job does not
refuse it, and it records once the blocking job has ended. Thumbnail
work (the job and the gallery's holds) is ``guarded`` but writes only
previews, never records, so it does not defer the recording (S5 M7).
refuse it. Thumbnail work (the job and the gallery's holds) is
``guarded`` but writes only previews, never records, so it does not
defer the pass either (S5 M7). Not deferred, the pass still records
nothing: only scans add items, so an unlinked output is ``skipped``.
"""
self._require_module()
from py.api.jobs import RESTORE_JOB_KIND, THUMBNAILS_JOB_KIND
@@ -898,11 +906,11 @@ class TestMonitorRecording(ReviewTestCase):
outcome = await asyncio.to_thread(
self.api.record_review_file, str(path)
)
self.assertEqual(outcome, "recorded")
self.assertEqual(len(self._items()), 1)
self.assertEqual(outcome, "skipped")
self.assertEqual(self._items(), [])
async def test_a_running_thumbnail_job_does_not_defer_monitor_recording(self):
"""S5 M7: a long rebuild never holds Needs review back (no records)."""
"""S5 M7: a long rebuild never defers the monitor's pass (no records)."""
self._require_module()
from py.api.jobs import THUMBNAILS_JOB_KIND
@@ -927,8 +935,9 @@ class TestMonitorRecording(ReviewTestCase):
finally:
gate.set()
await asyncio.wait_for(job.task, 5)
self.assertEqual(outcome, "recorded")
self.assertEqual(len(self._items()), 1)
# Not deferred (thumbnail work writes no records); only scans add items.
self.assertEqual(outcome, "skipped")
self.assertEqual(self._items(), [])
async def test_monitor_wiring_reaches_new_and_existing_handlers(self):
self._require_module()
@@ -996,12 +1005,12 @@ class TestRecordedSpelling(ReviewTestCase):
upper = self.output_dir / "Two.PNG"
write_png(upper, prompt=unlinkable_graph("cat", "dog"))
with _case_folding():
self.assertEqual(self.api.record_review_file(str(upper)), "recorded")
self.assertEqual(self._scan_record(upper), "recorded")
(before,) = self._items()
revision = self._review_revision()
lower = self.output_dir / "two.png"
os.rename(upper, lower) # keeps size and mtime_ns
self.assertEqual(self.api.record_review_file(str(lower)), "duplicate")
self.assertEqual(self._scan_record(lower), "duplicate")
(after,) = self._items()
self.assertEqual(self._spelled(), [str(lower)])
self.assertEqual(self._review_revision(), revision + 1)
@@ -1029,7 +1038,7 @@ class TestRecordedSpelling(ReviewTestCase):
self.assertEqual(capture.locator, after["locator"])
self.api._captures.release(capture.token)
# Seeing the same spelling again writes nothing.
self.assertEqual(self.api.record_review_file(str(lower)), "duplicate")
self.assertEqual(self._scan_record(lower), "duplicate")
self.assertEqual(self._review_revision(), revision + 1)
async def test_duplicate_insert_refreshes_the_spelling(self):
@@ -1070,7 +1079,8 @@ class TestRecordedSpelling(ReviewTestCase):
class TestCaptureStoreShutdown(ReviewTestCase):
async def test_recording_after_close_creates_no_store(self):
"""A monitor recording still in flight at shutdown leaks nothing (M2)."""
"""A scan recording still in flight at shutdown leaks nothing (M2),
and the monitor's pass after it captures nothing at all."""
self._require_module()
self._roots()
path = self.output_dir / "two.png"
@@ -1079,11 +1089,12 @@ class TestCaptureStoreShutdown(ReviewTestCase):
self.assertIsNone(self.api._captures)
with patch("utils.media_capture.tempfile.mkdtemp") as mkdtemp:
with self.assertRaises(CaptureError):
self.api.record_review_file(str(path))
# The monitor logs the failure and keeps its link.
self._scan_record(path)
# The monitor's pass records nothing, so it needs no store.
handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
handler.review_recorder = self.api.record_review_file
handler._record_review(str(path))
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
mkdtemp.assert_not_called()
self.assertIsNone(self.api._captures)
self.assertIsNone(self.api._captures_unsubscribe)
tests/test_review_choosable.py
+7 -3
View File
@@ -202,7 +202,7 @@ class TestNothingToChooseIsNotQueued(ChoosableTestCase):
self._roots()
path = self._write("none.png", opaque_graph("cat", "dog"))
self.assertEqual(self.api.record_review_file(str(path)), "untraced")
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
self.assertEqual(self._items(), [])
@@ -320,14 +320,18 @@ class TestUntracedItemsLeaveTheQueue(ChoosableTestCase):
await self._scan("all")
self.assertEqual(self._pending(), 0)
def test_monitor_removes_the_pending_item_of_an_untraced_output(self):
async def test_the_scan_not_the_monitor_removes_an_untraced_outputs_item(self):
"""The monitor only clears items of outputs it links; the next scan
removes the item of an output that offers nothing to choose."""
self._roots()
path = self._write("none.png", opaque_graph("cat", "dog"))
self._record_pre_fix("none.png")
# A pre-fix item of other bytes keeps the stat shortcut out of the way.
self._write("none.png", opaque_graph("a cow", "dog"))
self.assertEqual(self.api.record_review_file(str(path)), "untraced")
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
self.assertEqual(len(self._items()), 1)
await self._scan("all")
self.assertEqual(self._items(), [])
tests/test_review_inspection.py
+971
View File
@@ -0,0 +1,971 @@
"""Needs review choices from the inspection profile (follow-up to #284/#288).
A Qwen image-edit batch runs several KSamplers, each with its own
``TextEncodeQwenImageEdit(prompt=...)``, and every SaveImage shares one
prefix: the file cannot say which prompt made it. The scan's default
(ingestion) profile does not know the encoder or the string nodes, so every
candidate is unresolved and the file used to leave the queue as "could not
be traced". The inspection profile resolves every text, so these are real
choices. Ruling (option a): when the default profile is ambiguous with fewer
than two candidates that carry text, the scan asks the inspection profile;
with at least two choosable candidates it records the item from that
profile, and Apply re-resolves with the same profile. Nothing links
automatically from the inspection profile, and the live monitor never queues.
"""
import contextlib
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
try:
from tests.test_api_review import ReviewRoutesTestCase
from tests.test_review import unlinkable_graph, write_png
except ImportError: # discovered with tests/ as the top-level directory
from test_api_review import ReviewRoutesTestCase
from test_review import unlinkable_graph, write_png
from utils import inspection_semantics, review_capture
from utils.hashing import generate_prompt_hash
from utils.image_monitor import ImageGenerationHandler
from utils.prompt_tracker import PromptTracker
NAME = "kittybear_00001_.png"
PROMPTS = ("make it night", "add a red hat", "turn the cat into a bear")
@contextlib.contextmanager
def _counted(module, name):
"""Count the calls of ``module.name`` while calling through."""
real = getattr(module, name)
calls = []
def spy(*args, **kwargs):
calls.append(args)
return real(*args, **kwargs)
with patch.object(module, name, side_effect=spy):
yield calls
def qwen_edit_graph(*prompts, joined=()):
"""One Qwen image-edit sampler per prompt, all saved under one prefix.
Each sampler has its own ``TextEncodeQwenImageEdit``, used for both
positive and negative as the real batches do. ``joined`` prompts are
``(first, second)`` pairs joined by ``JoinStringMulti`` over two
``StringConstantMultiline`` nodes. Every SaveImage takes its prefix from
one linked string node.
"""
graph = {
"1": {"class_type": "UNETLoader", "inputs": {"unet_name": "q.safetensors"}},
"2": {"class_type": "CLIPLoader", "inputs": {"clip_name": "c.safetensors"}},
"3": {"class_type": "VAELoader", "inputs": {"vae_name": "v.safetensors"}},
"4": {"class_type": "LoadImage", "inputs": {"image": "source.png"}},
"5": {"class_type": "PrimitiveString", "inputs": {"value": "kittybear"}},
"6": {
"class_type": "EmptyLatentImage",
"inputs": {"width": 512, "height": 512, "batch_size": 1},
},
}
sides = [("literal", text) for text in prompts]
sides += [("joined", parts) for parts in joined]
for index, (kind, text) in enumerate(sides):
base = 100 + 10 * index
prompt = text
if kind == "joined":
for offset, part in enumerate(text, start=1):
graph[str(base + offset)] = {
"class_type": "StringConstantMultiline",
"inputs": {"string": part, "strip_newlines": True},
}
graph[str(base + 3)] = {
"class_type": "JoinStringMulti",
"inputs": {
"inputcount": 2,
"delimiter": ", ",
"return_list": False,
"string_1": [str(base + 1), 0],
"string_2": [str(base + 2), 0],
},
}
prompt = [str(base + 3), 0]
encoder, sampler = str(base), str(base + 5)
graph[encoder] = {
"class_type": "TextEncodeQwenImageEdit",
"inputs": {
"clip": ["2", 0],
"prompt": prompt,
"vae": ["3", 0],
"image": ["4", 0],
},
}
graph[sampler] = {
"class_type": "KSampler",
"inputs": {
"model": ["1", 0],
"positive": [encoder, 0],
"negative": [encoder, 0],
"latent_image": ["6", 0],
"seed": index,
},
}
graph[str(base + 6)] = {
"class_type": "VAEDecode",
"inputs": {"samples": [sampler, 0], "vae": ["3", 0]},
}
graph[str(base + 7)] = {
"class_type": "SaveImage",
"inputs": {"images": [str(base + 6), 0], "filename_prefix": ["5", 0]},
}
return graph
def _prefixed(graph, prefix):
"""``graph`` with every node id (and link to one) prefixed."""
renamed = {}
for node_id, node in graph.items():
inputs = {
key: (
[prefix + value[0], value[1]]
if isinstance(value, list) and len(value) == 2
else value
)
for key, value in node["inputs"].items()
}
renamed[prefix + node_id] = {**node, "inputs": inputs}
return renamed
def _version(graph=None):
"""The item version of a Qwen-edit graph under the current table."""
graph = graph if graph is not None else qwen_edit_graph(*PROMPTS)
return review_capture.inspection_review_version({"prompt": graph})
def _retarget_unet_loader(doc):
"""A table edit for a class the Qwen-edit graph uses (no prompt effect)."""
doc["facts"]["model_outputs"]["UNETLoader"] = "unet"
def _unrelated_encoder(doc):
"""A table edit for a class the Qwen-edit graph does not use."""
doc["encoders"].append(
{
"class": "UnrelatedTextEncode",
"slot": 0,
"strategy": "single",
"inputs": ["text"],
}
)
def _texts(evidence):
return sorted(
c["positive"]["text"]
for c in evidence["candidates"]
if isinstance(c.get("positive"), dict)
)
class InspectionTestCase(ReviewRoutesTestCase):
def _write(self, name=NAME, graph=None, color=(255, 0, 0)):
path = self.output_dir / name
write_png(
path,
prompt=graph if graph is not None else qwen_edit_graph(*PROMPTS),
color=color,
)
return path
def _pending(self):
return self.api.db.library_counts()["review_pending"]
def _evidence(self, item):
return json.loads(item["evidence"])
def _table(self, edit):
"""Point the inspection profile at an edited copy of the table."""
doc = json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8"))
edit(doc)
folder = tempfile.mkdtemp(dir=self.tmpdir)
table = Path(folder) / "inspection-semantics.json"
table.write_text(json.dumps(doc), encoding="utf-8")
patcher = patch.object(inspection_semantics, "SEMANTICS_PATH", table)
patcher.start()
inspection_semantics.get_semantics.cache_clear()
self.addCleanup(inspection_semantics.get_semantics.cache_clear)
self.addCleanup(patcher.stop)
async def _apply(self, positive):
item, detail, observed = await self._open()
candidate = self._candidate(detail, positive)
return item, await self._decide(
item["id"], self._body(detail, observed, candidate["hash"])
)
def _linked_text(self, path):
row = (
self.api.db.model.get_connection()
.execute(
"SELECT p.text FROM generated_images gi "
"JOIN prompts p ON p.id=gi.prompt_id WHERE gi.image_path=?",
(str(path),),
)
.fetchone()
)
return row[0] if row else None
class TestQwenEditBatchIsQueued(InspectionTestCase):
async def test_qwen_edit_batch_is_queued_with_its_texts(self):
self._roots()
graph = qwen_edit_graph(*PROMPTS, joined=[("a red scarf", "snow")])
self._write(graph=graph)
job = await self._scan()
(item,) = self._items()
evidence = self._evidence(item)
self.assertEqual(_texts(evidence), sorted(PROMPTS + ("a red scarf, snow",)))
self.assertEqual(item["candidate_count"], 4)
self.assertEqual(evidence["profile"], "inspection")
self.assertEqual(
evidence["semantics"], review_capture.semantics_version({"prompt": graph})
)
self.assertEqual(
item["resolver_version"],
review_capture.inspection_review_version({"prompt": graph}),
)
self.assertNotEqual(item["resolver_version"], review_capture.RESOLVER_VERSION)
self.assertEqual(job["scope"]["review_items"], 1)
(row,) = job["outcomes"]["review"]
self.assertEqual(row["path"], NAME)
self.assertIn("more than one prompt could match", row["detail"])
# Review choices only: nothing linked the file.
self.assertIsNone(self._linked_text(self.output_dir / NAME))
self.assertEqual(self._capture_files(), [])
async def test_rescan_keeps_one_item_and_skips_the_copy(self):
self._roots()
self._write()
await self._scan()
(item,) = self._items()
with _counted(self.api._captures, "capture") as copies:
job = await self._scan("all")
self.assertEqual(copies, [])
self.assertEqual([i["id"] for i in self._items()], [item["id"]])
self.assertEqual(job["scope"]["review_items"], 1)
self.assertEqual(job["scope"]["review_untraced"], 0)
async def test_more_than_sixteen_qwen_samplers_make_a_limited_item(self):
self._roots()
self._write(graph=qwen_edit_graph(*(f"edit {n}" for n in range(17))))
await self._scan()
(item,) = self._items()
evidence = self._evidence(item)
self.assertTrue(item["limited"])
self.assertEqual(item["candidate_count"], 0)
self.assertEqual(evidence["candidate_total"], 17)
self.assertEqual(evidence["profile"], "inspection")
async def test_evidence_over_the_cap_makes_a_limited_item(self):
self._roots()
self._write()
real = review_capture.canonical_evidence
def capped(value):
if value.get("candidates"):
raise ValueError("evidence exceeds the bound")
return real(value)
with patch.object(review_capture, "canonical_evidence", side_effect=capped):
await self._scan()
(item,) = self._items()
self.assertTrue(item["limited"])
self.assertEqual(self._evidence(item)["candidate_total"], 3)
self.assertEqual(self._evidence(item)["profile"], "inspection")
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."""
self._roots()
graph = {
**_prefixed(qwen_edit_graph("make it night"), "q"),
**unlinkable_graph("cat", "dog"),
}
self._write(graph=graph)
with patch.object(
review_capture,
"resolve_inspection",
side_effect=AssertionError("inspection profile ran"),
):
await self._scan()
(item,) = self._items()
evidence = self._evidence(item)
self.assertNotIn("profile", evidence)
self.assertNotIn("semantics", evidence)
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):
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"),
):
_item, (status, data) = await self._apply("cat")
self.assertEqual(status, 200, data)
self.assertEqual(self._linked_text(self.output_dir / "two.png"), "cat")
class TestApplyInspectionCandidate(InspectionTestCase):
async def test_apply_links_the_chosen_qwen_edit_text(self):
self._require_routes()
self._roots()
path = self._write()
await self._scan()
item, (status, data) = await self._apply("add a red hat")
self.assertEqual(status, 200, data)
self.assertEqual(data["item"]["state"], "selected")
self.assertEqual(data["total"], 0)
self.assertEqual(self._linked_text(path), "add a red hat")
links = self.api.db.get_image_prompt_pairs(data["pair"]["image_id"])
(link,) = links
provenance = link["provenance"]
self.assertEqual(provenance["kind"], "user-selection")
self.assertEqual(provenance["resolver_version"], _version())
self.assertEqual(provenance["positive"]["text"], "add a red hat")
(decision,) = self._decisions()
self.assertEqual(decision["item_id"], item["id"])
self._no_paths()
async def test_apply_rechecks_with_the_recorded_profile(self):
"""The default profile would not offer the candidate: Apply must use
the profile the item recorded, and refuses when that profile no
longer offers it."""
self._require_routes()
self._roots()
self._write()
await self._scan()
calls = []
real = review_capture.resolve_inspection
def spy(metadata):
calls.append(metadata)
return real(metadata)
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)
async def test_table_change_keeps_the_item_applicable(self):
"""A table change for a class the graph uses moves the item's
version; the pending item stays applicable while its candidate is
still offered."""
self._require_routes()
self._roots()
path = self._write()
await self._scan()
(old,) = self._items()
version = _version()
item, detail, observed = await self._open()
self._table(_retarget_unet_loader)
self.assertNotEqual(_version(), version)
candidate = self._candidate(detail, "add a red hat")
status, data = await self._decide(
item["id"], self._body(detail, observed, candidate["hash"])
)
self.assertEqual(status, 200, data)
self.assertEqual(item["id"], old["id"])
self.assertEqual(self._linked_text(path), "add a red hat")
async def test_table_change_that_drops_the_encoder_is_a_review_conflict(self):
self._require_routes()
self._roots()
self._write()
await self._scan()
item, detail, observed = await self._open()
candidate = self._candidate(detail, "add a red hat")
def drop_encoder(doc):
doc["encoders"] = [
entry
for entry in doc["encoders"]
if entry["class"] != "TextEncodeQwenImageEdit"
]
version = _version()
self._table(drop_encoder)
self.assertNotEqual(_version(), version)
before = self._snapshot()
status, data = await self._decide(
item["id"], self._body(detail, observed, candidate["hash"])
)
self.assertEqual(status, 409, data)
self.assertEqual(data["code"], "review-conflict")
self.assertEqual(self._snapshot(), before)
# The rescan finds nothing to choose under the new version: the
# stale, undecided item leaves the queue instead of staying stuck.
job = await self._scan("all")
self.assertEqual(self._items(), [])
self.assertEqual(job["scope"]["review_untraced"], 1)
class TestMonitorNeverQueues(InspectionTestCase):
def _handler(self):
handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
handler.review_recorder = self.api.record_review_file
return handler
async def test_monitor_pass_never_queues_but_the_next_scan_does(self):
self._roots()
path = self._write()
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
self.assertEqual(self._items(), [])
self.assertEqual(self._capture_files(), [])
await self._scan()
(item,) = self._items()
self.assertEqual(self._evidence(item)["profile"], "inspection")
async def test_monitor_keeps_the_inspection_item_of_an_unlinked_output(self):
"""The monitor's untraced settle judges each item by its own
evidence: an inspection item offers a choice and stays."""
self._roots()
path = self._write()
await self._scan()
(item,) = self._items()
# Other bytes, so the stat shortcut does not answer for the file.
self._write(color=(0, 0, 255))
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
self.assertEqual([i["id"] for i in self._items()], [item["id"]])
async def test_monitor_link_removes_the_inspection_item(self):
self._roots()
path = self._write()
await self._scan()
self.assertEqual(len(self._items()), 1)
prompt_id = self.api.db.save_prompt(
text="latest", prompt_hash=generate_prompt_hash("latest")
)
self._handler().link_image_to_prompt(
str(path),
{"id": prompt_id, "text": "latest", "fallback": True},
{"prompt": qwen_edit_graph(*PROMPTS)},
)
self.assertEqual(self._items(), [])
self.assertEqual(os.path.basename(str(path)), NAME)
class TestTableVersion(InspectionTestCase):
def test_table_version_follows_content_not_layout(self):
doc = {"schema": 1, "encoders": [{"class": "A", "slot": 0}]}
same = json.loads(json.dumps(doc, indent=4))
other = {"schema": 1, "encoders": [{"class": "B", "slot": 0}]}
version = inspection_semantics.table_version(doc)
self.assertRegex(version, "^[0-9a-f]{64}$")
self.assertEqual(inspection_semantics.table_version(same), version)
self.assertNotEqual(inspection_semantics.table_version(other), version)
self.assertEqual(
inspection_semantics.load_semantics().version,
inspection_semantics.table_version(
json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8"))
),
)
async def test_an_unusable_table_queues_nothing(self):
self._roots()
self._write()
unusable = inspection_semantics.SemanticsError("bad table")
with (
patch.object(inspection_semantics, "get_semantics", side_effect=unusable),
patch.object(review_capture, "get_semantics", side_effect=unusable),
):
self.assertIsNone(review_capture.inspection_review_version())
self.assertIsNone(_version())
job = await self._scan()
self.assertEqual(self._items(), [])
(row,) = job["outcomes"]["review"]
self.assertIn("could not be traced", row["detail"])
async def test_an_unusable_table_keeps_existing_items_on_rescan(self):
self._require_routes()
self._roots()
self._write()
await self._scan()
(item,) = self._items()
unusable = inspection_semantics.SemanticsError("bad table")
with (
patch.object(inspection_semantics, "get_semantics", side_effect=unusable),
patch.object(review_capture, "get_semantics", side_effect=unusable),
):
await self._scan("all")
self.assertEqual([i["id"] for i in self._items()], [item["id"]])
class TestVersionChanges(InspectionTestCase):
"""A version change never leaves two pending items for one file."""
async def test_table_change_for_a_graph_class_leaves_one_item(self):
self._require_routes()
self._roots()
path = self._write()
await self._scan()
(old,) = self._items()
self._table(_retarget_unet_loader)
job = await self._scan("all")
(item,) = self._items()
self.assertNotEqual(item["id"], old["id"])
self.assertEqual(item["resolver_version"], _version())
self.assertNotEqual(item["resolver_version"], old["resolver_version"])
self.assertEqual(job["scope"]["review_items"], 1)
self.assertEqual(self._pending(), 1)
_item, (status, data) = await self._apply("make it night")
self.assertEqual(status, 200, data)
self.assertEqual(self._linked_text(path), "make it night")
async def test_unrelated_table_edit_keeps_the_item_and_does_no_work(self):
self._roots()
self._write()
await self._scan()
(old,) = self._items()
version = _version()
self._table(_unrelated_encoder)
self.assertEqual(_version(), version)
self.assertNotEqual(
inspection_semantics.get_semantics().version,
json.loads(old["evidence"])["semantics"],
)
with (
_counted(self.api._captures, "capture") as copies,
_counted(review_capture, "resolve_inspection") as calls,
):
await self._scan("all")
self.assertEqual(copies, [])
self.assertEqual([i["id"] for i in self._items()], [old["id"]])
self.assertEqual(calls, [])
async def test_resolver_change_leaves_one_default_item(self):
self._roots()
write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
await self._scan()
(old,) = self._items()
self.assertEqual(old["resolver_version"], 1)
with patch.object(review_capture, "RESOLVER_VERSION", 2):
await self._scan("all")
(item,) = self._items()
self.assertNotEqual(item["id"], old["id"])
self.assertEqual(item["resolver_version"], 2)
self.assertEqual(self._pending(), 1)
async def test_decided_old_item_is_kept_and_no_duplicate_is_made(self):
self._require_routes()
self._roots()
self._write()
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())
self._table(_retarget_unet_loader)
await self._scan("all")
self.assertEqual((self._items(), self._decisions()), before)
# The decided bytes are not copied again on the next rescan.
with _counted(self.api._captures, "capture") as copies:
await self._scan("all")
self.assertEqual(copies, [])
self.assertEqual((self._items(), self._decisions()), before)
self.assertEqual(self._pending(), 0)
async def test_advanced_pending_item_is_kept_and_no_duplicate_is_made(self):
self._roots()
self._write()
await self._scan()
(old,) = self._items()
conn = self.api.db.model.get_connection()
with conn:
conn.execute(
"UPDATE media_review_items SET revision=1 WHERE id=?", (old["id"],)
)
before = self._items()
self._table(_retarget_unet_loader)
await self._scan("all")
self.assertEqual(self._items(), before)
class TestInspectionRuns(InspectionTestCase):
"""The heavier profile runs once per new file state, never again."""
async def test_new_file_is_inspected_once_and_not_on_rescan(self):
self._roots()
self._write()
with _counted(review_capture, "resolve_inspection") as first:
await self._scan()
self.assertEqual(len(self._items()), 1)
self.assertEqual(len(first), 1)
with _counted(review_capture, "resolve_inspection") as again:
await self._scan("all")
self.assertEqual(again, [])
self.assertEqual(len(self._items()), 1)
async def test_untraced_file_is_not_inspected_again_until_it_changes(self):
self._roots()
graph = qwen_edit_graph(*PROMPTS)
for node in graph.values():
if node["class_type"] == "TextEncodeQwenImageEdit":
node["class_type"] = "OpaqueTextEncode"
self._write(graph=graph)
with _counted(review_capture, "resolve_inspection") as first:
job = await self._scan()
self.assertEqual(len(first), 1)
(row,) = job["outcomes"]["review"]
self.assertIn("could not be traced", row["detail"])
with _counted(review_capture, "resolve_inspection") as again:
await self._scan("all")
self.assertEqual(again, [])
self._write(graph=graph, color=(0, 255, 0))
with _counted(review_capture, "resolve_inspection") as changed:
await self._scan("all")
self.assertEqual(len(changed), 1)
self.assertEqual(self._items(), [])
async def test_capture_with_other_records_is_inspected_again(self):
"""The scan's result is reused only for identical records."""
self._roots()
self._write()
real = review_capture.captured_review
def edited(store, capture, profile=review_capture.DEFAULT):
metadata, resolution = real(store, capture, profile)
graph = json.loads(json.dumps(metadata["prompt"]))
graph["100"]["inputs"]["prompt"] = "a captured edit"
return {**metadata, "prompt": graph}, resolution
with (
patch.object(review_capture, "captured_review", side_effect=edited),
_counted(review_capture, "resolve_inspection") as calls,
):
await self._scan()
self.assertEqual(len(calls), 2)
(item,) = self._items()
self.assertIn("a captured edit", _texts(self._evidence(item)))
class TestClassProjection(unittest.TestCase):
def test_projection_over_every_named_class_is_the_whole_table(self):
doc = json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8"))
names = set(inspection_semantics.load_semantics().all_classes)
names |= set(doc["facts"]["configuration_outputs"])
names |= set(doc["facts"]["model_outputs"])
names |= set(doc["facts"]["model_passthroughs"])
names |= set(doc["facts"].get("scalar_outputs", {}))
names |= {entry["class"] for entry in doc["lineage"]}
self.assertEqual(inspection_semantics.class_projection(doc, names), doc)
def test_class_version_moves_only_for_the_classes_it_covers(self):
doc = json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8"))
classes = frozenset({"KSampler", "UNETLoader"})
version = inspection_semantics.class_version(doc, classes)
_unrelated_encoder(doc)
self.assertEqual(inspection_semantics.class_version(doc, classes), version)
_retarget_unet_loader(doc)
self.assertNotEqual(inspection_semantics.class_version(doc, classes), version)
class TestRecordingEdges(InspectionTestCase):
async def test_store_refuses_a_new_version_beside_a_decided_item(self):
self._roots()
self._write()
await self._scan()
(old,) = self._items()
conn = self.api.db.model.get_connection()
with conn:
conn.execute(
"UPDATE media_review_items SET state='unresolved', revision=1 "
"WHERE id=?",
(old["id"],),
)
evidence = {**self._evidence(old), "resolver_version": 7}
inserted = self.api.db.record_review_item(
locator=old["locator"],
spelled_path=old["spelled_path"],
sha256=old["sha256"],
size=old["size"],
mtime_ns=old["mtime_ns"],
epoch=self.api.db.get_execution_epoch(),
resolver_version=7,
evidence=evidence,
candidate_count=old["candidate_count"],
limited=bool(old["limited"]),
)
self.assertFalse(inserted)
self.assertEqual([i["id"] for i in self._items()], [old["id"]])
async def test_already_linked_file_with_an_inspection_choice_is_unchanged(self):
self._roots()
path = self._write()
prompt_id = self.api.db.save_prompt(
text="latest", prompt_hash=generate_prompt_hash("latest")
)
handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
handler.link_image_to_prompt(
str(path),
{"id": prompt_id, "text": "latest", "fallback": True},
{"prompt": qwen_edit_graph(*PROMPTS)},
)
job = await self._scan("all")
self.assertEqual(self._items(), [])
self.assertEqual(job["outcomes"]["review"], [])
(row,) = job["outcomes"]["known"]
self.assertIn("Already linked", row["detail"])
self.assertEqual(self._linked_text(path), "latest")
async def test_untraced_memory_forgets_the_oldest_state(self):
from py.api import review_recording
self._roots()
graph = qwen_edit_graph(*PROMPTS)
for node in graph.values():
if node["class_type"] == "TextEncodeQwenImageEdit":
node["class_type"] = "OpaqueTextEncode"
self._write("a.png", graph=graph)
self._write("b.png", graph=graph)
with patch.object(review_recording, "MAX_UNTRACED_MEMORY", 1):
await self._scan()
with _counted(review_capture, "resolve_inspection") as calls:
await self._scan("all")
# One remembered state: each file evicts the other, so both run.
self.assertEqual(len(calls), 2)
self.assertEqual(len(self.api._untraced_inspections), 1)
def test_graph_classes(self):
self.assertEqual(review_capture.graph_classes({"parameters": "x"}), set())
self.assertEqual(
review_capture.graph_classes({"prompt": {"1": {"class_type": "A"}}}),
{"A"},
)
many = {
str(n): {"class_type": f"C{n}"}
for n in range(review_capture.MAX_FINGERPRINT_CLASSES + 1)
}
self.assertIsNone(review_capture.graph_classes({"prompt": many}))
self.assertEqual(
review_capture.semantics_version({"prompt": many}),
inspection_semantics.get_semantics().version,
)
def _move_qwen_encoder_slot(doc):
"""An edit of the Qwen-edit encoder's own table entry."""
for entry in doc["encoders"]:
if entry["class"] == "TextEncodeQwenImageEdit":
entry["slot"] = 1
def _unused_sampler(doc):
"""A new dict-keyed sampler class the Qwen-edit graph does not use."""
doc["samplers"]["UnusedSampler"] = dict(doc["samplers"]["KSampler"])
class TestFingerprintCoverage(InspectionTestCase):
def test_editing_the_encoders_entry_moves_the_version(self):
version = _version()
self._table(_move_qwen_encoder_slot)
self.assertIsNotNone(_version())
self.assertNotEqual(_version(), version)
def test_an_unused_dict_keyed_class_leaves_the_version(self):
version = _version()
table = inspection_semantics.get_semantics().version
self._table(_unused_sampler)
self.assertNotEqual(inspection_semantics.get_semantics().version, table)
self.assertEqual(_version(), version)
class TestInspectionResolverVersion(InspectionTestCase):
async def test_inspection_resolver_change_supersedes_the_item(self):
"""An inspection-only resolver change moves the item version, so the
rescan re-inspects and the new item is applicable (never a stuck
409)."""
self._require_routes()
self._roots()
path = self._write()
await self._scan()
(old,) = self._items()
version = _version()
real = review_capture.resolve_inspection
def newer(metadata):
resolution = real(metadata)
return {
**resolution,
"candidates": [
{**c, "reason": "a newer inspection resolver"}
for c in resolution["candidates"]
],
}
with (
patch.object(review_capture, "INSPECTION_RESOLVER_VERSION", 2),
patch.object(review_capture, "resolve_inspection", side_effect=newer),
):
self.assertNotEqual(_version(), version)
await self._scan("all")
(item,) = self._items()
self.assertNotEqual(item["id"], old["id"])
_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")
def test_profile_budgets_are_folded_into_the_version(self):
from dataclasses import replace
version = _version()
wider = replace(
review_capture.INSPECTION_PROFILE,
total_visits=review_capture.INSPECTION_PROFILE.total_visits + 1,
)
with patch.object(review_capture, "INSPECTION_PROFILE", wider):
self.assertNotEqual(_version(), version)
class TestScanContainment(InspectionTestCase):
async def test_an_inspection_failure_is_counted_and_the_scan_continues(self):
self._roots()
self._write("a.png")
self._write("b.png", graph=qwen_edit_graph("snow", "rain"))
real = review_capture.resolve_inspection
def fails_for_a(metadata):
if "make it night" in json.dumps(metadata):
raise RuntimeError("inspection blew up")
return real(metadata)
with (
patch.object(review_capture, "resolve_inspection", side_effect=fails_for_a),
self.assertLogs("prompt_manager.api", "WARNING") as logs,
):
job = await self._scan()
self.assertEqual(job["state"], "finished")
(item,) = self._items()
self.assertEqual(os.path.basename(item["locator"]), "b.png")
self.assertEqual(job["scope"]["review_errors"], 1)
rows = {row["path"]: row for row in job["outcomes"]["review"]}
self.assertIn("could not be traced", rows["a.png"]["detail"])
self.assertIn("more than one prompt could match", rows["b.png"]["detail"])
self.assertTrue(any("a.png" in line for line in logs.output))
class TestUntracedMemoryKeys(InspectionTestCase):
async def test_untraced_memory_is_keyed_by_a_16_byte_digest(self):
self._roots()
graph = qwen_edit_graph(*PROMPTS)
for node in graph.values():
if node["class_type"] == "TextEncodeQwenImageEdit":
node["class_type"] = "OpaqueTextEncode"
self._write(graph=graph)
await self._scan()
(key,) = self.api._untraced_inspections
self.assertIsInstance(key, bytes)
self.assertEqual(len(key), 16)
class TestMonitorNeverRecords(InspectionTestCase):
async def test_monitor_never_records_an_unlinked_ambiguous_output(self):
"""Only scans add items (maintainer ruling); the monitor only clears
the items of outputs it links."""
self._roots()
path = self.output_dir / "two.png"
write_png(path, prompt=unlinkable_graph("cat", "dog"))
with _counted(self.api._captures, "capture") as copies:
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
self.assertEqual(copies, [])
self.assertEqual(self._items(), [])
await self._scan()
self.assertEqual(len(self._items()), 1)
class TestStaleSettleContainment(InspectionTestCase):
async def test_a_failed_stale_settle_is_logged_and_counted(self):
import sqlite3
self._roots()
graph = qwen_edit_graph(*PROMPTS)
for node in graph.values():
if node["class_type"] == "TextEncodeQwenImageEdit":
node["class_type"] = "OpaqueTextEncode"
self._write(graph=graph)
with (
patch.object(
self.api.db,
"settle_stale_review_items",
side_effect=sqlite3.OperationalError("database is locked"),
),
self.assertLogs("prompt_manager.api", "WARNING") as logs,
):
job = await self._scan()
self.assertEqual(job["state"], "finished")
self.assertEqual(job["scope"]["review_errors"], 1)
self.assertTrue(any("Stale Needs review" in line for line in logs.output))
tests/test_review_outcome.py
+5 -3
View File
@@ -355,7 +355,9 @@ class TestMonitorRule(OutcomeTestCase):
self.assertEqual((self._items(), self._decisions()), before)
self.assertEqual(self._pending(), 0)
async def test_record_review_file_applies_the_scan_rule(self):
async def test_record_review_file_only_clears_what_it_links(self):
"""Only scans add items: the monitor's pass skips every unlinked file
and settles the items of a linked one."""
self._roots()
write_png(self.output_dir / "fallback.png", prompt=ambiguous_graph("a", "b"))
write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
@@ -366,8 +368,8 @@ class TestMonitorRule(OutcomeTestCase):
record = self.api.record_review_file
self.assertEqual(record(str(self.output_dir / "fallback.png")), "skipped")
self.assertEqual(record(str(linked)), "linked")
self.assertEqual(record(str(self.output_dir / "two.png")), "recorded")
self.assertEqual(len(self._items()), 1)
self.assertEqual(record(str(self.output_dir / "two.png")), "skipped")
self.assertEqual(self._items(), [])
if __name__ == "__main__":
utils/inspection_semantics.py
+81 -2
View File
@@ -16,10 +16,21 @@ nesting before it is validated. Every class appears in exactly one prompt
category; classes that are not listed stay ``unsupported-node``. ``lineage``
(image and latent pass-through outputs for multi-stage chains) describes no
prompt, so its classes are not claimed by a category.
``Semantics.version`` is the SHA-256 of the table's canonical JSON (sorted
keys, no whitespace): any change of content, never of layout, moves it.
``class_version(doc, classes)`` is the narrower fingerprint Needs review
folds into the version of the items it records from the inspection profile
(``utils.review_capture``): the SHA-256 of ``class_projection``, the table
restricted to what it says about ``classes`` (the node classes of one graph)
plus the class-independent ``schema`` and ``stage_inputs``. The resolver looks
the table up only by the class of a graph node, so an edit about other
classes cannot change how that graph resolves and leaves the version alone.
"""
import hashlib
import json
from dataclasses import dataclass
from dataclasses import dataclass, replace
from functools import lru_cache
from pathlib import Path
from types import MappingProxyType
@@ -118,6 +129,10 @@ class Semantics:
stage_inputs: Mapping[str, Tuple[str, ...]]
facts: Mapping[str, Any]
all_classes: FrozenSet[str]
# SHA-256 of the canonical table JSON (set by ``load_semantics``).
version: str = ""
# The canonical table JSON itself, for ``class_version``.
source: str = ""
def _freeze(value: Any) -> Frozen:
@@ -521,7 +536,71 @@ def load_semantics(path: Optional[Union[str, Path]] = None) -> Semantics:
except (OSError, UnicodeDecodeError, ValueError) as error:
raise SemanticsError(f"cannot read semantics table {source.name}: {error}")
_check_nesting(doc)
return _validate(doc)
return replace(
_validate(doc), version=table_version(doc), source=_canonical_json(doc)
)
# Table parts that hold no class names: they apply to every graph.
_CLASS_INDEPENDENT = frozenset({"schema", "stage_inputs"})
def _canonical_json(doc: Any) -> str:
return json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
def table_version(doc: Any) -> str:
"""SHA-256 of ``doc`` as canonical JSON: content, not layout."""
return hashlib.sha256(_canonical_json(doc).encode("utf-8")).hexdigest()
def _about(value: Any, classes: FrozenSet[str]) -> Any:
"""The part of one category that names a class in ``classes``.
Categories are keyed by class (objects), list classes (strings) or list
entries that carry a ``class``.
"""
if isinstance(value, dict):
return {key: item for key, item in value.items() if key in classes}
if isinstance(value, list):
return [
entry
for entry in value
if (isinstance(entry, str) and entry in classes)
or (isinstance(entry, dict) and entry.get("class") in classes)
]
return value
def class_projection(doc: Mapping[str, Any], classes) -> dict:
"""What ``doc`` says about ``classes``, plus its class-independent parts."""
classes = frozenset(classes)
projected = {}
for category, value in doc.items():
if category in _CLASS_INDEPENDENT:
projected[category] = value
elif category == "facts" and isinstance(value, dict):
projected[category] = {
table: _about(entries, classes) for table, entries in value.items()
}
else:
projected[category] = _about(value, classes)
return projected
def class_version(doc: Mapping[str, Any], classes) -> str:
"""SHA-256 of ``class_projection(doc, classes)`` as canonical JSON."""
return table_version(class_projection(doc, classes))
@lru_cache(maxsize=256)
def _cached_class_version(source: str, classes: FrozenSet[str]) -> str:
return class_version(json.loads(source), classes)
def semantics_class_version(semantics: Semantics, classes) -> str:
"""``class_version`` of a loaded table (cached per table and class set)."""
return _cached_class_version(semantics.source, frozenset(classes))
@lru_cache(maxsize=1)
utils/prompt_pair_resolver.py
+5
View File
@@ -44,6 +44,11 @@ _LOGGER = logging.getLogger("prompt_manager.prompt_pair_resolver")
# otherwise; a pending item stays applicable while its chosen candidate is
# still offered by the captured bytes.
RESOLVER_VERSION = 1
# The inspection profile's own version, folded into the version of Needs
# review items recorded from that profile (with the profile's budgets and
# the semantics table). Bump it when a change would make the inspection
# profile resolve the same metadata otherwise, even if ingestion is unchanged.
INSPECTION_RESOLVER_VERSION = 1
@dataclass(frozen=True)
utils/review_capture.py
+230 -24
View File
@@ -1,7 +1,8 @@
"""Capture-bound Needs review items for ambiguous files (#249 E2).
The output scan and the live monitor hand every file the ingestion resolver
finds ``ambiguous`` to :func:`record_ambiguous_file`. It captures the file
The output scan hands every unlinked file the ingestion resolver finds
``ambiguous`` to :func:`record_ambiguous_file` (only scans add items; the
live monitor records nothing). It captures the file
(the bounded private copy of ``utils.media_capture``, hashed while copied),
reads the metadata again from the capture, resolves it with the default
(ingestion) profile and stores only an ``ambiguous`` result, bound to the
@@ -21,6 +22,24 @@ candidate's ``hash`` is the SHA-256 of its canonical JSON without the hash.
More than ``MAX_CANDIDATES`` candidates or evidence over the 8 MiB bound make
a ``limited`` item without candidates (a result without any candidate is not
``reviewable`` and is never stored).
Inspection-profile choices (ruling a, after the maintainer's Qwen image-edit
batches): when the default profile is ambiguous without a choice, the scan
(``inspection=True``) asks ``INSPECTION_PROFILE``,
which knows more encoder and string nodes. With at least two choosable
candidates there, the item is recorded from that profile: its evidence names
``profile: "inspection"`` and ``semantics``, the fingerprint of what the table
says about the graph's node classes, and its ``resolver_version`` is
``inspection_review_version(metadata)``, which folds the profile,
``RESOLVER_VERSION`` and that fingerprint together. A table edit about those
classes moves the version the way a resolver change does (I2); an edit about
other classes moves nothing. A pending item stays applicable while the bytes
still offer its chosen candidate, and a rescan's new item supersedes it only
while it is pending and undecided (never beside a decision, see
``PromptDatabase.record_review_item``). Apply re-resolves with the
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).
"""
from __future__ import annotations
@@ -29,6 +48,7 @@ import hashlib
import json
import logging
import os
from dataclasses import astuple
from pathlib import Path
from typing import Any
@@ -40,9 +60,19 @@ try:
canonical_evidence,
choosable,
)
from .inspection_semantics import (
SemanticsError,
get_semantics,
semantics_class_version,
)
from .media_capture import CaptureCapacity, CaptureError
from .prompt_inspector import MetadataReadError, read_inspection_file
from .prompt_pair_resolver import RESOLVER_VERSION, resolve_prompt_pair
from .prompt_pair_resolver import (
INSPECTION_PROFILE,
INSPECTION_RESOLVER_VERSION,
RESOLVER_VERSION,
resolve_prompt_pair,
)
except ImportError: # imported with the repo root on sys.path
from database.library_commands import LibraryCommandError
from database.review_schema import (
@@ -51,9 +81,19 @@ except ImportError: # imported with the repo root on sys.path
canonical_evidence,
choosable,
)
from utils.inspection_semantics import (
SemanticsError,
get_semantics,
semantics_class_version,
)
from utils.media_capture import CaptureCapacity, CaptureError
from utils.prompt_inspector import MetadataReadError, read_inspection_file
from utils.prompt_pair_resolver import RESOLVER_VERSION, resolve_prompt_pair
from utils.prompt_pair_resolver import (
INSPECTION_PROFILE,
INSPECTION_RESOLVER_VERSION,
RESOLVER_VERSION,
resolve_prompt_pair,
)
logger = logging.getLogger("prompt_manager.review_capture")
@@ -61,11 +101,136 @@ logger = logging.getLogger("prompt_manager.review_capture")
REVIEW_SCOPE = "needs-review"
# Nested ambiguous stages are flattened at most this deep.
MAX_CANDIDATE_DEPTH = 8
# The resolver profile an item's evidence was built with (``profile``).
DEFAULT = "default"
INSPECTION = "inspection"
# 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
def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> str:
# Above this many distinct node classes the whole table is the fingerprint.
MAX_FINGERPRINT_CLASSES = 1024
def graph_classes(metadata: Any) -> frozenset | None:
"""The node classes of ``metadata``'s prompt graph (empty without one).
None when the graph names more than ``MAX_FINGERPRINT_CLASSES`` classes.
"""
graph = metadata.get("prompt") if isinstance(metadata, dict) else None
if not isinstance(graph, dict):
return frozenset()
classes = set()
for node in graph.values():
name = node.get("class_type") if isinstance(node, dict) else None
if isinstance(name, str):
classes.add(name)
if len(classes) > MAX_FINGERPRINT_CLASSES:
return None
return frozenset(classes)
def semantics_version(metadata: Any = None) -> str | None:
"""The semantics table fingerprint that ``metadata``'s graph depends on.
The table restricted to the graph's node classes
(``inspection_semantics.class_version``), so an edit about other classes
leaves it alone; the whole table's version without ``metadata`` or for a
graph of very many classes. None when the table is unusable.
"""
try:
semantics = get_semantics()
except SemanticsError:
return None
classes = graph_classes(metadata) if metadata is not None else None
if classes is None:
return semantics.version
return semantics_class_version(semantics, classes)
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.
"""
table = semantics_version(metadata)
if table is None:
return None
profile = repr(astuple(INSPECTION_PROFILE))
seed = f"{profile}:{RESOLVER_VERSION}:{INSPECTION_RESOLVER_VERSION}:{table}"
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 inspection_choice(metadata: dict) -> 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.
"""
if inspection_review_version(metadata) is None:
return None
resolution = resolve_inspection(metadata)
if resolution.get("state") != "ambiguous" or not reviewable(resolution):
return None
return resolution
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:
return INSPECTION
return DEFAULT
def item_versions(metadata: Any = None) -> tuple:
"""The item versions a file with ``metadata`` may already be queued under.
Without ``metadata``, only the ingestion version.
"""
version = inspection_review_version(metadata) if metadata is not None else None
return (RESOLVER_VERSION,) if version is None else (RESOLVER_VERSION, version)
def _same_records(captured: dict, scanned: dict) -> bool:
"""True when the capture holds the records the scan resolved."""
keys = (set(captured) | set(scanned)) - {"file_info"}
return all(captured.get(key) == scanned.get(key) for key in keys)
def record_ambiguous_file(
db,
store,
root,
root_index: int,
relative: str,
*,
inspection: bool = False,
scanned: tuple | None = None,
) -> str:
"""Persist the ambiguous evidence of one rooted file.
With ``inspection`` (the scan), a default result without a choice is
resolved again with the inspection profile, and recorded from it when
that offers a choice. ``scanned`` is the
scan's ``(metadata, inspection_resolution_or_None)`` for the file: its
metadata names the item versions that count, and when the capture holds
the same records its resolution is reused instead of resolving again.
A new item supersedes the file's other items of the same bytes (an older
resolver or table version) when every one is still pending and
undecided; when one carries a decision, nothing new is stored.
Returns ``recorded`` (new item with candidates), ``limited`` (new item
without candidates), ``duplicate`` (these bytes already carry this
evidence, or the file's current stat is the verified fingerprint of
@@ -79,9 +244,10 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st
``CaptureCapacity`` and database errors propagate to the caller, which
logs and counts them.
"""
spelled_path, queued = _queued_spelling(db, root, relative)
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:
return _already_queued(db, *queued)
return _already_queued(db, *queued, versions)
epoch = db.get_execution_epoch()
try:
# Background: the last worker stays free for an interactive request.
@@ -105,9 +271,26 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st
resolution = captured[1] if captured is not None else None
if resolution is None or resolution.get("state") != "ambiguous":
return "skipped"
profile, version = DEFAULT, RESOLVER_VERSION
if not reviewable(resolution):
return "untraced"
evidence, count, limited = review_evidence(resolution)
version = inspection_review_version(captured[0]) if inspection else None
if version 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])
if resolution.get("state") != "ambiguous" or not reviewable(resolution):
return "untraced"
evidence, count, limited = review_evidence(
resolution, profile, version, captured[0]
)
try:
inserted = db.record_review_item(
locator=capture.locator,
@@ -116,7 +299,7 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st
size=capture.size,
mtime_ns=capture.mtime_ns,
epoch=capture.epoch,
resolver_version=RESOLVER_VERSION,
resolver_version=version,
evidence=evidence,
candidate_count=count,
limited=limited,
@@ -124,18 +307,21 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st
except LibraryCommandError:
return "skipped"
if not inserted:
return _already_queued(db, capture.locator, capture.size, capture.mtime_ns)
return _already_queued(
db, capture.locator, capture.size, capture.mtime_ns, versions
)
return "limited" if limited else "recorded"
def _already_queued(db, locator: str, size: int, mtime_ns: int) -> str:
"""``duplicate`` while the existing item is pending, else ``decided``."""
if db.review_pending_at(locator, size, mtime_ns, RESOLVER_VERSION):
return "duplicate"
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:
if db.review_pending_at(locator, size, mtime_ns, version):
return "duplicate"
return "decided"
def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]:
def _queued_spelling(db, root, relative: str, versions) -> tuple[str, tuple | None]:
"""The file's spelling, and ``(locator, size, mtime_ns)`` when queued.
The second value is None unless the file's current stat is already
@@ -147,7 +333,9 @@ def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]:
an item already cites is not copied or hashed again (maintainer ruling:
the rescan stays cheap and the file stays in the default scope); its
items take the current spelling. A stat failure leaves the decision to
the capture.
the capture. ``versions`` are the item versions that count; bytes whose
item carries a decision count under any version (a new version never
adds an item beside a decision).
"""
spelled = os.path.join(os.path.realpath(os.fspath(root)), *relative.split("/"))
try:
@@ -155,8 +343,9 @@ def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]:
except OSError:
return spelled, None
locator = os.path.normcase(spelled)
if db.review_fingerprint_queued(
locator, facts.st_size, facts.st_mtime_ns, RESOLVER_VERSION
if db.review_decided_at(locator, facts.st_size, facts.st_mtime_ns) or any(
db.review_fingerprint_queued(locator, facts.st_size, facts.st_mtime_ns, v)
for v in versions
):
# Not copied again, but a case-only rename keeps the stat: record
# the current spelling so the item's target still opens.
@@ -165,12 +354,14 @@ def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]:
return spelled, None
def captured_review(store, capture) -> tuple[dict, dict] | None:
def captured_review(store, capture, profile: str = DEFAULT) -> 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. None when the capture is gone or carries no prompt metadata.
capture. ``resolution`` comes from ``profile`` (``DEFAULT`` or
``INSPECTION``). None when the capture is gone or carries no prompt
metadata.
"""
if store.acquire(capture.token) is None:
return None
@@ -197,6 +388,8 @@ def captured_review(store, capture) -> tuple[dict, dict] | None:
if read.format is not None:
info["format"] = read.format
canonical["file_info"] = info
if profile == INSPECTION:
return canonical, resolve_inspection(canonical)
return canonical, resolve_prompt_pair(canonical)
@@ -218,12 +411,25 @@ def reviewable(resolution: Any) -> bool:
return False
def review_evidence(resolution: dict) -> tuple[dict, int, bool]:
"""``(evidence, candidate_count, limited)`` for one ambiguous result."""
def review_evidence(
resolution: dict,
profile: str = DEFAULT,
version: int | None = None,
metadata: Any = None,
) -> 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.
"""
base = {
"resolver_version": RESOLVER_VERSION,
"resolver_version": RESOLVER_VERSION if version is None else version,
"reason": _text(resolution.get("reason")),
}
if profile == INSPECTION:
base["profile"] = INSPECTION
base["semantics"] = semantics_version(metadata)
candidates: list[dict] = []
seen: set[str] = set()
total = 0