feat(review): supersede stale items and inspect each file state once

A semantics table or resolver version change no longer leaves two pending
items for one file. A new item supersedes the other items of the same bytes
when every one is 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 skips those bytes under any version.

The inspection item version now folds a narrower fingerprint: the table
restricted to the graph's node classes plus its class-independent parts, so
an edit about other classes leaves every item alone.

The inspection profile runs once per file state. The scan resolves the
untraced file at batch level, skips files already queued or decided at their
current size and mtime under their version, and remembers files it found
untraced in this process. The capture reuses the scan's result when it holds
the same records.

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

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:

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
         ):

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.

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]

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,
+        )

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)

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