diff --git a/TESTING.md b/TESTING.md index 93b42d3c..43c93009 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2615,7 +2615,26 @@ still have made the file, so that text is not certain either. Both report 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 +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 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. +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 +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 diff --git a/py/api/admin.py b/py/api/admin.py index 0b9b89cb..5b7cb5b5 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 reviewable + from ...utils.review_capture import inspection_offers_choice, 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 reviewable + from utils.review_capture import inspection_offers_choice, reviewable try: @@ -252,6 +252,22 @@ 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. @@ -1714,7 +1730,10 @@ 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``, + 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. """ if not metadata: return None @@ -1756,9 +1775,13 @@ class AdminRoutesMixin: # Graphs without a recognized sampler keep the legacy extractor below. prompt_text = self._graph_prompt_text(metadata) if not prompt_text: - return "unresolved" if unchoosable else resolution["state"] + if unchoosable: + return _untraced_outcome(media_file, canonical, ambiguous, untraced) + return resolution["state"] elif resolution["state"] == "ambiguous": - return "unresolved" if unchoosable else "ambiguous" + if unchoosable: + return _untraced_outcome(media_file, canonical, ambiguous, untraced) + return "ambiguous" elif resolution["state"] == "unresolved" and not isinstance( canonical.get("parameters"), str ): diff --git a/py/api/review.py b/py/api/review.py index 6fb5f4fb..9490ae34 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 26c2de2f..968fb616 100644 --- a/py/api/review_recording.py +++ b/py/api/review_recording.py @@ -8,7 +8,8 @@ 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 the +scan asks that profile (``inspection=True``); the monitor never does. 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 @@ -57,7 +58,9 @@ class ReviewRecordingMixin: def _record_scanned_review(self, media_file, roots, review=None): """Record one ambiguous, unlinked scanned file (logged and counted).""" try: - outcome = self._record_review_sync(media_file, roots, review) + outcome = self._record_review_sync( + media_file, roots, review, inspection=True + ) except Exception: self.logger.warning( f"{media_file.name} could not be added to Needs review", @@ -73,9 +76,11 @@ 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): """``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. """ @@ -84,7 +89,12 @@ 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, ) if outcome == "untraced": self._settle_untraced_reviews([media_file], roots, review) diff --git a/tests/test_review_inspection.py b/tests/test_review_inspection.py new file mode 100644 index 00000000..132a8a29 --- /dev/null +++ b/tests/test_review_inspection.py @@ -0,0 +1,490 @@ +"""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 json +import os +import tempfile +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") + + +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 _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 _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"], inspection_semantics.get_semantics().version + ) + self.assertEqual( + item["resolver_version"], review_capture.inspection_review_version() + ) + 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 patch.object( + self.api._captures, "capture", side_effect=AssertionError("copied") + ): + job = await self._scan("all") + + 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"], review_capture.inspection_review_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 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.""" + 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, + ) + + item, detail, observed = await self._open() + self.assertEqual(item["id"], old["id"]) + 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(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" + ] + + self._table(drop_encoder) + 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) + + +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)), "untraced") + 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)), "untraced") + 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()) + job = await self._scan() + + self.assertEqual(self._items(), []) + (row,) = job["outcomes"]["review"] + self.assertIn("could not be traced", row["detail"]) diff --git a/utils/inspection_semantics.py b/utils/inspection_semantics.py index be0dce9a..c2b57867 100644 --- a/utils/inspection_semantics.py +++ b/utils/inspection_semantics.py @@ -16,10 +16,16 @@ 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. +Needs review folds it into the version of the items it records from the +inspection profile (``utils.review_capture``). """ +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 +124,8 @@ 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 = "" def _freeze(value: Any) -> Frozen: @@ -521,7 +529,13 @@ 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)) + + +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() @lru_cache(maxsize=1) diff --git a/utils/review_capture.py b/utils/review_capture.py index f26a46b2..62ecedab 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -21,6 +21,21 @@ 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``; 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 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 @@ -40,9 +55,14 @@ try: canonical_evidence, choosable, ) + from .inspection_semantics import SemanticsError, get_semantics 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, + 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 +71,14 @@ except ImportError: # imported with the repo root on sys.path canonical_evidence, choosable, ) + from utils.inspection_semantics import SemanticsError, get_semantics 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, + RESOLVER_VERSION, + resolve_prompt_pair, + ) logger = logging.getLogger("prompt_manager.review_capture") @@ -61,11 +86,77 @@ 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: +def semantics_version() -> str | None: + """The inspection semantics table's version, or None when it is unusable.""" + try: + return get_semantics().version + except SemanticsError: + return None + + +def inspection_review_version() -> int | None: + """The ``resolver_version`` of items 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. + """ + table = semantics_version() + if table is None: + return None + seed = f"{INSPECTION_PROFILE.name}:{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_offers_choice(metadata: dict) -> bool: + """True when the inspection profile finds a choice in ``metadata``. + + 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)) + + +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 _versions(inspection: bool) -> tuple: + """The item versions a file may already be queued under.""" + if not inspection: + return (RESOLVER_VERSION,) + version = inspection_review_version() + return (RESOLVER_VERSION,) if version is None else (RESOLVER_VERSION, version) + + +def record_ambiguous_file( + db, store, root, root_index: int, relative: str, *, inspection: bool = False +) -> 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. + 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 +170,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 = _versions(inspection) + 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 +197,16 @@ 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() 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 + if resolution.get("state") != "ambiguous" or not reviewable(resolution): + return "untraced" + evidence, count, limited = review_evidence(resolution, profile, version) try: inserted = db.record_review_item( locator=capture.locator, @@ -116,7 +215,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 +223,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 +249,7 @@ 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. """ spelled = os.path.join(os.path.realpath(os.fspath(root)), *relative.split("/")) try: @@ -155,8 +257,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 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 +268,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 +302,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 +325,21 @@ 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 +) -> 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. + """ 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() candidates: list[dict] = [] seen: set[str] = set() total = 0