diff --git a/TESTING.md b/TESTING.md index 43c93009..3474e076 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2622,18 +2622,28 @@ 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 table's -`version`, SHA-256 of its canonical JSON); the item's `resolver_version` is -`inspection_review_version()`, 2^40 plus 40 bits of SHA-256 over the profile, -`RESOLVER_VERSION` and the table version. A table change therefore behaves like -a resolver change (I2): a rescan records the new evidence beside the pending -item, which stays applicable while the bytes still offer its chosen candidate. +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, `RESOLVER_VERSION` 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 (twice per new file: once on the scanned metadata, once on the -capture; a rescan's stat shortcut skips the capture). The 16-candidate limited +a choice, and once per file state: the capture reuses the scan's result when +it holds the same records; a rescan skips files already queued or decided at +their current size and `mtime_ns` under their version, and files this process +already found untraced at the same state (an in-memory map of at most 65,536 +files, oldest forgotten first). The 16-candidate limited 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 diff --git a/database/review.py b/database/review.py index 51321d8f..7f9a25bd 100644 --- a/database/review.py +++ b/database/review.py @@ -90,6 +90,23 @@ _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 one file: the candidates a link settles. _PENDING_OF_LOCATOR = ( f"SELECT i.id FROM {FILES_TABLE} f " # nosec B608 @@ -274,6 +291,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 +320,15 @@ class ReviewOperations: ) if created or moved: conn.execute(_BUMP_IMAGE_INDEX) + 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 +346,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 +421,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: diff --git a/py/api/admin.py b/py/api/admin.py index 5b7cb5b5..4b2e9c76 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -45,7 +45,7 @@ try: parse_page_json, ) from ...utils.prompt_pair_resolver import resolve_prompt_pair - from ...utils.review_capture import inspection_offers_choice, reviewable + from ...utils.review_capture import reviewable except ImportError: from database.operations import PromptDatabase from database.pair_ingestion import ingest_resolved_pair @@ -56,7 +56,7 @@ except ImportError: parse_page_json, ) from utils.prompt_pair_resolver import resolve_prompt_pair - from utils.review_capture import inspection_offers_choice, reviewable + from utils.review_capture import reviewable try: @@ -252,22 +252,6 @@ def _file_size(path): return None -def _untraced_outcome(media_file, canonical, ambiguous, untraced): - """The status of an unlinked file the default profile found ambiguous - without a choice: ``"ambiguous"`` when the inspection profile offers one - (the file moves from ``untraced`` to ``ambiguous``), else ``"unresolved"``. - - Asked only when ``ambiguous`` is a list (the scan, which records review - items), so other callers never pay for the heavier profile. - """ - if ambiguous is None or not inspection_offers_choice(canonical): - return "unresolved" - if untraced is not None and media_file in untraced: - untraced.remove(media_file) - ambiguous.append(media_file) - return "ambiguous" - - def _with_file_size(canonical, media_file): """Record the file's size in ``file_info`` unless the metadata carries one. @@ -1662,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, @@ -1670,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}") @@ -1682,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 @@ -1717,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. @@ -1730,10 +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``, - unless the inspection profile offers a choice (only asked when - ``ambiguous`` is a list, so only by the scan): then it is - ``"ambiguous"`` and goes to ``ambiguous``. That profile never links. + ``"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 @@ -1754,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": @@ -1775,13 +1771,9 @@ class AdminRoutesMixin: # Graphs without a recognized sampler keep the legacy extractor below. prompt_text = self._graph_prompt_text(metadata) if not prompt_text: - if unchoosable: - return _untraced_outcome(media_file, canonical, ambiguous, untraced) - return resolution["state"] + return "unresolved" if unchoosable else resolution["state"] elif resolution["state"] == "ambiguous": - if unchoosable: - return _untraced_outcome(media_file, canonical, ambiguous, untraced) - return "ambiguous" + return "unresolved" if unchoosable else "ambiguous" elif resolution["state"] == "unresolved" and not isinstance( canonical.get("parameters"), str ): diff --git a/py/api/review_recording.py b/py/api/review_recording.py index 968fb616..41d38fae 100644 --- a/py/api/review_recording.py +++ b/py/api/review_recording.py @@ -33,6 +33,7 @@ the file again. """ import os +from collections import OrderedDict from pathlib import Path from .admin import SCAN_LOG_PREFIXES, _plural, _scan_location, _scan_roots @@ -44,10 +45,16 @@ 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). +MAX_UNTRACED_MEMORY = 65_536 + # The scan log's reason for items that leave the queue with nothing to choose. NOTHING_TO_CHOOSE = "nothing to choose" @@ -55,11 +62,14 @@ 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, inspection=True + media_file, roots, review, inspection=True, scanned=scanned ) except Exception: self.logger.warning( @@ -76,7 +86,9 @@ class ReviewRecordingMixin: if outcome in QUEUED_OUTCOMES: review["queued"] += 1 - def _record_review_sync(self, media_file, roots, review=None, inspection=False): + 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 @@ -95,11 +107,80 @@ class ReviewRecordingMixin: 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. + """ + choice = self._inspection_choice(media_file, roots, canonical) + if choice is None: + self._settle_untraced_reviews([media_file], roots, 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 = (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() + if state is not None and memory.get(locator) == state: + return None + resolution = review_capture.inspection_choice(canonical) + if resolution is not None: + memory.pop(locator, None) + return canonical, resolution + if state is not None: + memory[locator] = state + memory.move_to_end(locator) + while len(memory) > MAX_UNTRACED_MEMORY: + memory.popitem(last=False) + return None + + def _untraced_memory(self): + memory = getattr(self, "_untraced_inspections", None) + if memory is None: + memory = self._untraced_inspections = OrderedDict() + return memory + @staticmethod def _review_location(media_file, roots, real_roots=None): """``(root, root_index, relative, spelled)`` of a listed file, or None. diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index a03164d5..1ee07821 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_review_inspection.py b/tests/test_review_inspection.py index 132a8a29..ef3f3e92 100644 --- a/tests/test_review_inspection.py +++ b/tests/test_review_inspection.py @@ -13,9 +13,11 @@ 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 @@ -35,6 +37,20 @@ 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. @@ -124,6 +140,29 @@ def _prefixed(graph, prefix): 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"] @@ -142,6 +181,9 @@ class InspectionTestCase(ReviewRoutesTestCase): ) return path + def _pending(self): + return self.api.db.library_counts()["review_pending"] + def _evidence(self, item): return json.loads(item["evidence"]) @@ -192,10 +234,11 @@ class TestQwenEditBatchIsQueued(InspectionTestCase): self.assertEqual(item["candidate_count"], 4) self.assertEqual(evidence["profile"], "inspection") self.assertEqual( - evidence["semantics"], inspection_semantics.get_semantics().version + evidence["semantics"], review_capture.semantics_version({"prompt": graph}) ) self.assertEqual( - item["resolver_version"], review_capture.inspection_review_version() + 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) @@ -212,11 +255,10 @@ class TestQwenEditBatchIsQueued(InspectionTestCase): await self._scan() (item,) = self._items() - with patch.object( - self.api._captures, "capture", side_effect=AssertionError("copied") - ): + 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) @@ -309,9 +351,7 @@ class TestApplyInspectionCandidate(InspectionTestCase): (link,) = links provenance = link["provenance"] self.assertEqual(provenance["kind"], "user-selection") - self.assertEqual( - provenance["resolver_version"], review_capture.inspection_review_version() - ) + 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"]) @@ -339,45 +379,26 @@ class TestApplyInspectionCandidate(InspectionTestCase): self.assertEqual(len(calls), 1) async def test_table_change_keeps_the_item_applicable(self): - """A semantics table change moves the version (a rescan records the - new evidence beside the old item, as I2 does for the resolver), and - the pending item stays applicable while its candidate is offered.""" + """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 = review_capture.inspection_review_version() - - def add_encoder(doc): - doc["encoders"].append( - { - "class": "UnrelatedTextEncode", - "slot": 0, - "strategy": "single", - "inputs": ["text"], - } - ) - - self._table(add_encoder) - self.assertNotEqual(review_capture.inspection_review_version(), version) - - await self._scan("all") - items = self._items() - self.assertEqual(len(items), 2) - self.assertEqual(items[0]["id"], old["id"]) - self.assertEqual( - self._evidence(items[1])["semantics"], - inspection_semantics.get_semantics().version, - ) - + version = _version() item, detail, observed = await self._open() - self.assertEqual(item["id"], old["id"]) + + 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): @@ -483,8 +504,284 @@ class TestTableVersion(InspectionTestCase): 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"]) + + +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, + ) diff --git a/utils/inspection_semantics.py b/utils/inspection_semantics.py index c2b57867..f334ade6 100644 --- a/utils/inspection_semantics.py +++ b/utils/inspection_semantics.py @@ -19,8 +19,13 @@ 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. -Needs review folds it into the version of the items it records from the -inspection profile (``utils.review_capture``). +``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 @@ -126,6 +131,8 @@ class Semantics: 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: @@ -529,13 +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 replace(_validate(doc), version=table_version(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.""" - text = json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=True) - return hashlib.sha256(text.encode("utf-8")).hexdigest() + 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/review_capture.py b/utils/review_capture.py index 62ecedab..8a4f8e22 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -27,12 +27,15 @@ batches): when the default profile is ambiguous without a choice, the scan (``inspection=True``; the live monitor never) 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 the semantics table ``version``, and its -``resolver_version`` is ``inspection_review_version()``, which folds the -profile, ``RESOLVER_VERSION`` and the table version together. A table change -therefore moves the version the way a resolver change does (I2): a rescan -records the new evidence beside the pending item, which stays applicable -while the bytes still offer its chosen candidate. Apply re-resolves with the +``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). @@ -55,7 +58,11 @@ try: canonical_evidence, choosable, ) - from .inspection_semantics import SemanticsError, get_semantics + 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 ( @@ -71,7 +78,11 @@ except ImportError: # imported with the repo root on sys.path canonical_evidence, choosable, ) - from utils.inspection_semantics import SemanticsError, get_semantics + 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 ( @@ -95,21 +106,54 @@ INSPECTION_VERSION_BASE = 1 << 40 _INSPECTION_VERSION_HEX = 10 -def semantics_version() -> str | None: - """The inspection semantics table's version, or None when it is unusable.""" +# 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: - return get_semantics().version + 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() -> int | None: - """The ``resolver_version`` of items recorded from the inspection profile. +def inspection_review_version(metadata: Any = None) -> int | None: + """The ``resolver_version`` of an item recorded from the inspection profile. - Folds the profile, ``RESOLVER_VERSION`` and the table version, so either - change moves it. None when the table cannot be loaded. + Folds the profile, ``RESOLVER_VERSION`` and ``semantics_version(metadata)``, + so a resolver change or a table change about the graph's classes moves + it. None when the table cannot be loaded. """ - table = semantics_version() + table = semantics_version(metadata) if table is None: return None seed = f"{INSPECTION_PROFILE.name}:{RESOLVER_VERSION}:{table}" @@ -122,15 +166,18 @@ def resolve_inspection(metadata: dict) -> dict: return resolve_prompt_pair(metadata, profile=INSPECTION_PROFILE) -def inspection_offers_choice(metadata: dict) -> bool: - """True when the inspection profile finds a choice in ``metadata``. +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() is None: - return False - return reviewable(resolve_inspection(metadata)) + 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: @@ -140,22 +187,43 @@ def item_profile(evidence: Any) -> str: return DEFAULT -def _versions(inspection: bool) -> tuple: - """The item versions a file may already be queued under.""" - if not inspection: - return (RESOLVER_VERSION,) - version = inspection_review_version() +def item_versions(metadata: Any = None) -> tuple: + """The item versions a file with ``metadata`` may already be queued under. + + Without ``metadata`` (the monitor), only the ingestion version. + """ + 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 + 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; the live monitor passes False. + that offers a choice; the live monitor passes False. ``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 @@ -170,7 +238,7 @@ def record_ambiguous_file( ``CaptureCapacity`` and database errors propagate to the caller, which logs and counts them. """ - versions = _versions(inspection) + 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, versions) @@ -199,14 +267,24 @@ def record_ambiguous_file( return "skipped" profile, version = DEFAULT, RESOLVER_VERSION if not reviewable(resolution): - version = inspection_review_version() if inspection else None + version = inspection_review_version(captured[0]) if inspection else None if version is None: return "untraced" - # Heavier, so only for a file that is ambiguous with nothing to choose. - resolution, profile = resolve_inspection(captured[0]), INSPECTION + 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) + evidence, count, limited = review_evidence( + resolution, profile, version, captured[0] + ) try: inserted = db.record_review_item( locator=capture.locator, @@ -249,7 +327,9 @@ def _queued_spelling(db, root, relative: str, versions) -> tuple[str, tuple | No 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. ``versions`` are the item versions that count. + 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: @@ -257,7 +337,7 @@ def _queued_spelling(db, root, relative: str, versions) -> tuple[str, tuple | No except OSError: return spelled, None locator = os.path.normcase(spelled) - if any( + 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 ): @@ -326,12 +406,16 @@ def reviewable(resolution: Any) -> bool: def review_evidence( - resolution: dict, profile: str = DEFAULT, version: int | None = None + 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 the semantics table - ``version`` it was resolved with; ``version`` is the item's version. + 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 if version is None else version, @@ -339,7 +423,7 @@ def review_evidence( } if profile == INSPECTION: base["profile"] = INSPECTION - base["semantics"] = semantics_version() + base["semantics"] = semantics_version(metadata) candidates: list[dict] = [] seen: set[str] = set() total = 0