diff --git a/TESTING.md b/TESTING.md index 7881dc3..52667c4 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2725,15 +2725,52 @@ still have made the file, so that text is not certain either. Both report `unresolved` ("could not be traced"), are never linked, and their pending, undecided items whose own evidence offers no choice (a limited item counts as a choice when it holds at least two candidates) leave the queue on the next -scan or monitor pass (`settle_untraced_review_items`). The maintainer's -second live check found 39 items whose 11 candidates were all "unresolved". The queue heals itself: after each batch, the scan removes the pending +scan (`settle_untraced_review_items`). The maintainer's +second live check found 39 items whose 11 candidates were all "unresolved". +Inspection-profile choices (`tests/test_review_inspection.py`, ruling a): when +the default profile is ambiguous without a choice and nothing links the file, +the scan (never the monitor) resolves it again with `INSPECTION_PROFILE`, which +knows `TextEncodeQwenImageEdit`, `StringConstantMultiline`, `JoinStringMulti` +and the rest of `web/js/inspection-semantics.json`. With at least two choosable +candidates there, the file is `ambiguous` and is queued from that profile's +evidence, which adds `profile: "inspection"` and `semantics`: the SHA-256 of +`class_projection`, the table restricted to the graph's node classes plus its +class-independent `schema` and `stage_inputs` (the resolver looks the table up +only by a graph node's class). The item's `resolver_version` is +`inspection_review_version(metadata)`, 2^40 plus 40 bits of SHA-256 over the +profile with its budgets, `RESOLVER_VERSION`, `INSPECTION_RESOLVER_VERSION` +(bump it when the inspection profile would resolve the same metadata otherwise) +and that fingerprint. A table edit about the +graph's classes behaves like a resolver change (I2); an edit about other +classes changes nothing. One file never holds two pending items for the same +bytes: a new version's item supersedes the byte record's other items when all +are pending and undecided (revision 0, no decision row, the self-heal's safe +delete); beside a decided or advanced item nothing new is stored, and the +rescan's stat shortcut then skips those bytes under any version. A pending item +stays applicable while the bytes still offer its chosen candidate. +Apply re-resolves the capture with the profile the item recorded and re-checks +the chosen hash; the link goes through the same compatible-only ingestion. The +inspection profile never links on its own, and ingestion keeps the default +profile (D1). The heavier profile runs only for files already ambiguous without +a choice, and once per file state: the capture reuses the scan's result when +it holds the same records; a rescan skips files already queued or decided at +their current size and `mtime_ns` under their version, and files this process +already found untraced at the same state (an in-memory map of at most 65,536 +files, oldest forgotten first, keyed by 16-byte digests: 8.6 MiB measured at +the cap). A file whose inspection raises is logged, counted in +`review_errors` and reported `unresolved`; the scan goes on. An older version's +pending, undecided item of bytes now untraced leaves the queue +(`settle_stale_review_items`), so an item never stays stuck at 409. The 16-candidate limited +rule and the 8 MiB bound apply unchanged. The queue heals itself: after each batch, the scan removes the pending items of every listed file (read or skipped as known) that a saved row now links, and the monitor does the same for each ambiguous output it links (`settle_linked_review_items`; the review index revision moves). Only an undecided item goes (`pending`, revision 0, no decision row): a decision is never deleted or changed, so no schema state was added. The monitor never -queues an output it linked; an unlinked file passed to `record_review_file` -follows the scan's rule (a file the fallback would link is `skipped`). The +records an item (only scans add items, maintainer ruling 2026-10-23): an +unlinked file passed to `record_review_file` is `skipped` with no capture, +and only a linked one's pending items are settled +(`test_monitor_never_records_an_unlinked_ambiguous_output`). The maintainer's live check of #284/#288 found the queue at 10,232 against a review group of 4,677: every fallback-linked ambiguous file had been queued. It captures the file (scope `needs-review`), reads its metadata again from the capture, @@ -2765,9 +2802,10 @@ Apply, and waits up to 2 s for room before counting the file as not added pass runs after its own link, best effort: a failure is logged and the link stays. Recording follows the registry's `records` admission (#289): the scan's recording runs under its own `records` job, and -the monitor's (one file, not a job) is refused while a `records` or `guarded` -job or hold runs, except thumbnail work, which writes no records (S5 M7). It then records nothing, logs that the file waits for the -next scan, and answers `deferred`; the next "new" scan reads the ambiguous file +the monitor's pass (one file, not a job, which only settles) is refused +while a `records` or `guarded` job or hold runs, except thumbnail work, which +writes no records (S5 M7). It then writes nothing, logs that the file waits +for the next scan, and answers `deferred`; the next "new" scan reads the ambiguous file again (`test_monitor_recording_is_refused_beside_record_changing_jobs`). A new item moves `review_index_revision`. A rescan does not copy or hash an ambiguous file whose current size and `mtime_ns` equal the verified fingerprint of bytes an item already cites (a diff --git a/database/review.py b/database/review.py index 51321d8..f985c6b 100644 --- a/database/review.py +++ b/database/review.py @@ -90,6 +90,30 @@ _PENDING_AT = ( "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.resolver_version=? " "AND i.state='pending' LIMIT 1" ) +# Bytes verified at this stat whose item carries a decision (any version). +_DECIDED_AT = ( + f"SELECT 1 FROM {FILES_TABLE} f " # nosec B608 + f"JOIN {BYTES_TABLE} b ON b.file_id=f.id " + f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id " + "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? " + "AND (i.state<>'pending' OR i.revision<>0 OR EXISTS (SELECT 1 FROM " + f"{DECISIONS_TABLE} d WHERE d.item_id=i.id)) LIMIT 1" +) +# The other items of one byte record (another version or evidence), with +# whether each may be superseded: pending, revision 0, no decision row. +_OTHER_ITEMS_OF_BYTES = ( + "SELECT i.id, (i.state='pending' AND i.revision=0 AND NOT EXISTS " # nosec B608 + f"(SELECT 1 FROM {DECISIONS_TABLE} d WHERE d.item_id=i.id)) " + f"FROM {ITEMS_TABLE} i WHERE i.byte_id=? " + "AND NOT (i.resolver_version=? AND i.evidence_hash=?)" +) +# Pending items of the bytes verified at one stat, with their versions. +_PENDING_AT_STAT = ( + f"SELECT i.id,i.resolver_version FROM {FILES_TABLE} f " # nosec B608 + f"JOIN {BYTES_TABLE} b ON b.file_id=f.id " + f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id " + "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.state='pending'" +) # Pending items of one file: the candidates a link settles. _PENDING_OF_LOCATOR = ( f"SELECT i.id FROM {FILES_TABLE} f " # nosec B608 @@ -274,6 +298,12 @@ class ReviewOperations: ``LibraryCommandError("storage-changed")`` when the storage epoch is no longer ``epoch`` (a restore since the capture) and ``ValueError`` for invalid facts or evidence. + + One file never holds two pending items for the same bytes: the new + item supersedes the byte record's other items (an older resolver or + semantics version) when every one is pending and undecided (the safe + delete of the self-heal); when one carries a decision or a later + revision, it is kept and nothing new is stored (False). """ _validate_target(locator, sha256, size, mtime_ns) _validate_spelling(spelled_path, locator) @@ -297,6 +327,18 @@ class ReviewOperations: ) if created or moved: conn.execute(_BUMP_IMAGE_INDEX) + # The byte record is created or refreshed even when the item is + # refused below (an observe would do the same); that commits + # with the image index bump, and the review queue is unchanged. + others = conn.execute( + _OTHER_ITEMS_OF_BYTES, (byte_id, resolver_version, evidence_hash) + ).fetchall() + if any(not removable for _id, removable in others): + return False + superseded = sum( + conn.execute(_SETTLE_ITEM, (item_id,)).rowcount + for item_id, _removable in others + ) cursor = conn.execute( _INSERT_ITEM, ( @@ -314,9 +356,9 @@ class ReviewOperations: ) inserted = cursor.rowcount > 0 refreshed = _refresh_spelling(conn, locator, spelled_path) - if inserted or refreshed: + if inserted or refreshed or superseded: conn.execute(_BUMP_REVIEW_INDEX) - if inserted or refreshed: + if inserted or refreshed or superseded: notify_review_change(self.model.db_path) return inserted @@ -389,6 +431,19 @@ class ReviewOperations: ) return row is not None + def review_decided_at(self, locator: str, size: int, mtime_ns: int) -> bool: + """True when bytes verified at this stat carry a decided item. + + Any version: a newer resolver or table never adds an item beside a + decision, so the rescan need not copy those bytes again. + """ + row = ( + self.model.get_connection() + .execute(_DECIDED_AT, (locator, size, mtime_ns)) + .fetchone() + ) + return row is not None + def review_pending_at( self, locator: str, size: int, mtime_ns: int, resolver_version: int ) -> bool: @@ -445,6 +500,38 @@ class ReviewOperations: notify_review_change(self.model.db_path) return removed + def settle_stale_review_items( + self, locator: str, size: int, mtime_ns: int, versions + ) -> int: + """Remove pending, undecided items of the bytes verified at this stat + whose ``resolver_version`` is not one of ``versions``. + + The caller found these bytes untraced under the current versions, so + an older version's item could only refuse Apply. The same safe delete + as the other settles; one indexed read when nothing is pending. + """ + _validate_locator(locator) + current = set(versions) + conn = self.model.get_connection() + stale = [ + item_id + for item_id, version in conn.execute( + _PENDING_AT_STAT, (locator, size, mtime_ns) + ) + if version not in current + ] + if not stale: + return 0 + removed = 0 + with self._pair_transaction() as conn: + for item_id in stale: + removed += conn.execute(_SETTLE_ITEM, (item_id,)).rowcount + if removed: + conn.execute(_BUMP_REVIEW_INDEX) + if removed: + notify_review_change(self.model.db_path) + return removed + def settle_untraced_review_items(self, locators) -> int: """Remove the pending, undecided items that offer nothing to choose. diff --git a/py/api/admin.py b/py/api/admin.py index 0b9b89c..4b2e9c7 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -1646,7 +1646,7 @@ class AdminRoutesMixin: if self._scan_job and self._scan_job.cancelled: break counts["processed"] += 1 - ambiguous, untraced = [], [] + ambiguous, untraced, records = [], [], {} try: outcome = self._ingest_scanned_file( media_file, @@ -1654,6 +1654,7 @@ class AdminRoutesMixin: detailed=True, ambiguous=ambiguous, untraced=untraced, + records=records, ) except Exception as e: self.logger.error(f"Error processing {media_file.name}: {e}") @@ -1666,7 +1667,11 @@ class AdminRoutesMixin: else: self._record_scanned_review(media_file, roots, review) elif untraced and outcome == "unresolved": - self._settle_untraced_reviews(untraced, roots, review) + # Untraced by the default profile: the inspection profile may + # still offer a choice (queued for review, never linked). + outcome = self._untraced_scan_outcome( + media_file, roots, records.get("canonical"), review + ) if outcome is None: note(media_file, None) continue @@ -1701,7 +1706,13 @@ class AdminRoutesMixin: return positive.strip() or None def _ingest_scanned_file( - self, media_file, metadata, detailed=False, ambiguous=None, untraced=None + self, + media_file, + metadata, + detailed=False, + ambiguous=None, + untraced=None, + records=None, ): """Enrich exact pairs; retain legacy positive-only import for absent pairs. @@ -1714,7 +1725,9 @@ class AdminRoutesMixin: ``"ambiguous"`` (nothing linked it) is a Needs review case. An ambiguous result with nothing to choose (``reviewable`` is False: fewer than two candidates carry a prompt) that nothing links returns - ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``. + ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``; + its parsed records go to ``records["canonical"]`` when ``records`` is + a dict. """ if not metadata: return None @@ -1735,6 +1748,8 @@ class AdminRoutesMixin: if unchoosable: if untraced is not None: untraced.append(media_file) + if records is not None: + records["canonical"] = canonical elif resolution["state"] == "ambiguous" and ambiguous is not None: ambiguous.append(media_file) if resolution["state"] == "resolved": diff --git a/py/api/review.py b/py/api/review.py index 6fb5f4f..9490ae3 100644 --- a/py/api/review.py +++ b/py/api/review.py @@ -38,13 +38,17 @@ try: from ...database.library_commands import LibraryCommandError from ...database.review_decisions import ReviewCommandError from ...utils.media_capture import CaptureError - from ...utils.review_capture import captured_review, review_evidence + from ...utils.review_capture import ( + captured_review, + item_profile, + review_evidence, + ) except ImportError: from database.library import display_title from database.library_commands import LibraryCommandError from database.review_decisions import ReviewCommandError from utils.media_capture import CaptureError - from utils.review_capture import captured_review, review_evidence + from utils.review_capture import captured_review, item_profile, review_evidence ITEMS_URL = "/prompt_manager/review/items" DEFAULT_LIMIT = 40 @@ -641,7 +645,8 @@ def _verify_and_read(store, capture, item, root, need_metadata: bool): """Worker thread: ``(relative, facts, captured)``, or None off the root. Formats the item's target under the capture's root (resolving the root's - real path), re-hashes the current file, then reads the capture. + real path), re-hashes the current file, then reads the capture and + resolves it with the profile the item's evidence was recorded with. """ found = format_review_target( item["spelled_path"], @@ -652,7 +657,9 @@ def _verify_and_read(store, capture, item, root, need_metadata: bool): return None relative = found[1] facts = store.verify(root, relative) - captured = captured_review(store, capture) if need_metadata else None + captured = None + if need_metadata: + captured = captured_review(store, capture, item_profile(item["evidence"])) return relative, facts, captured diff --git a/py/api/review_recording.py b/py/api/review_recording.py index 7cb60c6..6de5964 100644 --- a/py/api/review_recording.py +++ b/py/api/review_recording.py @@ -8,21 +8,25 @@ already. A file the fallback links, an exact pair the ingester stored and an already linked file are never queued. Neither is an ambiguous file with nothing to choose: fewer than two of its candidates carry a prompt (``utils.review_capture.reviewable``); the scan reports it unresolved ("could -not be traced"). +not be traced") unless the inspection profile finds a choice in it. + +Only scans add items (maintainer ruling, 2026-10-23). The monitor's pass +records nothing: it only removes the pending items of an output it linked. The queue heals itself: a pending item whose file is now linked (by the fallback, a later resolution or the monitor) is removed by the next scan -that lists the file or by the monitor's next pass over it, and so is a -pending item without a choice whose file is still ambiguous without one -(``settle_untraced_review_items``). Only an undecided item goes +that lists the file or by the monitor's next pass over it. The next scan +also removes a pending item without a choice whose file is still ambiguous +without one (``settle_untraced_review_items``), and an older version's +item of bytes now untraced (``settle_stale_review_items``). Only an undecided item goes (``pending``, revision 0, no decision row); a decision is never deleted or changed (``settle_linked_review_items``). The scan records through ``_record_scanned_review`` (best effort: a failure -is logged and counted); the monitor calls ``record_review_file`` from its -worker thread after its own link. Both go through -``utils.review_capture.record_ambiguous_file`` on the API's capture store. -Nothing here links a file. +is logged and counted), which goes through +``utils.review_capture.record_ambiguous_file`` on the API's capture store; +the monitor calls ``record_review_file`` from its worker thread after its +own link. Nothing here links a file. Recording changes records, so it follows the registry's ``records`` admission (#289). The scan records under its own ``records`` job. The @@ -35,6 +39,7 @@ defer the recording (#249 S5 M7): a long rebuild no longer holds Needs review back for its whole run. """ +import hashlib import os from pathlib import Path @@ -50,10 +55,29 @@ QUEUED_OUTCOMES = ("recorded", "limited", "duplicate") NEW_OUTCOMES = ("recorded", "limited") try: + from ...utils import review_capture from ...utils.review_capture import record_ambiguous_file except ImportError: + from utils import review_capture from utils.review_capture import record_ambiguous_file +# Untraced files remembered per process so a rescan of an unchanged file does +# not run the inspection profile again (oldest forgotten first). A plain dict +# of 16-byte digests (the locator, and its size, mtime_ns and item version): +# measured at 8.6 MiB with 65,536 entries (tracemalloc, CPython 3.14). +MAX_UNTRACED_MEMORY = 65_536 + + +def _digest(text: str) -> bytes: + data = text.encode("utf-8", "surrogatepass") + return hashlib.blake2b(data, digest_size=16).digest() + + +def _memory_key(locator: str) -> bytes: + """A 16-byte digest of ``locator`` (the untraced memory's key).""" + return _digest(locator) + + # The scan log's reason for items that leave the queue with nothing to choose. NOTHING_TO_CHOOSE = "nothing to choose" @@ -61,10 +85,15 @@ NOTHING_TO_CHOOSE = "nothing to choose" class ReviewRecordingMixin: """Record the scan's and the monitor's ambiguous files for Needs review.""" - def _record_scanned_review(self, media_file, roots, review=None): - """Record one ambiguous, unlinked scanned file (logged and counted).""" + def _record_scanned_review(self, media_file, roots, review=None, scanned=None): + """Record one ambiguous, unlinked scanned file (logged and counted). + + ``scanned`` is the scan's ``(metadata, inspection_resolution)``. + """ try: - outcome = self._record_review_sync(media_file, roots, review) + outcome = self._record_review_sync( + media_file, roots, review, inspection=True, scanned=scanned + ) except Exception: self.logger.warning( f"{media_file.name} could not be added to Needs review", @@ -80,9 +109,13 @@ class ReviewRecordingMixin: if outcome in QUEUED_OUTCOMES: review["queued"] += 1 - def _record_review_sync(self, media_file, roots, review=None): + def _record_review_sync( + self, media_file, roots, review=None, inspection=False, scanned=None + ): """``record_ambiguous_file`` for a file under one of ``roots`` (blocking). + ``inspection`` (the scan only) lets a file without a choice under the + default profile be recorded from the inspection profile's choices. An ``untraced`` file (its captured bytes offer nothing to choose) also loses its pending items that offer nothing either. """ @@ -91,12 +124,134 @@ class ReviewRecordingMixin: return "skipped" root, index, relative, _spelled = located outcome = record_ambiguous_file( - self.db, self._capture_store(), Path(root), index, relative + self.db, + self._capture_store(), + Path(root), + index, + relative, + inspection=inspection, + scanned=scanned, ) if outcome == "untraced": self._settle_untraced_reviews([media_file], roots, review) return outcome + def _untraced_scan_outcome(self, media_file, roots, canonical, review=None): + """The scan status of an unlinked file the default profile left + ambiguous without a choice (blocking). + + ``ambiguous`` when the inspection profile offers a choice: the file is + recorded for review from that profile (never linked), or ``unchanged`` + when a saved row links it already. Otherwise ``unresolved``, and its + pending items without a choice leave the queue. The profile runs at + most once per file state: not for bytes already queued (or decided) + at the file's current size and ``mtime_ns`` under the version that + its records depend on, and not again for a state it already found + untraced in this process. + """ + try: + choice = self._inspection_choice(media_file, roots, canonical) + except Exception: + # One bad file never stops the scan: it stays "could not be traced". + self.logger.warning( + f"{media_file.name} could not be inspected for Needs review", + exc_info=True, + ) + if review is not None: + review["errors"] += 1 + return "unresolved" + if choice is None: + self._settle_untraced_reviews([media_file], roots, review) + self._settle_stale_reviews(media_file, roots, canonical, review) + return "unresolved" + if self._review_file_linked(media_file, roots): + return "unchanged" + self._record_scanned_review(media_file, roots, review, scanned=choice) + return "ambiguous" + + def _inspection_choice(self, media_file, roots, canonical): + """``(canonical, resolution_or_None)`` when the file offers a choice. + + The resolution is None when the file's current state is already in + the queue (so the profile did not run). None when it offers none. + """ + if canonical is None: + return None + version = review_capture.inspection_review_version(canonical) + located = self._review_location(media_file, roots) + if version is None or located is None: + return None + locator = os.path.normcase(located[3]) + try: + facts = os.lstat(located[3]) + except OSError: + facts = None + state = None + if facts is not None: + state = _digest(f"{facts.st_size}:{facts.st_mtime_ns}:{version}") + if self.db.review_decided_at( + locator, facts.st_size, facts.st_mtime_ns + ) or self.db.review_fingerprint_queued( + locator, facts.st_size, facts.st_mtime_ns, version + ): + return canonical, None + memory = self._untraced_memory() + key = _memory_key(locator) + if state is not None and memory.get(key) == state: + return None + resolution = review_capture.inspection_choice(canonical) + if resolution is not None: + memory.pop(key, None) + return canonical, resolution + if state is not None: + memory.pop(key, None) # re-inserted last: the newest state + memory[key] = state + while len(memory) > MAX_UNTRACED_MEMORY: + del memory[next(iter(memory))] + return None + + def _settle_stale_reviews(self, media_file, roots, canonical, review=None): + """Remove the pending, undecided items of an untraced file's current + bytes recorded under another version (best effort, counted in + ``review["untraced"]``). + + The current resolver and table offer nothing to choose in these + bytes, so an item of an older version would only refuse Apply. + """ + located = self._review_location(media_file, roots) + if located is None or canonical is None: + return 0 + # An unusable semantics table says nothing about the current + # versions; never let it retire items a working table recorded. + if review_capture.inspection_review_version(canonical) is None: + return 0 + try: + facts = os.lstat(located[3]) + removed = self.db.settle_stale_review_items( + os.path.normcase(located[3]), + facts.st_size, + facts.st_mtime_ns, + review_capture.item_versions(canonical), + ) + except Exception: + self.logger.warning( + f"Stale Needs review items of {media_file.name} could not be " + "removed", + exc_info=True, + ) + if review is not None: + review["errors"] += 1 + return 0 + if review is not None: + review["untraced"] += removed + return removed + + def _untraced_memory(self): + memory = getattr(self, "_untraced_inspections", None) + if memory is None: + memory = self._untraced_inspections = {} + return memory + @staticmethod def _review_location(media_file, roots, real_roots=None): """``(root, root_index, relative, spelled)`` of a listed file, or None. @@ -178,24 +333,19 @@ class ReviewRecordingMixin: review["untraced"] += removed return removed - def _fallback_links(self, metadata): - """True when the scan's legacy fallback would link these records.""" - return bool(metadata.get("prompt")) and bool(self._graph_prompt_text(metadata)) - def record_review_file(self, image_path): """Apply the Needs review rule to one monitor output (its worker thread). - Maps ``image_path`` to the gallery root that lists it; a file under no - current root is ``skipped``. A linked file is never queued: its - pending items are removed and the answer is ``linked``. An unlinked - file the scan's fallback would link is ``skipped`` (the next scan - links it); any other file is recorded when it is ambiguous with a - choice, and is ``untraced`` (its pending items without a choice - removed) when it is ambiguous without one. While a - job runs that a ``records`` job may not run beside, other than - thumbnail work (``PREVIEW_ONLY_KINDS``), nothing is written and the - answer is ``deferred``. Errors propagate: the monitor logs - them and keeps its link. + Only scans add items (maintainer ruling, 2026-10-23): the monitor + only clears the items of outputs it links. Maps ``image_path`` to the + gallery root that lists it; a file under no current root is + ``skipped``. A linked file's pending items are removed and the answer + is ``linked``. Any other file is ``skipped``: nothing is captured or + recorded, and the next scan reads it. While a job runs that a + ``records`` job may not run beside, other than thumbnail work + (``PREVIEW_ONLY_KINDS``), nothing is written and the answer is + ``deferred``. Errors propagate: the monitor logs them and keeps its + link. """ busy = self._jobs.blocker( REVIEW_RECORDING_KIND, admission="records", compatible=PREVIEW_ONLY_KINDS @@ -215,9 +365,7 @@ class ReviewRecordingMixin: if self._review_file_linked(located, roots): self._settle_scanned_reviews([located], roots) return "linked" - if self._fallback_links(self._extract_comfyui_metadata(str(located))): - return "skipped" - return self._record_review_sync(located, roots) + return "skipped" @staticmethod def _log_review_items(job, review): diff --git a/tests/browser/workspace-review.spec.mjs b/tests/browser/workspace-review.spec.mjs index cc9b8ed..38a3184 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 595904d..b2f5c19 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_admin.py b/tests/test_api_admin.py index a03164d..1ee0782 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -2243,7 +2243,12 @@ class TestActivityJobs(AdminAPITestCase): (self.output_dir / name).write_bytes(b"not an image") def fake_ingest( - media_file, metadata, detailed=False, ambiguous=None, untraced=None + media_file, + metadata, + detailed=False, + ambiguous=None, + untraced=None, + records=None, ): key = Path(media_file).relative_to(self.output_dir).as_posix() outcome = names[key] diff --git a/tests/test_api_review_events.py b/tests/test_api_review_events.py index 8a1b5db..9443c22 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 ad41001..a313e31 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,12 +863,13 @@ class TestMonitorRecording(ReviewTestCase): async def test_monitor_recording_is_refused_beside_record_changing_jobs(self): """S1 admission: a ``records`` recording waits for no job; it is refused. - While a ``records`` or ``guarded`` job runs, the monitor's recording - records nothing and answers ``deferred`` (its fallback link stands and + While a ``records`` or ``guarded`` job runs, the monitor's pass + writes nothing and answers ``deferred`` (its fallback link stands and a later scan reads the ambiguous file again); a ``read`` job does not - refuse it, and it records once the blocking job has ended. Thumbnail - work (the job and the gallery's holds) is ``guarded`` but writes only - previews, never records, so it does not defer the recording (S5 M7). + refuse it. Thumbnail work (the job and the gallery's holds) is + ``guarded`` but writes only previews, never records, so it does not + defer the pass either (S5 M7). Not deferred, the pass still records + nothing: only scans add items, so an unlinked output is ``skipped``. """ self._require_module() from py.api.jobs import RESTORE_JOB_KIND, THUMBNAILS_JOB_KIND @@ -898,11 +906,11 @@ class TestMonitorRecording(ReviewTestCase): outcome = await asyncio.to_thread( self.api.record_review_file, str(path) ) - self.assertEqual(outcome, "recorded") - self.assertEqual(len(self._items()), 1) + self.assertEqual(outcome, "skipped") + self.assertEqual(self._items(), []) async def test_a_running_thumbnail_job_does_not_defer_monitor_recording(self): - """S5 M7: a long rebuild never holds Needs review back (no records).""" + """S5 M7: a long rebuild never defers the monitor's pass (no records).""" self._require_module() from py.api.jobs import THUMBNAILS_JOB_KIND @@ -927,8 +935,9 @@ class TestMonitorRecording(ReviewTestCase): finally: gate.set() await asyncio.wait_for(job.task, 5) - self.assertEqual(outcome, "recorded") - self.assertEqual(len(self._items()), 1) + # Not deferred (thumbnail work writes no records); only scans add items. + self.assertEqual(outcome, "skipped") + self.assertEqual(self._items(), []) async def test_monitor_wiring_reaches_new_and_existing_handlers(self): self._require_module() @@ -996,12 +1005,12 @@ class TestRecordedSpelling(ReviewTestCase): upper = self.output_dir / "Two.PNG" write_png(upper, prompt=unlinkable_graph("cat", "dog")) with _case_folding(): - self.assertEqual(self.api.record_review_file(str(upper)), "recorded") + self.assertEqual(self._scan_record(upper), "recorded") (before,) = self._items() revision = self._review_revision() lower = self.output_dir / "two.png" os.rename(upper, lower) # keeps size and mtime_ns - self.assertEqual(self.api.record_review_file(str(lower)), "duplicate") + self.assertEqual(self._scan_record(lower), "duplicate") (after,) = self._items() self.assertEqual(self._spelled(), [str(lower)]) self.assertEqual(self._review_revision(), revision + 1) @@ -1029,7 +1038,7 @@ class TestRecordedSpelling(ReviewTestCase): self.assertEqual(capture.locator, after["locator"]) self.api._captures.release(capture.token) # Seeing the same spelling again writes nothing. - self.assertEqual(self.api.record_review_file(str(lower)), "duplicate") + self.assertEqual(self._scan_record(lower), "duplicate") self.assertEqual(self._review_revision(), revision + 1) async def test_duplicate_insert_refreshes_the_spelling(self): @@ -1070,7 +1079,8 @@ class TestRecordedSpelling(ReviewTestCase): class TestCaptureStoreShutdown(ReviewTestCase): async def test_recording_after_close_creates_no_store(self): - """A monitor recording still in flight at shutdown leaks nothing (M2).""" + """A scan recording still in flight at shutdown leaks nothing (M2), + and the monitor's pass after it captures nothing at all.""" self._require_module() self._roots() path = self.output_dir / "two.png" @@ -1079,11 +1089,12 @@ class TestCaptureStoreShutdown(ReviewTestCase): self.assertIsNone(self.api._captures) with patch("utils.media_capture.tempfile.mkdtemp") as mkdtemp: with self.assertRaises(CaptureError): - self.api.record_review_file(str(path)) - # The monitor logs the failure and keeps its link. + self._scan_record(path) + # The monitor's pass records nothing, so it needs no store. handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) handler.review_recorder = self.api.record_review_file handler._record_review(str(path)) + self.assertEqual(self.api.record_review_file(str(path)), "skipped") mkdtemp.assert_not_called() self.assertIsNone(self.api._captures) self.assertIsNone(self.api._captures_unsubscribe) diff --git a/tests/test_review_choosable.py b/tests/test_review_choosable.py index de86540..ba6fd3d 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 new file mode 100644 index 0000000..36cba58 --- /dev/null +++ b/tests/test_review_inspection.py @@ -0,0 +1,971 @@ +"""Needs review choices from the inspection profile (follow-up to #284/#288). + +A Qwen image-edit batch runs several KSamplers, each with its own +``TextEncodeQwenImageEdit(prompt=...)``, and every SaveImage shares one +prefix: the file cannot say which prompt made it. The scan's default +(ingestion) profile does not know the encoder or the string nodes, so every +candidate is unresolved and the file used to leave the queue as "could not +be traced". The inspection profile resolves every text, so these are real +choices. Ruling (option a): when the default profile is ambiguous with fewer +than two candidates that carry text, the scan asks the inspection profile; +with at least two choosable candidates it records the item from that +profile, and Apply re-resolves with the same profile. Nothing links +automatically from the inspection profile, and the live monitor never queues. +""" + +import contextlib +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +try: + from tests.test_api_review import ReviewRoutesTestCase + from tests.test_review import unlinkable_graph, write_png +except ImportError: # discovered with tests/ as the top-level directory + from test_api_review import ReviewRoutesTestCase + from test_review import unlinkable_graph, write_png + +from utils import inspection_semantics, review_capture +from utils.hashing import generate_prompt_hash +from utils.image_monitor import ImageGenerationHandler +from utils.prompt_tracker import PromptTracker + +NAME = "kittybear_00001_.png" +PROMPTS = ("make it night", "add a red hat", "turn the cat into a bear") + + +@contextlib.contextmanager +def _counted(module, name): + """Count the calls of ``module.name`` while calling through.""" + real = getattr(module, name) + calls = [] + + def spy(*args, **kwargs): + calls.append(args) + return real(*args, **kwargs) + + with patch.object(module, name, side_effect=spy): + yield calls + + +def qwen_edit_graph(*prompts, joined=()): + """One Qwen image-edit sampler per prompt, all saved under one prefix. + + Each sampler has its own ``TextEncodeQwenImageEdit``, used for both + positive and negative as the real batches do. ``joined`` prompts are + ``(first, second)`` pairs joined by ``JoinStringMulti`` over two + ``StringConstantMultiline`` nodes. Every SaveImage takes its prefix from + one linked string node. + """ + graph = { + "1": {"class_type": "UNETLoader", "inputs": {"unet_name": "q.safetensors"}}, + "2": {"class_type": "CLIPLoader", "inputs": {"clip_name": "c.safetensors"}}, + "3": {"class_type": "VAELoader", "inputs": {"vae_name": "v.safetensors"}}, + "4": {"class_type": "LoadImage", "inputs": {"image": "source.png"}}, + "5": {"class_type": "PrimitiveString", "inputs": {"value": "kittybear"}}, + "6": { + "class_type": "EmptyLatentImage", + "inputs": {"width": 512, "height": 512, "batch_size": 1}, + }, + } + sides = [("literal", text) for text in prompts] + sides += [("joined", parts) for parts in joined] + for index, (kind, text) in enumerate(sides): + base = 100 + 10 * index + prompt = text + if kind == "joined": + for offset, part in enumerate(text, start=1): + graph[str(base + offset)] = { + "class_type": "StringConstantMultiline", + "inputs": {"string": part, "strip_newlines": True}, + } + graph[str(base + 3)] = { + "class_type": "JoinStringMulti", + "inputs": { + "inputcount": 2, + "delimiter": ", ", + "return_list": False, + "string_1": [str(base + 1), 0], + "string_2": [str(base + 2), 0], + }, + } + prompt = [str(base + 3), 0] + encoder, sampler = str(base), str(base + 5) + graph[encoder] = { + "class_type": "TextEncodeQwenImageEdit", + "inputs": { + "clip": ["2", 0], + "prompt": prompt, + "vae": ["3", 0], + "image": ["4", 0], + }, + } + graph[sampler] = { + "class_type": "KSampler", + "inputs": { + "model": ["1", 0], + "positive": [encoder, 0], + "negative": [encoder, 0], + "latent_image": ["6", 0], + "seed": index, + }, + } + graph[str(base + 6)] = { + "class_type": "VAEDecode", + "inputs": {"samples": [sampler, 0], "vae": ["3", 0]}, + } + graph[str(base + 7)] = { + "class_type": "SaveImage", + "inputs": {"images": [str(base + 6), 0], "filename_prefix": ["5", 0]}, + } + return graph + + +def _prefixed(graph, prefix): + """``graph`` with every node id (and link to one) prefixed.""" + renamed = {} + for node_id, node in graph.items(): + inputs = { + key: ( + [prefix + value[0], value[1]] + if isinstance(value, list) and len(value) == 2 + else value + ) + for key, value in node["inputs"].items() + } + renamed[prefix + node_id] = {**node, "inputs": inputs} + return renamed + + +def _version(graph=None): + """The item version of a Qwen-edit graph under the current table.""" + graph = graph if graph is not None else qwen_edit_graph(*PROMPTS) + return review_capture.inspection_review_version({"prompt": graph}) + + +def _retarget_unet_loader(doc): + """A table edit for a class the Qwen-edit graph uses (no prompt effect).""" + doc["facts"]["model_outputs"]["UNETLoader"] = "unet" + + +def _unrelated_encoder(doc): + """A table edit for a class the Qwen-edit graph does not use.""" + doc["encoders"].append( + { + "class": "UnrelatedTextEncode", + "slot": 0, + "strategy": "single", + "inputs": ["text"], + } + ) + + +def _texts(evidence): + return sorted( + c["positive"]["text"] + for c in evidence["candidates"] + if isinstance(c.get("positive"), dict) + ) + + +class InspectionTestCase(ReviewRoutesTestCase): + def _write(self, name=NAME, graph=None, color=(255, 0, 0)): + path = self.output_dir / name + write_png( + path, + prompt=graph if graph is not None else qwen_edit_graph(*PROMPTS), + color=color, + ) + return path + + def _pending(self): + return self.api.db.library_counts()["review_pending"] + + def _evidence(self, item): + return json.loads(item["evidence"]) + + def _table(self, edit): + """Point the inspection profile at an edited copy of the table.""" + doc = json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8")) + edit(doc) + folder = tempfile.mkdtemp(dir=self.tmpdir) + table = Path(folder) / "inspection-semantics.json" + table.write_text(json.dumps(doc), encoding="utf-8") + patcher = patch.object(inspection_semantics, "SEMANTICS_PATH", table) + patcher.start() + inspection_semantics.get_semantics.cache_clear() + self.addCleanup(inspection_semantics.get_semantics.cache_clear) + self.addCleanup(patcher.stop) + + async def _apply(self, positive): + item, detail, observed = await self._open() + candidate = self._candidate(detail, positive) + return item, await self._decide( + item["id"], self._body(detail, observed, candidate["hash"]) + ) + + def _linked_text(self, path): + row = ( + self.api.db.model.get_connection() + .execute( + "SELECT p.text FROM generated_images gi " + "JOIN prompts p ON p.id=gi.prompt_id WHERE gi.image_path=?", + (str(path),), + ) + .fetchone() + ) + return row[0] if row else None + + +class TestQwenEditBatchIsQueued(InspectionTestCase): + async def test_qwen_edit_batch_is_queued_with_its_texts(self): + self._roots() + graph = qwen_edit_graph(*PROMPTS, joined=[("a red scarf", "snow")]) + self._write(graph=graph) + + job = await self._scan() + + (item,) = self._items() + evidence = self._evidence(item) + self.assertEqual(_texts(evidence), sorted(PROMPTS + ("a red scarf, snow",))) + self.assertEqual(item["candidate_count"], 4) + self.assertEqual(evidence["profile"], "inspection") + self.assertEqual( + evidence["semantics"], review_capture.semantics_version({"prompt": graph}) + ) + self.assertEqual( + item["resolver_version"], + review_capture.inspection_review_version({"prompt": graph}), + ) + self.assertNotEqual(item["resolver_version"], review_capture.RESOLVER_VERSION) + self.assertEqual(job["scope"]["review_items"], 1) + (row,) = job["outcomes"]["review"] + self.assertEqual(row["path"], NAME) + self.assertIn("more than one prompt could match", row["detail"]) + # Review choices only: nothing linked the file. + self.assertIsNone(self._linked_text(self.output_dir / NAME)) + self.assertEqual(self._capture_files(), []) + + async def test_rescan_keeps_one_item_and_skips_the_copy(self): + self._roots() + self._write() + await self._scan() + (item,) = self._items() + + with _counted(self.api._captures, "capture") as copies: + job = await self._scan("all") + + self.assertEqual(copies, []) + self.assertEqual([i["id"] for i in self._items()], [item["id"]]) + self.assertEqual(job["scope"]["review_items"], 1) + self.assertEqual(job["scope"]["review_untraced"], 0) + + async def test_more_than_sixteen_qwen_samplers_make_a_limited_item(self): + self._roots() + self._write(graph=qwen_edit_graph(*(f"edit {n}" for n in range(17)))) + + await self._scan() + + (item,) = self._items() + evidence = self._evidence(item) + self.assertTrue(item["limited"]) + self.assertEqual(item["candidate_count"], 0) + self.assertEqual(evidence["candidate_total"], 17) + self.assertEqual(evidence["profile"], "inspection") + + async def test_evidence_over_the_cap_makes_a_limited_item(self): + self._roots() + self._write() + real = review_capture.canonical_evidence + + def capped(value): + if value.get("candidates"): + raise ValueError("evidence exceeds the bound") + return real(value) + + with patch.object(review_capture, "canonical_evidence", side_effect=capped): + await self._scan() + + (item,) = self._items() + self.assertTrue(item["limited"]) + self.assertEqual(self._evidence(item)["candidate_total"], 3) + self.assertEqual(self._evidence(item)["profile"], "inspection") + + +class TestDefaultProfileStillDecides(InspectionTestCase): + async def test_ambiguous_file_with_texts_keeps_the_default_profile(self): + """A default-ambiguous file with a choice never runs the inspection + profile, even when that profile would offer more candidates.""" + self._roots() + graph = { + **_prefixed(qwen_edit_graph("make it night"), "q"), + **unlinkable_graph("cat", "dog"), + } + self._write(graph=graph) + + with patch.object( + review_capture, + "resolve_inspection", + side_effect=AssertionError("inspection profile ran"), + ): + await self._scan() + + (item,) = self._items() + evidence = self._evidence(item) + self.assertNotIn("profile", evidence) + self.assertNotIn("semantics", evidence) + self.assertEqual(item["resolver_version"], review_capture.RESOLVER_VERSION) + self.assertEqual(_texts(evidence), ["cat", "dog"]) + + async def test_apply_of_a_default_item_never_runs_the_inspection_profile(self): + self._roots() + write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog")) + await self._scan() + with patch.object( + review_capture, + "resolve_inspection", + side_effect=AssertionError("inspection profile ran"), + ): + _item, (status, data) = await self._apply("cat") + self.assertEqual(status, 200, data) + self.assertEqual(self._linked_text(self.output_dir / "two.png"), "cat") + + +class TestApplyInspectionCandidate(InspectionTestCase): + async def test_apply_links_the_chosen_qwen_edit_text(self): + self._require_routes() + self._roots() + path = self._write() + await self._scan() + + item, (status, data) = await self._apply("add a red hat") + + self.assertEqual(status, 200, data) + self.assertEqual(data["item"]["state"], "selected") + self.assertEqual(data["total"], 0) + self.assertEqual(self._linked_text(path), "add a red hat") + links = self.api.db.get_image_prompt_pairs(data["pair"]["image_id"]) + (link,) = links + provenance = link["provenance"] + self.assertEqual(provenance["kind"], "user-selection") + self.assertEqual(provenance["resolver_version"], _version()) + self.assertEqual(provenance["positive"]["text"], "add a red hat") + (decision,) = self._decisions() + self.assertEqual(decision["item_id"], item["id"]) + self._no_paths() + + async def test_apply_rechecks_with_the_recorded_profile(self): + """The default profile would not offer the candidate: Apply must use + the profile the item recorded, and refuses when that profile no + longer offers it.""" + self._require_routes() + self._roots() + self._write() + await self._scan() + calls = [] + real = review_capture.resolve_inspection + + def spy(metadata): + calls.append(metadata) + return real(metadata) + + with patch.object(review_capture, "resolve_inspection", side_effect=spy): + _item, (status, data) = await self._apply("make it night") + + self.assertEqual(status, 200, data) + self.assertEqual(len(calls), 1) + + async def test_table_change_keeps_the_item_applicable(self): + """A table change for a class the graph uses moves the item's + version; the pending item stays applicable while its candidate is + still offered.""" + self._require_routes() + self._roots() + path = self._write() + await self._scan() + (old,) = self._items() + version = _version() + item, detail, observed = await self._open() + + self._table(_retarget_unet_loader) + self.assertNotEqual(_version(), version) + + candidate = self._candidate(detail, "add a red hat") + status, data = await self._decide( + item["id"], self._body(detail, observed, candidate["hash"]) + ) + self.assertEqual(status, 200, data) + self.assertEqual(item["id"], old["id"]) + self.assertEqual(self._linked_text(path), "add a red hat") + + async def test_table_change_that_drops_the_encoder_is_a_review_conflict(self): + self._require_routes() + self._roots() + self._write() + await self._scan() + item, detail, observed = await self._open() + candidate = self._candidate(detail, "add a red hat") + + def drop_encoder(doc): + doc["encoders"] = [ + entry + for entry in doc["encoders"] + if entry["class"] != "TextEncodeQwenImageEdit" + ] + + version = _version() + self._table(drop_encoder) + self.assertNotEqual(_version(), version) + before = self._snapshot() + status, data = await self._decide( + item["id"], self._body(detail, observed, candidate["hash"]) + ) + + self.assertEqual(status, 409, data) + self.assertEqual(data["code"], "review-conflict") + self.assertEqual(self._snapshot(), before) + # The rescan finds nothing to choose under the new version: the + # stale, undecided item leaves the queue instead of staying stuck. + job = await self._scan("all") + self.assertEqual(self._items(), []) + self.assertEqual(job["scope"]["review_untraced"], 1) + + +class TestMonitorNeverQueues(InspectionTestCase): + def _handler(self): + handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + handler.review_recorder = self.api.record_review_file + return handler + + async def test_monitor_pass_never_queues_but_the_next_scan_does(self): + self._roots() + path = self._write() + + self.assertEqual(self.api.record_review_file(str(path)), "skipped") + self.assertEqual(self._items(), []) + self.assertEqual(self._capture_files(), []) + + await self._scan() + (item,) = self._items() + self.assertEqual(self._evidence(item)["profile"], "inspection") + + async def test_monitor_keeps_the_inspection_item_of_an_unlinked_output(self): + """The monitor's untraced settle judges each item by its own + evidence: an inspection item offers a choice and stays.""" + self._roots() + path = self._write() + await self._scan() + (item,) = self._items() + # Other bytes, so the stat shortcut does not answer for the file. + self._write(color=(0, 0, 255)) + + self.assertEqual(self.api.record_review_file(str(path)), "skipped") + self.assertEqual([i["id"] for i in self._items()], [item["id"]]) + + async def test_monitor_link_removes_the_inspection_item(self): + self._roots() + path = self._write() + await self._scan() + self.assertEqual(len(self._items()), 1) + prompt_id = self.api.db.save_prompt( + text="latest", prompt_hash=generate_prompt_hash("latest") + ) + + self._handler().link_image_to_prompt( + str(path), + {"id": prompt_id, "text": "latest", "fallback": True}, + {"prompt": qwen_edit_graph(*PROMPTS)}, + ) + + self.assertEqual(self._items(), []) + self.assertEqual(os.path.basename(str(path)), NAME) + + +class TestTableVersion(InspectionTestCase): + def test_table_version_follows_content_not_layout(self): + doc = {"schema": 1, "encoders": [{"class": "A", "slot": 0}]} + same = json.loads(json.dumps(doc, indent=4)) + other = {"schema": 1, "encoders": [{"class": "B", "slot": 0}]} + + version = inspection_semantics.table_version(doc) + + self.assertRegex(version, "^[0-9a-f]{64}$") + self.assertEqual(inspection_semantics.table_version(same), version) + self.assertNotEqual(inspection_semantics.table_version(other), version) + self.assertEqual( + inspection_semantics.load_semantics().version, + inspection_semantics.table_version( + json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8")) + ), + ) + + async def test_an_unusable_table_queues_nothing(self): + self._roots() + self._write() + unusable = inspection_semantics.SemanticsError("bad table") + with ( + patch.object(inspection_semantics, "get_semantics", side_effect=unusable), + patch.object(review_capture, "get_semantics", side_effect=unusable), + ): + self.assertIsNone(review_capture.inspection_review_version()) + self.assertIsNone(_version()) + job = await self._scan() + + self.assertEqual(self._items(), []) + (row,) = job["outcomes"]["review"] + self.assertIn("could not be traced", row["detail"]) + + async def test_an_unusable_table_keeps_existing_items_on_rescan(self): + self._require_routes() + self._roots() + self._write() + await self._scan() + (item,) = self._items() + + unusable = inspection_semantics.SemanticsError("bad table") + with ( + patch.object(inspection_semantics, "get_semantics", side_effect=unusable), + patch.object(review_capture, "get_semantics", side_effect=unusable), + ): + await self._scan("all") + + self.assertEqual([i["id"] for i in self._items()], [item["id"]]) + + +class TestVersionChanges(InspectionTestCase): + """A version change never leaves two pending items for one file.""" + + async def test_table_change_for_a_graph_class_leaves_one_item(self): + self._require_routes() + self._roots() + path = self._write() + await self._scan() + (old,) = self._items() + + self._table(_retarget_unet_loader) + job = await self._scan("all") + + (item,) = self._items() + self.assertNotEqual(item["id"], old["id"]) + self.assertEqual(item["resolver_version"], _version()) + self.assertNotEqual(item["resolver_version"], old["resolver_version"]) + self.assertEqual(job["scope"]["review_items"], 1) + self.assertEqual(self._pending(), 1) + _item, (status, data) = await self._apply("make it night") + self.assertEqual(status, 200, data) + self.assertEqual(self._linked_text(path), "make it night") + + async def test_unrelated_table_edit_keeps_the_item_and_does_no_work(self): + self._roots() + self._write() + await self._scan() + (old,) = self._items() + version = _version() + + self._table(_unrelated_encoder) + self.assertEqual(_version(), version) + self.assertNotEqual( + inspection_semantics.get_semantics().version, + json.loads(old["evidence"])["semantics"], + ) + with ( + _counted(self.api._captures, "capture") as copies, + _counted(review_capture, "resolve_inspection") as calls, + ): + await self._scan("all") + + self.assertEqual(copies, []) + self.assertEqual([i["id"] for i in self._items()], [old["id"]]) + self.assertEqual(calls, []) + + async def test_resolver_change_leaves_one_default_item(self): + self._roots() + write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog")) + await self._scan() + (old,) = self._items() + self.assertEqual(old["resolver_version"], 1) + + with patch.object(review_capture, "RESOLVER_VERSION", 2): + await self._scan("all") + + (item,) = self._items() + self.assertNotEqual(item["id"], old["id"]) + self.assertEqual(item["resolver_version"], 2) + self.assertEqual(self._pending(), 1) + + async def test_decided_old_item_is_kept_and_no_duplicate_is_made(self): + self._require_routes() + self._roots() + self._write() + await self._scan() + item, detail, observed = await self._open() + status, data = await self._decide( + item["id"], self._body(detail, observed, "unresolved") + ) + self.assertEqual(status, 200, data) + before = (self._items(), self._decisions()) + + self._table(_retarget_unet_loader) + await self._scan("all") + self.assertEqual((self._items(), self._decisions()), before) + # The decided bytes are not copied again on the next rescan. + with _counted(self.api._captures, "capture") as copies: + await self._scan("all") + + self.assertEqual(copies, []) + self.assertEqual((self._items(), self._decisions()), before) + self.assertEqual(self._pending(), 0) + + async def test_advanced_pending_item_is_kept_and_no_duplicate_is_made(self): + self._roots() + self._write() + await self._scan() + (old,) = self._items() + conn = self.api.db.model.get_connection() + with conn: + conn.execute( + "UPDATE media_review_items SET revision=1 WHERE id=?", (old["id"],) + ) + before = self._items() + + self._table(_retarget_unet_loader) + await self._scan("all") + + self.assertEqual(self._items(), before) + + +class TestInspectionRuns(InspectionTestCase): + """The heavier profile runs once per new file state, never again.""" + + async def test_new_file_is_inspected_once_and_not_on_rescan(self): + self._roots() + self._write() + with _counted(review_capture, "resolve_inspection") as first: + await self._scan() + self.assertEqual(len(self._items()), 1) + self.assertEqual(len(first), 1) + + with _counted(review_capture, "resolve_inspection") as again: + await self._scan("all") + + self.assertEqual(again, []) + self.assertEqual(len(self._items()), 1) + + async def test_untraced_file_is_not_inspected_again_until_it_changes(self): + self._roots() + graph = qwen_edit_graph(*PROMPTS) + for node in graph.values(): + if node["class_type"] == "TextEncodeQwenImageEdit": + node["class_type"] = "OpaqueTextEncode" + self._write(graph=graph) + with _counted(review_capture, "resolve_inspection") as first: + job = await self._scan() + self.assertEqual(len(first), 1) + (row,) = job["outcomes"]["review"] + self.assertIn("could not be traced", row["detail"]) + + with _counted(review_capture, "resolve_inspection") as again: + await self._scan("all") + self.assertEqual(again, []) + + self._write(graph=graph, color=(0, 255, 0)) + with _counted(review_capture, "resolve_inspection") as changed: + await self._scan("all") + self.assertEqual(len(changed), 1) + self.assertEqual(self._items(), []) + + async def test_capture_with_other_records_is_inspected_again(self): + """The scan's result is reused only for identical records.""" + self._roots() + self._write() + real = review_capture.captured_review + + def edited(store, capture, profile=review_capture.DEFAULT): + metadata, resolution = real(store, capture, profile) + graph = json.loads(json.dumps(metadata["prompt"])) + graph["100"]["inputs"]["prompt"] = "a captured edit" + return {**metadata, "prompt": graph}, resolution + + with ( + patch.object(review_capture, "captured_review", side_effect=edited), + _counted(review_capture, "resolve_inspection") as calls, + ): + await self._scan() + + self.assertEqual(len(calls), 2) + (item,) = self._items() + self.assertIn("a captured edit", _texts(self._evidence(item))) + + +class TestClassProjection(unittest.TestCase): + def test_projection_over_every_named_class_is_the_whole_table(self): + doc = json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8")) + names = set(inspection_semantics.load_semantics().all_classes) + names |= set(doc["facts"]["configuration_outputs"]) + names |= set(doc["facts"]["model_outputs"]) + names |= set(doc["facts"]["model_passthroughs"]) + names |= set(doc["facts"].get("scalar_outputs", {})) + names |= {entry["class"] for entry in doc["lineage"]} + + self.assertEqual(inspection_semantics.class_projection(doc, names), doc) + + def test_class_version_moves_only_for_the_classes_it_covers(self): + doc = json.loads(inspection_semantics.SEMANTICS_PATH.read_text("utf-8")) + classes = frozenset({"KSampler", "UNETLoader"}) + version = inspection_semantics.class_version(doc, classes) + _unrelated_encoder(doc) + self.assertEqual(inspection_semantics.class_version(doc, classes), version) + _retarget_unet_loader(doc) + self.assertNotEqual(inspection_semantics.class_version(doc, classes), version) + + +class TestRecordingEdges(InspectionTestCase): + async def test_store_refuses_a_new_version_beside_a_decided_item(self): + self._roots() + self._write() + await self._scan() + (old,) = self._items() + conn = self.api.db.model.get_connection() + with conn: + conn.execute( + "UPDATE media_review_items SET state='unresolved', revision=1 " + "WHERE id=?", + (old["id"],), + ) + evidence = {**self._evidence(old), "resolver_version": 7} + + inserted = self.api.db.record_review_item( + locator=old["locator"], + spelled_path=old["spelled_path"], + sha256=old["sha256"], + size=old["size"], + mtime_ns=old["mtime_ns"], + epoch=self.api.db.get_execution_epoch(), + resolver_version=7, + evidence=evidence, + candidate_count=old["candidate_count"], + limited=bool(old["limited"]), + ) + + self.assertFalse(inserted) + self.assertEqual([i["id"] for i in self._items()], [old["id"]]) + + async def test_already_linked_file_with_an_inspection_choice_is_unchanged(self): + self._roots() + path = self._write() + prompt_id = self.api.db.save_prompt( + text="latest", prompt_hash=generate_prompt_hash("latest") + ) + handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + handler.link_image_to_prompt( + str(path), + {"id": prompt_id, "text": "latest", "fallback": True}, + {"prompt": qwen_edit_graph(*PROMPTS)}, + ) + + job = await self._scan("all") + + self.assertEqual(self._items(), []) + self.assertEqual(job["outcomes"]["review"], []) + (row,) = job["outcomes"]["known"] + self.assertIn("Already linked", row["detail"]) + self.assertEqual(self._linked_text(path), "latest") + + async def test_untraced_memory_forgets_the_oldest_state(self): + from py.api import review_recording + + self._roots() + graph = qwen_edit_graph(*PROMPTS) + for node in graph.values(): + if node["class_type"] == "TextEncodeQwenImageEdit": + node["class_type"] = "OpaqueTextEncode" + self._write("a.png", graph=graph) + self._write("b.png", graph=graph) + with patch.object(review_recording, "MAX_UNTRACED_MEMORY", 1): + await self._scan() + with _counted(review_capture, "resolve_inspection") as calls: + await self._scan("all") + + # One remembered state: each file evicts the other, so both run. + self.assertEqual(len(calls), 2) + self.assertEqual(len(self.api._untraced_inspections), 1) + + def test_graph_classes(self): + self.assertEqual(review_capture.graph_classes({"parameters": "x"}), set()) + self.assertEqual( + review_capture.graph_classes({"prompt": {"1": {"class_type": "A"}}}), + {"A"}, + ) + many = { + str(n): {"class_type": f"C{n}"} + for n in range(review_capture.MAX_FINGERPRINT_CLASSES + 1) + } + self.assertIsNone(review_capture.graph_classes({"prompt": many})) + self.assertEqual( + review_capture.semantics_version({"prompt": many}), + inspection_semantics.get_semantics().version, + ) + + +def _move_qwen_encoder_slot(doc): + """An edit of the Qwen-edit encoder's own table entry.""" + for entry in doc["encoders"]: + if entry["class"] == "TextEncodeQwenImageEdit": + entry["slot"] = 1 + + +def _unused_sampler(doc): + """A new dict-keyed sampler class the Qwen-edit graph does not use.""" + doc["samplers"]["UnusedSampler"] = dict(doc["samplers"]["KSampler"]) + + +class TestFingerprintCoverage(InspectionTestCase): + def test_editing_the_encoders_entry_moves_the_version(self): + version = _version() + self._table(_move_qwen_encoder_slot) + self.assertIsNotNone(_version()) + self.assertNotEqual(_version(), version) + + def test_an_unused_dict_keyed_class_leaves_the_version(self): + version = _version() + table = inspection_semantics.get_semantics().version + self._table(_unused_sampler) + self.assertNotEqual(inspection_semantics.get_semantics().version, table) + self.assertEqual(_version(), version) + + +class TestInspectionResolverVersion(InspectionTestCase): + async def test_inspection_resolver_change_supersedes_the_item(self): + """An inspection-only resolver change moves the item version, so the + rescan re-inspects and the new item is applicable (never a stuck + 409).""" + self._require_routes() + self._roots() + path = self._write() + await self._scan() + (old,) = self._items() + version = _version() + real = review_capture.resolve_inspection + + def newer(metadata): + resolution = real(metadata) + return { + **resolution, + "candidates": [ + {**c, "reason": "a newer inspection resolver"} + for c in resolution["candidates"] + ], + } + + with ( + patch.object(review_capture, "INSPECTION_RESOLVER_VERSION", 2), + patch.object(review_capture, "resolve_inspection", side_effect=newer), + ): + self.assertNotEqual(_version(), version) + await self._scan("all") + (item,) = self._items() + self.assertNotEqual(item["id"], old["id"]) + _item, (status, data) = await self._apply("add a red hat") + + self.assertEqual(status, 200, data) + self.assertEqual(self._linked_text(path), "add a red hat") + + def test_profile_budgets_are_folded_into_the_version(self): + from dataclasses import replace + + version = _version() + wider = replace( + review_capture.INSPECTION_PROFILE, + total_visits=review_capture.INSPECTION_PROFILE.total_visits + 1, + ) + with patch.object(review_capture, "INSPECTION_PROFILE", wider): + self.assertNotEqual(_version(), version) + + +class TestScanContainment(InspectionTestCase): + async def test_an_inspection_failure_is_counted_and_the_scan_continues(self): + self._roots() + self._write("a.png") + self._write("b.png", graph=qwen_edit_graph("snow", "rain")) + real = review_capture.resolve_inspection + + def fails_for_a(metadata): + if "make it night" in json.dumps(metadata): + raise RuntimeError("inspection blew up") + return real(metadata) + + with ( + patch.object(review_capture, "resolve_inspection", side_effect=fails_for_a), + self.assertLogs("prompt_manager.api", "WARNING") as logs, + ): + job = await self._scan() + + self.assertEqual(job["state"], "finished") + (item,) = self._items() + self.assertEqual(os.path.basename(item["locator"]), "b.png") + self.assertEqual(job["scope"]["review_errors"], 1) + rows = {row["path"]: row for row in job["outcomes"]["review"]} + self.assertIn("could not be traced", rows["a.png"]["detail"]) + self.assertIn("more than one prompt could match", rows["b.png"]["detail"]) + self.assertTrue(any("a.png" in line for line in logs.output)) + + +class TestUntracedMemoryKeys(InspectionTestCase): + async def test_untraced_memory_is_keyed_by_a_16_byte_digest(self): + self._roots() + graph = qwen_edit_graph(*PROMPTS) + for node in graph.values(): + if node["class_type"] == "TextEncodeQwenImageEdit": + node["class_type"] = "OpaqueTextEncode" + self._write(graph=graph) + + await self._scan() + + (key,) = self.api._untraced_inspections + self.assertIsInstance(key, bytes) + self.assertEqual(len(key), 16) + + +class TestMonitorNeverRecords(InspectionTestCase): + async def test_monitor_never_records_an_unlinked_ambiguous_output(self): + """Only scans add items (maintainer ruling); the monitor only clears + the items of outputs it links.""" + self._roots() + path = self.output_dir / "two.png" + write_png(path, prompt=unlinkable_graph("cat", "dog")) + + with _counted(self.api._captures, "capture") as copies: + self.assertEqual(self.api.record_review_file(str(path)), "skipped") + + self.assertEqual(copies, []) + self.assertEqual(self._items(), []) + await self._scan() + self.assertEqual(len(self._items()), 1) + + +class TestStaleSettleContainment(InspectionTestCase): + async def test_a_failed_stale_settle_is_logged_and_counted(self): + import sqlite3 + + self._roots() + graph = qwen_edit_graph(*PROMPTS) + for node in graph.values(): + if node["class_type"] == "TextEncodeQwenImageEdit": + node["class_type"] = "OpaqueTextEncode" + self._write(graph=graph) + with ( + patch.object( + self.api.db, + "settle_stale_review_items", + side_effect=sqlite3.OperationalError("database is locked"), + ), + self.assertLogs("prompt_manager.api", "WARNING") as logs, + ): + job = await self._scan() + + self.assertEqual(job["state"], "finished") + self.assertEqual(job["scope"]["review_errors"], 1) + self.assertTrue(any("Stale Needs review" in line for line in logs.output)) diff --git a/tests/test_review_outcome.py b/tests/test_review_outcome.py index c4952c8..b24a524 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/inspection_semantics.py b/utils/inspection_semantics.py index be0dce9..f334ade 100644 --- a/utils/inspection_semantics.py +++ b/utils/inspection_semantics.py @@ -16,10 +16,21 @@ nesting before it is validated. Every class appears in exactly one prompt category; classes that are not listed stay ``unsupported-node``. ``lineage`` (image and latent pass-through outputs for multi-stage chains) describes no prompt, so its classes are not claimed by a category. + +``Semantics.version`` is the SHA-256 of the table's canonical JSON (sorted +keys, no whitespace): any change of content, never of layout, moves it. +``class_version(doc, classes)`` is the narrower fingerprint Needs review +folds into the version of the items it records from the inspection profile +(``utils.review_capture``): the SHA-256 of ``class_projection``, the table +restricted to what it says about ``classes`` (the node classes of one graph) +plus the class-independent ``schema`` and ``stage_inputs``. The resolver looks +the table up only by the class of a graph node, so an edit about other +classes cannot change how that graph resolves and leaves the version alone. """ +import hashlib import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from functools import lru_cache from pathlib import Path from types import MappingProxyType @@ -118,6 +129,10 @@ class Semantics: stage_inputs: Mapping[str, Tuple[str, ...]] facts: Mapping[str, Any] all_classes: FrozenSet[str] + # SHA-256 of the canonical table JSON (set by ``load_semantics``). + version: str = "" + # The canonical table JSON itself, for ``class_version``. + source: str = "" def _freeze(value: Any) -> Frozen: @@ -521,7 +536,71 @@ def load_semantics(path: Optional[Union[str, Path]] = None) -> Semantics: except (OSError, UnicodeDecodeError, ValueError) as error: raise SemanticsError(f"cannot read semantics table {source.name}: {error}") _check_nesting(doc) - return _validate(doc) + return replace( + _validate(doc), version=table_version(doc), source=_canonical_json(doc) + ) + + +# Table parts that hold no class names: they apply to every graph. +_CLASS_INDEPENDENT = frozenset({"schema", "stage_inputs"}) + + +def _canonical_json(doc: Any) -> str: + return json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def table_version(doc: Any) -> str: + """SHA-256 of ``doc`` as canonical JSON: content, not layout.""" + return hashlib.sha256(_canonical_json(doc).encode("utf-8")).hexdigest() + + +def _about(value: Any, classes: FrozenSet[str]) -> Any: + """The part of one category that names a class in ``classes``. + + Categories are keyed by class (objects), list classes (strings) or list + entries that carry a ``class``. + """ + if isinstance(value, dict): + return {key: item for key, item in value.items() if key in classes} + if isinstance(value, list): + return [ + entry + for entry in value + if (isinstance(entry, str) and entry in classes) + or (isinstance(entry, dict) and entry.get("class") in classes) + ] + return value + + +def class_projection(doc: Mapping[str, Any], classes) -> dict: + """What ``doc`` says about ``classes``, plus its class-independent parts.""" + classes = frozenset(classes) + projected = {} + for category, value in doc.items(): + if category in _CLASS_INDEPENDENT: + projected[category] = value + elif category == "facts" and isinstance(value, dict): + projected[category] = { + table: _about(entries, classes) for table, entries in value.items() + } + else: + projected[category] = _about(value, classes) + return projected + + +def class_version(doc: Mapping[str, Any], classes) -> str: + """SHA-256 of ``class_projection(doc, classes)`` as canonical JSON.""" + return table_version(class_projection(doc, classes)) + + +@lru_cache(maxsize=256) +def _cached_class_version(source: str, classes: FrozenSet[str]) -> str: + return class_version(json.loads(source), classes) + + +def semantics_class_version(semantics: Semantics, classes) -> str: + """``class_version`` of a loaded table (cached per table and class set).""" + return _cached_class_version(semantics.source, frozenset(classes)) @lru_cache(maxsize=1) diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index 4ad10a5..6eb10be 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 f26a46b..1859dee 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 @@ -21,6 +22,24 @@ candidate's ``hash`` is the SHA-256 of its canonical JSON without the hash. More than ``MAX_CANDIDATES`` candidates or evidence over the 8 MiB bound make a ``limited`` item without candidates (a result without any candidate is not ``reviewable`` and is never stored). + +Inspection-profile choices (ruling a, after the maintainer's Qwen image-edit +batches): when the default profile is ambiguous without a choice, the scan +(``inspection=True``) asks ``INSPECTION_PROFILE``, +which knows more encoder and string nodes. With at least two choosable +candidates there, the item is recorded from that profile: its evidence names +``profile: "inspection"`` and ``semantics``, the fingerprint of what the table +says about the graph's node classes, and its ``resolver_version`` is +``inspection_review_version(metadata)``, which folds the profile, +``RESOLVER_VERSION`` and that fingerprint together. A table edit about those +classes moves the version the way a resolver change does (I2); an edit about +other classes moves nothing. A pending item stays applicable while the bytes +still offer its chosen candidate, and a rescan's new item supersedes it only +while it is pending and undecided (never beside a decision, see +``PromptDatabase.record_review_item``). Apply re-resolves with the +profile the item recorded (``captured_review(..., item_profile(evidence))``). +The inspection profile only ever produces review choices; ingestion's own +links stay on the default profile, which never loads the table (D1). """ from __future__ import annotations @@ -29,6 +48,7 @@ import hashlib import json import logging import os +from dataclasses import astuple from pathlib import Path from typing import Any @@ -40,9 +60,19 @@ try: canonical_evidence, choosable, ) + from .inspection_semantics import ( + SemanticsError, + get_semantics, + semantics_class_version, + ) from .media_capture import CaptureCapacity, CaptureError from .prompt_inspector import MetadataReadError, read_inspection_file - from .prompt_pair_resolver import RESOLVER_VERSION, resolve_prompt_pair + from .prompt_pair_resolver import ( + INSPECTION_PROFILE, + INSPECTION_RESOLVER_VERSION, + RESOLVER_VERSION, + resolve_prompt_pair, + ) except ImportError: # imported with the repo root on sys.path from database.library_commands import LibraryCommandError from database.review_schema import ( @@ -51,9 +81,19 @@ except ImportError: # imported with the repo root on sys.path canonical_evidence, choosable, ) + from utils.inspection_semantics import ( + SemanticsError, + get_semantics, + semantics_class_version, + ) from utils.media_capture import CaptureCapacity, CaptureError from utils.prompt_inspector import MetadataReadError, read_inspection_file - from utils.prompt_pair_resolver import RESOLVER_VERSION, resolve_prompt_pair + from utils.prompt_pair_resolver import ( + INSPECTION_PROFILE, + INSPECTION_RESOLVER_VERSION, + RESOLVER_VERSION, + resolve_prompt_pair, + ) logger = logging.getLogger("prompt_manager.review_capture") @@ -61,11 +101,136 @@ logger = logging.getLogger("prompt_manager.review_capture") REVIEW_SCOPE = "needs-review" # Nested ambiguous stages are flattened at most this deep. MAX_CANDIDATE_DEPTH = 8 +# The resolver profile an item's evidence was built with (``profile``). +DEFAULT = "default" +INSPECTION = "inspection" +# Inspection item versions: above every plausible ``RESOLVER_VERSION`` and, +# with 40 bits of the folded digest, below 2**53 (exact as a JSON number). +INSPECTION_VERSION_BASE = 1 << 40 +_INSPECTION_VERSION_HEX = 10 -def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> str: +# Above this many distinct node classes the whole table is the fingerprint. +MAX_FINGERPRINT_CLASSES = 1024 + + +def graph_classes(metadata: Any) -> frozenset | None: + """The node classes of ``metadata``'s prompt graph (empty without one). + + None when the graph names more than ``MAX_FINGERPRINT_CLASSES`` classes. + """ + graph = metadata.get("prompt") if isinstance(metadata, dict) else None + if not isinstance(graph, dict): + return frozenset() + classes = set() + for node in graph.values(): + name = node.get("class_type") if isinstance(node, dict) else None + if isinstance(name, str): + classes.add(name) + if len(classes) > MAX_FINGERPRINT_CLASSES: + return None + return frozenset(classes) + + +def semantics_version(metadata: Any = None) -> str | None: + """The semantics table fingerprint that ``metadata``'s graph depends on. + + The table restricted to the graph's node classes + (``inspection_semantics.class_version``), so an edit about other classes + leaves it alone; the whole table's version without ``metadata`` or for a + graph of very many classes. None when the table is unusable. + """ + try: + semantics = get_semantics() + except SemanticsError: + return None + classes = graph_classes(metadata) if metadata is not None else None + if classes is None: + return semantics.version + return semantics_class_version(semantics, classes) + + +def inspection_review_version(metadata: Any = None) -> int | None: + """The ``resolver_version`` of an item recorded from the inspection profile. + + Folds the profile with its budgets, ``RESOLVER_VERSION``, + ``INSPECTION_RESOLVER_VERSION`` and ``semantics_version(metadata)``, so + a change of either resolver, of the profile's bounds or of the table + about the graph's classes moves it. None when the table cannot be loaded. + """ + table = semantics_version(metadata) + if table is None: + return None + profile = repr(astuple(INSPECTION_PROFILE)) + seed = f"{profile}:{RESOLVER_VERSION}:{INSPECTION_RESOLVER_VERSION}:{table}" + folded = hashlib.sha256(seed.encode("utf-8")).hexdigest() + return INSPECTION_VERSION_BASE + int(folded[:_INSPECTION_VERSION_HEX], 16) + + +def resolve_inspection(metadata: dict) -> dict: + """``metadata`` resolved with the inspection profile (bounded work).""" + return resolve_prompt_pair(metadata, profile=INSPECTION_PROFILE) + + +def inspection_choice(metadata: dict) -> dict | None: + """The inspection profile's ambiguous result when it offers a choice. + + The scan asks only for a file the default profile found ambiguous + without a choice that nothing links. + """ + if inspection_review_version(metadata) is None: + return None + resolution = resolve_inspection(metadata) + if resolution.get("state") != "ambiguous" or not reviewable(resolution): + return None + return resolution + + +def item_profile(evidence: Any) -> str: + """The profile an item's stored evidence was built with.""" + if isinstance(evidence, dict) and evidence.get("profile") == INSPECTION: + return INSPECTION + return DEFAULT + + +def item_versions(metadata: Any = None) -> tuple: + """The item versions a file with ``metadata`` may already be queued under. + + Without ``metadata``, only the ingestion version. + """ + version = inspection_review_version(metadata) if metadata is not None else None + return (RESOLVER_VERSION,) if version is None else (RESOLVER_VERSION, version) + + +def _same_records(captured: dict, scanned: dict) -> bool: + """True when the capture holds the records the scan resolved.""" + keys = (set(captured) | set(scanned)) - {"file_info"} + return all(captured.get(key) == scanned.get(key) for key in keys) + + +def record_ambiguous_file( + db, + store, + root, + root_index: int, + relative: str, + *, + inspection: bool = False, + scanned: tuple | None = None, +) -> str: """Persist the ambiguous evidence of one rooted file. + With ``inspection`` (the scan), a default result without a choice is + resolved again with the inspection profile, and recorded from it when + that offers a choice. ``scanned`` is the + scan's ``(metadata, inspection_resolution_or_None)`` for the file: its + metadata names the item versions that count, and when the capture holds + the same records its resolution is reused instead of resolving again. + + A new item supersedes the file's other items of the same bytes (an older + resolver or table version) when every one is still pending and + undecided; when one carries a decision, nothing new is stored. + Returns ``recorded`` (new item with candidates), ``limited`` (new item without candidates), ``duplicate`` (these bytes already carry this evidence, or the file's current stat is the verified fingerprint of @@ -79,9 +244,10 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st ``CaptureCapacity`` and database errors propagate to the caller, which logs and counts them. """ - spelled_path, queued = _queued_spelling(db, root, relative) + versions = item_versions(scanned[0] if scanned is not None else None) + spelled_path, queued = _queued_spelling(db, root, relative, versions) if queued is not None: - return _already_queued(db, *queued) + return _already_queued(db, *queued, versions) epoch = db.get_execution_epoch() try: # Background: the last worker stays free for an interactive request. @@ -105,9 +271,26 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st resolution = captured[1] if captured is not None else None if resolution is None or resolution.get("state") != "ambiguous": return "skipped" + profile, version = DEFAULT, RESOLVER_VERSION if not reviewable(resolution): - return "untraced" - evidence, count, limited = review_evidence(resolution) + version = inspection_review_version(captured[0]) if inspection else None + if version is None: + return "untraced" + profile = INSPECTION + if ( + scanned is not None + and scanned[1] is not None + and _same_records(captured[0], scanned[0]) + ): + resolution = scanned[1] # the same records: no second run + else: + # Heavier, so only for a file ambiguous with nothing to choose. + resolution = resolve_inspection(captured[0]) + if resolution.get("state") != "ambiguous" or not reviewable(resolution): + return "untraced" + evidence, count, limited = review_evidence( + resolution, profile, version, captured[0] + ) try: inserted = db.record_review_item( locator=capture.locator, @@ -116,7 +299,7 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st size=capture.size, mtime_ns=capture.mtime_ns, epoch=capture.epoch, - resolver_version=RESOLVER_VERSION, + resolver_version=version, evidence=evidence, candidate_count=count, limited=limited, @@ -124,18 +307,21 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st except LibraryCommandError: return "skipped" if not inserted: - return _already_queued(db, capture.locator, capture.size, capture.mtime_ns) + return _already_queued( + db, capture.locator, capture.size, capture.mtime_ns, versions + ) return "limited" if limited else "recorded" -def _already_queued(db, locator: str, size: int, mtime_ns: int) -> str: - """``duplicate`` while the existing item is pending, else ``decided``.""" - if db.review_pending_at(locator, size, mtime_ns, RESOLVER_VERSION): - return "duplicate" +def _already_queued(db, locator: str, size: int, mtime_ns: int, versions) -> str: + """``duplicate`` while an existing item is pending, else ``decided``.""" + for version in versions: + if db.review_pending_at(locator, size, mtime_ns, version): + return "duplicate" return "decided" -def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]: +def _queued_spelling(db, root, relative: str, versions) -> tuple[str, tuple | None]: """The file's spelling, and ``(locator, size, mtime_ns)`` when queued. The second value is None unless the file's current stat is already @@ -147,7 +333,9 @@ def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]: an item already cites is not copied or hashed again (maintainer ruling: the rescan stays cheap and the file stays in the default scope); its items take the current spelling. A stat failure leaves the decision to - the capture. + the capture. ``versions`` are the item versions that count; bytes whose + item carries a decision count under any version (a new version never + adds an item beside a decision). """ spelled = os.path.join(os.path.realpath(os.fspath(root)), *relative.split("/")) try: @@ -155,8 +343,9 @@ def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]: except OSError: return spelled, None locator = os.path.normcase(spelled) - if db.review_fingerprint_queued( - locator, facts.st_size, facts.st_mtime_ns, RESOLVER_VERSION + if db.review_decided_at(locator, facts.st_size, facts.st_mtime_ns) or any( + db.review_fingerprint_queued(locator, facts.st_size, facts.st_mtime_ns, v) + for v in versions ): # Not copied again, but a case-only rename keeps the stat: record # the current spelling so the item's target still opens. @@ -165,12 +354,14 @@ def _queued_spelling(db, root, relative: str) -> tuple[str, tuple | None]: return spelled, None -def captured_review(store, capture) -> tuple[dict, dict] | None: +def captured_review(store, capture, profile: str = DEFAULT) -> tuple[dict, dict] | None: """``(metadata, resolution)`` of the capture's own bytes, else None. ``metadata`` is what the scan hands the ingester: prompt and workflow JSON-decoded and ``file_info`` (size, dimensions, format) from the - capture. None when the capture is gone or carries no prompt metadata. + capture. ``resolution`` comes from ``profile`` (``DEFAULT`` or + ``INSPECTION``). None when the capture is gone or carries no prompt + metadata. """ if store.acquire(capture.token) is None: return None @@ -197,6 +388,8 @@ def captured_review(store, capture) -> tuple[dict, dict] | None: if read.format is not None: info["format"] = read.format canonical["file_info"] = info + if profile == INSPECTION: + return canonical, resolve_inspection(canonical) return canonical, resolve_prompt_pair(canonical) @@ -218,12 +411,25 @@ def reviewable(resolution: Any) -> bool: return False -def review_evidence(resolution: dict) -> tuple[dict, int, bool]: - """``(evidence, candidate_count, limited)`` for one ambiguous result.""" +def review_evidence( + resolution: dict, + profile: str = DEFAULT, + version: int | None = None, + metadata: Any = None, +) -> tuple[dict, int, bool]: + """``(evidence, candidate_count, limited)`` for one ambiguous result. + + Inspection evidence also names its ``profile`` and ``semantics``, the + table fingerprint of ``metadata``'s graph it was resolved with; + ``version`` is the item's version. + """ base = { - "resolver_version": RESOLVER_VERSION, + "resolver_version": RESOLVER_VERSION if version is None else version, "reason": _text(resolution.get("reason")), } + if profile == INSPECTION: + base["profile"] = INSPECTION + base["semantics"] = semantics_version(metadata) candidates: list[dict] = [] seen: set[str] = set() total = 0