From ae112ceeb0be62ceb2244a89f2baf456a2750a10 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Fri, 23 Oct 2026 16:53:19 -0700 Subject: [PATCH] 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. --- TESTING.md | 24 ++- database/review.py | 42 ++++++ py/api/review_recording.py | 125 +++++++++++----- tests/browser/workspace-review.spec.mjs | 20 +-- tests/browser/workspace_review_fixture.py | 60 ++++---- tests/test_api_review_events.py | 12 +- tests/test_review.py | 36 +++-- tests/test_review_choosable.py | 10 +- tests/test_review_inspection.py | 172 +++++++++++++++++++++- tests/test_review_outcome.py | 8 +- utils/prompt_pair_resolver.py | 5 + utils/review_capture.py | 24 +-- 12 files changed, 425 insertions(+), 113 deletions(-) diff --git a/TESTING.md b/TESTING.md index 3474e076..a044dec9 100644 --- a/TESTING.md +++ b/TESTING.md @@ -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 diff --git a/database/review.py b/database/review.py index 7f9a25bd..f985c6b6 100644 --- a/database/review.py +++ b/database/review.py @@ -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. diff --git a/py/api/review_recording.py b/py/api/review_recording.py index 41d38fae..ac3039de 100644 --- a/py/api/review_recording.py +++ b/py/api/review_recording.py @@ -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): diff --git a/tests/browser/workspace-review.spec.mjs b/tests/browser/workspace-review.spec.mjs index cc9b8ed3..38a31840 100644 --- a/tests/browser/workspace-review.spec.mjs +++ b/tests/browser/workspace-review.spec.mjs @@ -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); }); diff --git a/tests/browser/workspace_review_fixture.py b/tests/browser/workspace_review_fixture.py index 595904d6..b2f5c191 100644 --- a/tests/browser/workspace_review_fixture.py +++ b/tests/browser/workspace_review_fixture.py @@ -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}) diff --git a/tests/test_api_review_events.py b/tests/test_api_review_events.py index 8a1b5dbc..9443c222 100644 --- a/tests/test_api_review_events.py +++ b/tests/test_api_review_events.py @@ -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") diff --git a/tests/test_review.py b/tests/test_review.py index 70776c02..e2e44ed9 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -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) diff --git a/tests/test_review_choosable.py b/tests/test_review_choosable.py index de865405..ba6fd3df 100644 --- a/tests/test_review_choosable.py +++ b/tests/test_review_choosable.py @@ -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(), []) diff --git a/tests/test_review_inspection.py b/tests/test_review_inspection.py index ef3f3e92..93cb78d5 100644 --- a/tests/test_review_inspection.py +++ b/tests/test_review_inspection.py @@ -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)) diff --git a/tests/test_review_outcome.py b/tests/test_review_outcome.py index c4952c80..b24a5242 100644 --- a/tests/test_review_outcome.py +++ b/tests/test_review_outcome.py @@ -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__": diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index 4ad10a5c..6eb10be7 100644 --- a/utils/prompt_pair_resolver.py +++ b/utils/prompt_pair_resolver.py @@ -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) diff --git a/utils/review_capture.py b/utils/review_capture.py index 8a4f8e22..1859dee7 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -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.