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

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

Browse Source 
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.
This commit is contained in:
Vito Sansevero
2026-10-23 16:53:19 -07:00
parent 143b2a6b44
commit ae112ceeb0
12 changed files with 425 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+16 -8
View File
@@ -2614,7 +2614,7 @@ 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
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,
@@ -2627,7 +2627,9 @@ evidence, which adds `profile: "inspection"` and `semantics`: the SHA-256 of
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, `RESOLVER_VERSION` and that fingerprint. A table edit about 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
@@ -2643,15 +2645,21 @@ 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). The 16-candidate limited
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,
@@ -2683,9 +2691,9 @@ 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. 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. 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
+42
View File
@@ -107,6 +107,13 @@ _OTHER_ITEMS_OF_BYTES = (
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
@@ -320,6 +327,9 @@ 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()
@@ -490,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/review_recording.py
+89 -36
View File
@@ -8,22 +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") unless the inspection profile finds a choice in it. Only the
scan asks that profile (``inspection=True``); the monitor never does.
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
@@ -32,8 +35,8 @@ monitor's pass lasts one file and is not a job: it is refused while a
the file again.
"""
import hashlib
import os
from collections import OrderedDict
from pathlib import Path
from .admin import SCAN_LOG_PREFIXES, _plural, _scan_location, _scan_roots
@@ -52,9 +55,22 @@ except ImportError:
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).
# 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"
@@ -126,9 +142,20 @@ class ReviewRecordingMixin:
its records depend on, and not again for a state it already found
untraced in this process.
"""
choice = self._inspection_choice(media_file, roots, canonical)
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"
@@ -154,7 +181,7 @@ class ReviewRecordingMixin:
facts = None
state = None
if facts is not None:
state = (facts.st_size, facts.st_mtime_ns, version)
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(
@@ -162,23 +189,56 @@ class ReviewRecordingMixin:
):
return canonical, None
memory = self._untraced_memory()
if state is not None and memory.get(locator) == state:
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(locator, None)
memory.pop(key, None)
return canonical, resolution
if state is not None:
memory[locator] = state
memory.move_to_end(locator)
memory.pop(key, None) # re-inserted last: the newest state
memory[key] = state
while len(memory) > MAX_UNTRACED_MEMORY:
memory.popitem(last=False)
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
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 = OrderedDict()
memory = self._untraced_inspections = {}
return memory
@staticmethod
@@ -262,23 +322,18 @@ 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, 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, 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")
if busy is not None:
@@ -296,9 +351,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_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
+23 -13
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,10 +863,11 @@ 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.
refuse it. Once the blocking job has ended it 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
@@ -889,8 +897,8 @@ class TestMonitorRecording(ReviewTestCase):
async with self.api._jobs.hold("probe-read", "Read probe", admission="read"):
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_monitor_wiring_reaches_new_and_existing_handlers(self):
self._require_module()
@@ -958,12 +966,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)
@@ -991,7 +999,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):
@@ -1032,7 +1040,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"
@@ -1041,11 +1050,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
+170 -2
View File
@@ -416,7 +416,9 @@ class TestApplyInspectionCandidate(InspectionTestCase):
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"])
@@ -425,6 +427,11 @@ class TestApplyInspectionCandidate(InspectionTestCase):
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):
@@ -437,7 +444,7 @@ class TestMonitorNeverQueues(InspectionTestCase):
self._roots()
path = self._write()
self.assertEqual(self.api.record_review_file(str(path)), "untraced")
self.assertEqual(self.api.record_review_file(str(path)), "skipped")
self.assertEqual(self._items(), [])
self.assertEqual(self._capture_files(), [])
@@ -455,7 +462,7 @@ class TestMonitorNeverQueues(InspectionTestCase):
# 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)), "untraced")
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):
@@ -785,3 +792,164 @@ class TestRecordingEdges(InspectionTestCase):
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/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
+15 -9
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
@@ -24,7 +25,7 @@ a ``limited`` item without candidates (a result without any candidate is not
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``; the live monitor never) asks ``INSPECTION_PROFILE``,
(``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
@@ -47,6 +48,7 @@ import hashlib
import json
import logging
import os
from dataclasses import astuple
from pathlib import Path
from typing import Any
@@ -67,6 +69,7 @@ try:
from .prompt_inspector import MetadataReadError, read_inspection_file
from .prompt_pair_resolver import (
INSPECTION_PROFILE,
INSPECTION_RESOLVER_VERSION,
RESOLVER_VERSION,
resolve_prompt_pair,
)
@@ -87,6 +90,7 @@ except ImportError: # imported with the repo root on sys.path
from utils.prompt_inspector import MetadataReadError, read_inspection_file
from utils.prompt_pair_resolver import (
INSPECTION_PROFILE,
INSPECTION_RESOLVER_VERSION,
RESOLVER_VERSION,
resolve_prompt_pair,
)
@@ -149,14 +153,16 @@ def semantics_version(metadata: Any = None) -> str | None:
def inspection_review_version(metadata: Any = None) -> int | None:
"""The ``resolver_version`` of an item recorded from the inspection profile.
Folds the profile, ``RESOLVER_VERSION`` and ``semantics_version(metadata)``,
so a resolver change or a table change about the graph's classes moves
it. None when the table cannot be loaded.
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
seed = f"{INSPECTION_PROFILE.name}:{RESOLVER_VERSION}:{table}"
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)
@@ -190,7 +196,7 @@ def item_profile(evidence: Any) -> str:
def item_versions(metadata: Any = None) -> tuple:
"""The item versions a file with ``metadata`` may already be queued under.
Without ``metadata`` (the monitor), only the ingestion version.
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)
@@ -216,7 +222,7 @@ def record_ambiguous_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; the live monitor passes False. ``scanned`` is the
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.