fix(review): queue only ambiguous files that nothing links, and heal the queue

The maintainer's live scan (19,908 files, scope "new") reported 4,677 files
under Needs review, but the queue held 10,232 items: every file the strict
resolver found ambiguous was queued, including the ones the legacy
browser-parser fallback then linked (all sampled "added" rows) and the ones
already linked to a known prompt (135 of 200 sampled "known" rows).

A file is now queued only when its final outcome is ambiguous and unlinked:
the resolver finds it ambiguous, the fallback links nothing and no saved row
links it already. An ambiguous file that a saved row links is reported as
unchanged ("Already linked to its prompt"), not as Needs review.

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.
The monitor does the same for each ambiguous output it links, and never
queues one. Only an undecided item is removed (pending, revision 0, no
decision row), so a decision is never deleted or changed and the schema
needs no new state.

The scan job's scope.review_items now counts this scan's files that are in
the pending queue after it, new or already there, so it agrees with the
review group's ambiguous rows and with the queue total. review_added and
review_settled are new, and the log gains "already in Needs review" and
"left Needs review · now linked" lines. A duplicate whose item is decided is
reported as "decided" and is not counted.

TESTING.md

@@ -2213,10 +2213,24 @@ validation, annotation group included. The browser fixtures clear
 never the metadata row.
 
 Needs review items (#249, `utils/review_capture.py`, `database/review.py`,
-`tests/test_review.py`): the output scan (Activity's Scan for metadata and the
-legacy `/scan`) and the live monitor hand every file the ingestion resolver
-finds `ambiguous` to `record_ambiguous_file`, whether or not the browser parser
-then links it (`conflict` and `unresolved` files are not queued). It captures
+`tests/test_review.py`, `tests/test_review_outcome.py`): a file is queued only
+when its final outcome is ambiguous and unlinked. The output scan (Activity's
+Scan for metadata and the legacy `/scan`) hands `record_ambiguous_file` a file
+the ingestion resolver finds `ambiguous` only when the browser-parser fallback
+then links nothing (status `ambiguous`) and no saved row links the file
+already; an ambiguous file a saved row links is reported `unchanged` ("Already
+linked to its prompt"), not Needs review. A file the fallback links, an exact
+pair the ingester stores, and `conflict` and `unresolved` files are not
+queued. The queue heals itself: after each batch, the scan removes the pending
+items of every listed file (read or skipped as known) that a saved row now
+links, and the monitor does the same for each ambiguous output it links
+(`settle_linked_review_items`; the review index revision moves). Only an
+undecided item goes (`pending`, revision 0, no decision row): a decision is
+never deleted or changed, so no schema state was added. The monitor never
+queues an output it linked; an unlinked file passed to `record_review_file`
+follows the scan's rule (a file the fallback would link is `skipped`). The
+maintainer's live check of #284/#288 found the queue at 10,232 against a review
+group of 4,677: every fallback-linked ambiguous file had been queued. It captures
 the file (scope `needs-review`), reads its metadata again from the capture,
 resolves it with the default profile and stores only an ambiguous result,
 bound to the capture's byte record (created or refreshed like an observe) and
@@ -2230,14 +2244,20 @@ resolver_version, evidence_hash)` and never touches an existing item or its
 decisions; changed bytes add a new pending item and keep the old one. The
 capture is released before the write; no model is loaded and no pixel is
 decoded. Scan statuses, counters, outcome rows and the legacy SSE events are
-unchanged; the job's `scope` gains `review_items` (new items) and
-`review_errors`, and the log ends with `? N images added to Needs review` (and
-`! N images could not be added to Needs review` on failures, e.g. capture
-capacity). Recording captures in the background: it never takes the store's
+unchanged apart from the already-linked case above; the job's `scope` gains
+`review_items` (this scan's files that are in the pending queue after it, new
+or already there: the review group's ambiguous rows, less files a capture could
+not read or a decision already settled), `review_added` (new items),
+`review_settled` (items removed because their file is linked now) and
+`review_errors`. The log ends with `? N images added to Needs review`,
+`? N images already in Needs review`, `= N images left Needs review · now
+linked` and `! N images could not be added to Needs review` (failures, e.g.
+capture capacity), each only when N > 0. Recording captures in the background: it never takes the store's
 last worker or capture slot, which stay free for an interactive Tag dialog or
 Apply, and waits up to 2 s for room before counting the file as not added
-(`test_background_recording_leaves_a_worker_for_the_tag_dialog`). The monitor records after its fallback link, best effort: a failure
-is logged and the link stays. Recording follows the registry's `records`
+(`test_background_recording_leaves_a_worker_for_the_tag_dialog`). The monitor's
+pass runs after its own link, best effort: a failure is logged and the link
+stays. Recording follows the registry's `records`
 admission (#289): the scan's recording runs under its own `records` job, and
 the monitor's (one file, not a job) is refused while a `records` or `guarded`
 job or hold runs. It then records nothing, logs that the file waits for the

__init__.py

@@ -122,9 +122,10 @@ try:
     _global_prompt_tracker = get_prompt_tracker(_global_db)
     _global_image_monitor = get_image_monitor(_global_db, _global_prompt_tracker)
 
-    # Ambiguous new outputs go to Needs review through the API's capture store
-    # (the one the scan and the Tag dialog use). Recording stops before that
-    # store is closed at shutdown.
+    # The Needs review pass for ambiguous new outputs runs through the API's
+    # capture store (the one the scan and the Tag dialog use): a linked output
+    # is never queued and its pending items leave the queue. It stops before
+    # that store is closed at shutdown.
     _review_api = globals().get("api")
     if _review_api is not None:
         _global_image_monitor.set_review_recorder(_review_api.record_review_file)

database/review.py

@@ -32,6 +32,7 @@ from .annotation_schema import (
 from .annotations import _new_id, _validate_locator, _validate_target
 from .models import normalize_image_path, utc_now_iso
 from .review_schema import (
+    DECISIONS_TABLE,
     ITEMS_TABLE,
     MAX_CANDIDATES,
     MAX_SPELLED_PATH_BYTES,
@@ -80,6 +81,26 @@ _QUEUED = (
     "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.resolver_version=? "
     "LIMIT 1"
 )
+_PENDING_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.resolver_version=? "
+    "AND i.state='pending' LIMIT 1"
+)
+# Pending items of one file: the candidates a link settles.
+_PENDING_OF_LOCATOR = (
+    f"SELECT i.id 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 i.state='pending'"
+)
+# Only an undecided item goes: a decision row (RESTRICT) is never deleted.
+_SETTLE_ITEM = (
+    f"DELETE FROM {ITEMS_TABLE} WHERE id=? AND state='pending' "  # nosec B608
+    f"AND revision=0 AND NOT EXISTS (SELECT 1 FROM {DECISIONS_TABLE} d "
+    f"WHERE d.item_id={ITEMS_TABLE}.id)"
+)
 _EPOCH = "SELECT storage_epoch FROM execution_history_metadata WHERE singleton_id=1"
 _PENDING_COUNT = (
     f"SELECT COUNT(*) FROM {ITEMS_TABLE} WHERE state='pending'"  # nosec B608
@@ -342,6 +363,60 @@ class ReviewOperations:
         )
         return row is not None
 
+    def review_pending_at(
+        self, locator: str, size: int, mtime_ns: int, resolver_version: int
+    ) -> bool:
+        """True when a *pending* item cites bytes verified at this exact stat.
+
+        ``review_fingerprint_queued`` also matches decided items; a file whose
+        only item is decided is not in the queue.
+        """
+        row = (
+            self.model.get_connection()
+            .execute(_PENDING_AT, (locator, size, mtime_ns, resolver_version))
+            .fetchone()
+        )
+        return row is not None
+
+    def review_file_linked(self, spellings) -> bool:
+        """True when a saved row links the file under any of ``spellings``."""
+        with self._annotation_read() as conn:
+            return bool(dependent_images(conn, spellings))
+
+    def settle_linked_review_items(self, files) -> int:
+        """Remove the pending, undecided items of files that are now linked.
+
+        ``files`` holds ``(locator, spellings)`` per file. An item leaves the
+        queue only when it is still ``pending`` at revision 0 with no decision
+        row and a saved row links its file under one of ``spellings``; a
+        decided item is never touched. Files without a pending item cost one
+        indexed read and open no write transaction. Returns how many items
+        were removed (the review index revision moves when any was).
+        """
+        files = [(locator, tuple(spellings)) for locator, spellings in files]
+        for locator, _spellings in files:
+            _validate_locator(locator)
+        conn = self.model.get_connection()
+        candidates = [
+            (locator, spellings)
+            for locator, spellings in files
+            if conn.execute(_PENDING_OF_LOCATOR, (locator,)).fetchone() is not None
+        ]
+        if not candidates:
+            return 0
+        removed = 0
+        with self._pair_transaction() as conn:
+            for locator, spellings in candidates:
+                if not dependent_images(conn, spellings):
+                    continue
+                for (item_id,) in conn.execute(
+                    _PENDING_OF_LOCATOR, (locator,)
+                ).fetchall():
+                    removed += conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
+            if removed:
+                conn.execute(_BUMP_REVIEW_INDEX)
+        return removed
+
     def review_queue_page(self, *, limit: int, after: tuple | None = None) -> dict:
         """One page of pending items, oldest first, from one snapshot.
 

py/api/admin.py

@@ -211,6 +211,28 @@ def _time_left(done, total, batches, elapsed):
     return f" · about {minutes} min left"
 
 
+def new_review_counts():
+    """The scan's Needs review tally: ``added`` new items, ``queued`` files
+    in the queue after this scan (new or already there), ``settled`` items
+    removed because their file is linked now, and ``errors``."""
+    return {"added": 0, "queued": 0, "settled": 0, "errors": 0}
+
+
+def _review_scope(review):
+    """The scan job's ``scope`` keys for its Needs review tally.
+
+    ``review_items`` is the number of this scan's files that are in the
+    pending queue after it: the review group's ambiguous files, less those a
+    capture could not read or a decision already settled.
+    """
+    return {
+        "review_items": review["queued"],
+        "review_added": review["added"],
+        "review_settled": review["settled"],
+        "review_errors": review["errors"],
+    }
+
+
 def _scan_status_line(done, total, batches, elapsed):
     """ "Processing file N of M", with an estimate once enough batches ran."""
     line = f"Processing file {done:,} of {total:,}"
@@ -1445,8 +1467,8 @@ class AdminRoutesMixin:
             known = set()
             if media_files and job.scope.get("files") != "all":
                 known = await self._known_scan_files(job, media_files)
-            review = {"recorded": 0, "errors": 0}
-            job.scope.update(review_items=0, review_errors=0)
+            review = new_review_counts()
+            job.scope.update(_review_scope(review))
             async for event in self._scan_batches(
                 media_files, counts, job, roots, known, review
             ):
@@ -1524,11 +1546,12 @@ class AdminRoutesMixin:
                 )
                 for key in counts:
                     counts[key] += batch_counts[key]
-                if review is not None:
-                    job.scope.update(
-                        review_items=review["recorded"],
-                        review_errors=review["errors"],
-                    )
+            if review is not None:
+                # After the batch's links: a file linked now leaves the queue.
+                await self._run_in_executor(
+                    self._settle_scanned_reviews, batch, roots, review
+                )
+                job.scope.update(_review_scope(review))
             for group, row, log_line in outcomes:
                 job.record(group, row, log_line)
 
@@ -1585,10 +1608,11 @@ class AdminRoutesMixin:
 
         Returns the legacy counts. When ``outcomes`` is a list, every processed
         file appends its ``(group, row, log_line)`` for the scan job, with the
-        path relative to its root in ``roots`` (see ``_scan_roots``). A file the
-        resolver finds ambiguous is also recorded for Needs review (best
-        effort, counted in ``review`` when given); statuses and counts do not
-        depend on it.
+        path relative to its root in ``roots`` (see ``_scan_roots``). A file
+        whose final outcome is ambiguous and unlinked (the resolver found it
+        ambiguous and the fallback did not link it) is recorded for Needs
+        review (best effort, counted in ``review`` when given); one a saved
+        row already links is reported ``unchanged`` and never queued.
         """
         counts = dict.fromkeys(SCAN_COUNT_KEYS, 0)
 
@@ -1610,8 +1634,11 @@ class AdminRoutesMixin:
                 counts["failed"] += 1
                 note(media_file, "error")
                 continue
-            if ambiguous:
-                self._record_scanned_review(media_file, roots, review)
+            if ambiguous and outcome == "ambiguous":
+                if self._review_file_linked(media_file, roots):
+                    outcome = "unchanged"  # already linked: not a review case
+                else:
+                    self._record_scanned_review(media_file, roots, review)
             if outcome is None:
                 note(media_file, None)
                 continue
@@ -1655,7 +1682,8 @@ class AdminRoutesMixin:
         ``status``, the prompt's display ``title`` and, for a resolved pair,
         the pair counts. When the resolver finds the file ambiguous (whether
         or not the browser parser then links it) and ``ambiguous`` is a list,
-        ``media_file`` is appended to it.
+        ``media_file`` is appended to it; only a file that then returns
+        ``"ambiguous"`` (nothing linked it) is a Needs review case.
         """
         if not metadata:
             return None

py/api/review_recording.py

@@ -1,16 +1,29 @@
 """Needs review recording for the output scan and the live monitor (#249).
 
-The scan hands every file the ingestion resolver finds ambiguous to
-``_record_scanned_review`` (best effort: a failure is logged and counted);
-the monitor calls ``record_review_file`` from its worker thread after its
-fallback link. Both go through ``utils.review_capture.record_ambiguous_file``
-on the API's capture store. Nothing here links a file.
+The rule (maintainer test of #284/#288): a file is queued only when its
+final outcome is ambiguous and unlinked. The strict resolver must find it
+``ambiguous``, the scan's legacy fallback (the browser parser,
+``_graph_prompt_text``) must not link it, and no saved row may link it
+already. A file the fallback links, an exact pair the ingester stored and an
+already linked file are never queued.
+
+The queue heals itself: a pending item whose file is now linked (by the
+fallback, a later resolution or the monitor) is removed by the next scan
+that lists the file or by the monitor's next pass over it. Only an
+undecided item goes (``pending``, revision 0, no decision row); a decision
+is never deleted or changed (``settle_linked_review_items``).
+
+The scan records through ``_record_scanned_review`` (best effort: a failure
+is logged and counted); the monitor calls ``record_review_file`` from its
+worker thread after its own link. Both go through
+``utils.review_capture.record_ambiguous_file`` on the API's capture store.
+Nothing here links a file.
 
 Recording changes records, so it follows the registry's ``records``
 admission (#289). The scan records under its own ``records`` job. The
-monitor's recording lasts one file and is not a job: it is refused while a
+monitor's pass lasts one file and is not a job: it is refused while a
 ``records`` or ``guarded`` job runs (``deferred``), and the next scan reads
-the ambiguous file again.
+the file again.
 """
 
 import os
@@ -20,6 +33,9 @@ from .admin import SCAN_LOG_PREFIXES, _plural, _scan_location, _scan_roots
 
 # The kind the monitor's recording checks admission under (never registered).
 REVIEW_RECORDING_KIND = "review_recording"
+# record_ambiguous_file outcomes that leave the file in the pending queue.
+QUEUED_OUTCOMES = ("recorded", "limited", "duplicate")
+NEW_OUTCOMES = ("recorded", "limited")
 
 try:
     from ...utils.review_capture import record_ambiguous_file
@@ -31,7 +47,7 @@ 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 scanned file; a failure is logged and counted."""
+        """Record one ambiguous, unlinked scanned file (logged and counted)."""
         try:
             outcome = self._record_review_sync(media_file, roots)
         except Exception:
@@ -42,48 +58,130 @@ class ReviewRecordingMixin:
             if review is not None:
                 review["errors"] += 1
             return
-        if review is not None and outcome in ("recorded", "limited"):
-            review["recorded"] += 1
+        if review is None:
+            return
+        if outcome in NEW_OUTCOMES:
+            review["added"] += 1
+        if outcome in QUEUED_OUTCOMES:
+            review["queued"] += 1
 
     def _record_review_sync(self, media_file, roots):
         """``record_ambiguous_file`` for a file under one of ``roots`` (blocking)."""
-        relative, _label, index = _scan_location(media_file, roots)
-        root = next((r for r, _l, i in roots if i == index), None)
-        if index is None or root is None:
+        located = self._review_location(media_file, roots)
+        if located is None:
             return "skipped"
+        root, index, relative, _spelled = located
         return record_ambiguous_file(
             self.db, self._capture_store(), Path(root), index, relative
         )
 
-    def record_review_file(self, image_path):
-        """Needs review recording for the live monitor (its worker thread).
+    @staticmethod
+    def _review_location(media_file, roots, real_roots=None):
+        """``(root, root_index, relative, spelled)`` of a listed file, or None.
 
-        Maps ``image_path`` to the gallery root that lists it (the listing's
-        ``root_index``); a file under no current root is ``skipped``. While a
-        job runs that a ``records`` job may not run beside, nothing is recorded
-        and the answer is ``deferred``. Errors propagate: the monitor logs them
-        and keeps its link.
+        ``spelled`` is the capture's spelling of the file (the real root joined
+        with the relative path); ``os.path.normcase`` of it is the locator
+        review items are stored under. ``real_roots`` caches each root's
+        ``realpath`` across a batch.
+        """
+        relative, _label, index = _scan_location(media_file, roots)
+        root = next((r for r, _l, i in roots if i == index), None)
+        if index is None or root is None:
+            return None
+        if real_roots is None:
+            real_roots = {}
+        real = real_roots.get(index)
+        if real is None:
+            real = real_roots[index] = os.path.realpath(os.fspath(root))
+        spelled = os.path.join(real, *relative.split("/"))
+        return root, index, relative, spelled
+
+    def _review_file_linked(self, media_file, roots):
+        """True when a saved row links the scanned file (any spelling)."""
+        located = self._review_location(media_file, roots)
+        spellings = (str(media_file),) + ((located[3],) if located else ())
+        return self.db.review_file_linked(spellings)
+
+    def _settle_scanned_reviews(self, media_files, roots, review=None):
+        """Remove the pending items of ``media_files`` that are now linked.
+
+        Blocking; best effort like recording (a failure is logged and
+        counted in ``review["errors"]``). Returns how many items left.
+        """
+        real_roots = {}
+        files = []
+        for media_file in media_files:
+            located = self._review_location(media_file, roots, real_roots)
+            if located is not None:
+                spelled = located[3]
+                files.append((os.path.normcase(spelled), (str(media_file), spelled)))
+        try:
+            settled = self.db.settle_linked_review_items(files) if files else 0
+        except Exception:
+            self.logger.warning(
+                "Needs review items of linked files could not be removed",
+                exc_info=True,
+            )
+            if review is not None:
+                review["errors"] += 1
+            return 0
+        if review is not None:
+            review["settled"] += settled
+        return settled
+
+    def _fallback_links(self, metadata):
+        """True when the scan's legacy fallback would link these records."""
+        return bool(metadata.get("prompt")) and bool(self._graph_prompt_text(metadata))
+
+    def record_review_file(self, image_path):
+        """Apply the Needs review rule to one monitor output (its worker thread).
+
+        Maps ``image_path`` to the gallery root that lists it; a file under no
+        current root is ``skipped``. A linked file is never queued: its
+        pending items are removed and the answer is ``linked``. An unlinked
+        file the scan's fallback would link is ``skipped`` (the next scan
+        links it); any other file is recorded when it is ambiguous. While a
+        job runs that a ``records`` job may not run beside, nothing is written
+        and the answer is ``deferred``. Errors propagate: the monitor logs
+        them and keeps its link.
         """
         busy = self._jobs.blocker(REVIEW_RECORDING_KIND, admission="records")
         if busy is not None:
             self.logger.info(
-                f"{os.path.basename(str(image_path))} was not added to Needs "
-                f"review while {busy.title} runs; the next scan reads it again"
+                f"Needs review was not updated for "
+                f"{os.path.basename(str(image_path))} while {busy.title} runs; "
+                "the next scan reads it again"
             )
             return "deferred"
         full = os.path.abspath(str(image_path))
         located = Path(os.path.realpath(os.path.dirname(full))) / os.path.basename(full)
-        return self._record_review_sync(
-            located, _scan_roots(self._get_all_output_dirs())
-        )
+        roots = _scan_roots(self._get_all_output_dirs())
+        if self._review_location(located, roots) is None:
+            return "skipped"
+        if self._review_file_linked(located, roots):
+            self._settle_scanned_reviews([located], roots)
+            return "linked"
+        if self._fallback_links(self._extract_comfyui_metadata(str(located))):
+            return "skipped"
+        return self._record_review_sync(located, roots)
 
     @staticmethod
     def _log_review_items(job, review):
         """The scan log's Needs review summary lines (none when nothing moved)."""
-        if review["recorded"]:
+        prefix = SCAN_LOG_PREFIXES["review"]
+        if review["added"]:
             job.log_line(
-                f"{SCAN_LOG_PREFIXES['review']} "
-                f"{_plural(review['recorded'], 'image')} added to Needs review"
+                f"{prefix} {_plural(review['added'], 'image')} added to Needs review"
+            )
+        already = review["queued"] - review["added"]
+        if already:
+            job.log_line(
+                f"{prefix} {_plural(already, 'image')} already in Needs review"
+            )
+        if review["settled"]:
+            job.log_line(
+                f"{SCAN_LOG_PREFIXES['known']} "
+                f"{_plural(review['settled'], 'image')} left Needs review · now linked"
             )
         if review["errors"]:
             job.log_line(

tests/test_api_review.py

@@ -69,10 +69,23 @@ class ReviewRoutesTestCase(ReviewTestCase):
 
     def _record(self, path):
         self._roots()
-        outcome = self.api.record_review_file(str(path))
+        outcome = self._record_raw(path)
         self.assertIn(outcome, ("recorded", "limited"), outcome)
         return outcome
 
+    def _record_raw(self, path):
+        """Record ``path`` as the producers' recorder does, without their rule.
+
+        These tests are about decisions on an item; whether the scan or the
+        monitor would queue the file (``tests/test_review_outcome.py``) is
+        not their subject, and an item can outlive the moment its file gets
+        linked (until the next scan or monitor pass settles it).
+        """
+        from py.api.admin import _scan_roots
+
+        roots = _scan_roots(self.api._get_all_output_dirs())
+        return self.api._record_review_sync(Path(str(path)), roots)
+
     async def _json(self, method, url, **kwargs):
         resp = await self.client.request(method, url, **kwargs)
         data = await resp.json()
@@ -278,7 +291,7 @@ class TestQueueReads(ReviewRoutesTestCase):
                 "Steps: 20, Sampler: Euler"
             ),
         )
-        self.assertEqual(self.api.record_review_file(str(limited)), "limited")
+        self.assertEqual(self._record_raw(str(limited)), "limited")
         self._record(self._ambiguous("two.png"))
         items = (await self._list())["items"]
         self.assertEqual([i["limited"] for i in items], [True, False])
@@ -1121,7 +1134,7 @@ class TestInteractiveCapacity(ReviewRoutesTestCase):
             return original(store_, stream, size)
 
         def record():
-            outcome["value"] = self.api.record_review_file(str(busy))
+            outcome["value"] = self._record_raw(str(busy))
 
         recorder = threading.Thread(target=record)
         with patch.object(CaptureStore, "_read", blocking):
@@ -1132,7 +1145,7 @@ class TestInteractiveCapacity(ReviewRoutesTestCase):
                 # Another background recording finds no worker it may use...
                 with patch("utils.media_capture.BACKGROUND_WAIT_SECONDS", 0.01):
                     with self.assertRaises(CaptureCapacity):
-                        self.api.record_review_file(str(busy))
+                        self._record_raw(str(busy))
                 # ...while the interactive observe still gets the last one.
                 observed = await self._observe(item["target"])
                 self.assertEqual(

tests/test_review.py

@@ -72,6 +72,29 @@ def ambiguous_graph(*positives):
     return graph
 
 
+def unlinkable_graph(*positives):
+    """Ambiguous, and the scan's browser-parser fallback links nothing.
+
+    Only such a file is queued (an ambiguous file the fallback links is not).
+    The parser reads the first sampler (node 3), whose positive input points
+    at a node missing from the graph; the resolver offers the other samplers
+    as candidates beside that unresolved one.
+    """
+    graph = ambiguous_graph(*positives)
+    graph["3"] = {
+        "class_type": "KSampler",
+        "inputs": {"model": ["1", 0], "positive": ["99", 0], "negative": ["2", 0]},
+    }
+    return graph
+
+
+def resolved_texts(evidence):
+    """The positive texts of the candidates that resolved, sorted."""
+    return sorted(
+        c["positive"]["text"] for c in evidence["candidates"] if c["positive"]
+    )
+
+
 def resolved_graph(text="a red square"):
     return ambiguous_graph(text)
 
@@ -155,7 +178,9 @@ class TestScanRecording(ReviewTestCase):
         self._roots()
         ambiguous = self.output_dir / "sub" / "two.png"
         ambiguous.parent.mkdir()
-        write_png(ambiguous, prompt=ambiguous_graph("cat", "dog"))
+        write_png(ambiguous, prompt=unlinkable_graph("cat", "dog"))
+        # Ambiguous too, but the browser-parser fallback links it: not queued.
+        write_png(self.output_dir / "linked.png", prompt=ambiguous_graph("a", "b"))
         write_png(self.output_dir / "resolved.png", prompt=resolved_graph())
         write_png(self.output_dir / "plain.png")
         write_png(
@@ -178,20 +203,19 @@ class TestScanRecording(ReviewTestCase):
             (item["state"], item["revision"], item["limited"]), ("pending", 0, 0)
         )
         self.assertEqual(item["resolver_version"], 1)
-        self.assertEqual(item["candidate_count"], 2)
+        self.assertEqual(item["candidate_count"], 3)
         self.assertEqual(item["evidence_hash"], evidence_digest(item["evidence"]))
         evidence = json.loads(item["evidence"])
         self.assertEqual(canonical_evidence(evidence), item["evidence"])
         self.assertEqual(evidence["resolver_version"], 1)
-        self.assertEqual(
-            sorted(c["positive"]["text"] for c in evidence["candidates"]),
-            ["cat", "dog"],
-        )
+        self.assertEqual(resolved_texts(evidence), ["cat", "dog"])
         for candidate in evidence["candidates"]:
             self.assertEqual(
                 set(candidate),
                 {"hash", "sampler_id", "state", "reason", "positive", "negative"},
             )
+            if candidate["positive"] is None:  # the parser's unresolved sampler
+                continue
             self.assertEqual(
                 set(candidate["positive"]), {"text", "storage_text", "provenance"}
             )
@@ -210,7 +234,7 @@ class TestScanRecording(ReviewTestCase):
     async def test_scan_statuses_counters_and_outcome_rows_unchanged(self):
         self._require_module()
         self._roots()
-        write_png(self.output_dir / "two.png", prompt=ambiguous_graph("cat", "dog"))
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
         write_png(self.output_dir / "one.png", prompt=resolved_graph())
         write_png(
             self.output_dir / "delims.png",
@@ -243,8 +267,8 @@ class TestScanRecording(ReviewTestCase):
         with self.assertLogs("prompt_manager.api", logging.WARNING) as logs:
             without = run(other, broken)
         self.assertEqual(recorded, without)
-        self.assertEqual(recorded[0]["ambiguous"], 1)
-        self.assertEqual(recorded[0]["added"], 2)
+        self.assertEqual(recorded[0]["ambiguous"], 2)
+        self.assertEqual(recorded[0]["added"], 1)
         self.assertTrue(any("Needs review" in line for line in logs.output))
         self.assertEqual(len(self._items()), 2)
         other.close_annotation_captures()
@@ -258,7 +282,7 @@ class TestScanRecording(ReviewTestCase):
     async def test_new_scope_rereads_ambiguous_files_and_stays_idempotent(self):
         self._require_module()
         self._roots()
-        write_png(self.output_dir / "two.png", prompt=ambiguous_graph("cat", "dog"))
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
         write_png(self.output_dir / "one.png", prompt=resolved_graph())
         first = await self._scan("new")
         revision = self._review_revision()
@@ -274,8 +298,11 @@ class TestScanRecording(ReviewTestCase):
 
         self.assertEqual(reads, ["two.png"])
         self.assertEqual(first["scope"]["review_items"], 1)
-        self.assertEqual(second["scope"]["review_items"], 0)
+        # Still in the queue (review_items), but nothing new was added.
+        self.assertEqual(second["scope"]["review_items"], 1)
+        self.assertEqual(second["scope"]["review_added"], 0)
         self.assertNotIn("added to Needs review", "\n".join(second["log"]))
+        self.assertIn("? 1 image already in Needs review", "\n".join(second["log"]))
         self.assertEqual(len(self._items()), 1)
         self.assertEqual(self._review_revision(), revision)
         self.assertEqual(self._capture_files(), [])
@@ -287,7 +314,7 @@ class TestScanRecording(ReviewTestCase):
         self._require_module()
         self._roots()
         path = self.output_dir / "two.png"
-        write_png(path, prompt=ambiguous_graph("cat", "dog"))
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
         await self._scan("new")
         captures = []
         original = CaptureStore.capture
@@ -308,7 +335,8 @@ class TestScanRecording(ReviewTestCase):
             second = await self._scan("new")
             self.assertEqual(captures, [])
             self.assertEqual(reads, ["two.png"])
-            self.assertEqual(second["scope"]["review_items"], 0)
+            self.assertEqual(second["scope"]["review_added"], 0)
+            self.assertEqual(second["scope"]["review_items"], 1)
             # A touched file (same bytes, new mtime) is captured again: a
             # duplicate of its item, whose fingerprint is refreshed.
             stat = path.stat()
@@ -323,7 +351,7 @@ class TestScanRecording(ReviewTestCase):
     async def test_repeat_scan_keeps_decisions(self):
         self._require_module()
         self._roots()
-        write_png(self.output_dir / "two.png", prompt=ambiguous_graph("cat", "dog"))
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
         await self._scan("all")
         item = self._items()[0]
         candidate = json.loads(item["evidence"])["candidates"][0]["hash"]
@@ -362,10 +390,10 @@ class TestScanRecording(ReviewTestCase):
         self._require_module()
         self._roots()
         path = self.output_dir / "two.png"
-        write_png(path, prompt=ambiguous_graph("cat", "dog"))
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
         await self._scan()
         old = self._items()[0]
-        write_png(path, prompt=ambiguous_graph("cat", "dog"), color=(0, 0, 255))
+        write_png(path, prompt=unlinkable_graph("cat", "dog"), color=(0, 0, 255))
         self.assertNotEqual(digest(path), old["sha256"])
 
         job = await self._scan()
@@ -387,7 +415,7 @@ class TestScanRecording(ReviewTestCase):
             parameters=DELIMITERS,
         )
         many = [f"prompt {n}" for n in range(17)]
-        write_png(self.output_dir / "many.png", prompt=ambiguous_graph(*many))
+        write_png(self.output_dir / "many.png", prompt=unlinkable_graph(*many))
 
         job = await self._scan()
 
@@ -399,14 +427,16 @@ class TestScanRecording(ReviewTestCase):
             self.assertEqual(evidence["candidates"], [])
             self.assertIs(evidence["limited"], True)
         self.assertEqual(
-            json.loads(items["many.png"]["evidence"])["candidate_total"], 17
+            # 17 samplers and the parser's unresolved one.
+            json.loads(items["many.png"]["evidence"])["candidate_total"],
+            18,
         )
         self.assertEqual(job["scope"]["review_items"], 2)
 
     async def test_capacity_errors_are_logged_and_counted_not_fatal(self):
         self._require_module()
         self._roots()
-        write_png(self.output_dir / "two.png", prompt=ambiguous_graph("cat", "dog"))
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
 
         def full(*args, **kwargs):
             raise CaptureCapacity("Too many captures are in progress")
@@ -415,8 +445,8 @@ class TestScanRecording(ReviewTestCase):
             job = await self._scan()
 
         self.assertEqual(job["state"], "finished")
-        # The browser parser links this ambiguous file: Activity counts it Added.
-        self.assertEqual(job["counters"]["added"], 1)
+        # Nothing links this ambiguous file: Activity counts it under review.
+        self.assertEqual(job["counters"]["review"], 1)
         self.assertEqual(job["scope"]["review_items"], 0)
         self.assertEqual(job["scope"]["review_errors"], 1)
         self.assertIn(
@@ -427,7 +457,7 @@ class TestScanRecording(ReviewTestCase):
     async def test_legacy_scan_events_are_unchanged(self):
         self._require_module()
         self._roots()
-        write_png(self.output_dir / "two.png", prompt=ambiguous_graph("cat", "dog"))
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
         resp = await self.client.request("POST", "/prompt_manager/scan")
         payloads = _sse_payloads(await resp.text())
         final = payloads[-1]
@@ -755,15 +785,17 @@ class TestMonitorRecording(ReviewTestCase):
         ).fetchone()
         return row[0] if row else None
 
-    async def test_monitor_keeps_fallback_link_and_records_item_best_effort(self):
+    async def test_monitor_keeps_its_link_and_never_queues_the_linked_file(self):
+        """The monitor's pass is best effort (a failure is logged, the link
+        stays); a file the monitor linked is never queued."""
         self._require_module()
         self._roots()
         path = self.output_dir / "two.png"
-        write_png(path, prompt=ambiguous_graph("cat", "dog"))
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
         prompt_id = self.api.db.save_prompt(
             text="latest", prompt_hash=generate_prompt_hash("latest")
         )
-        metadata = {"prompt": ambiguous_graph("cat", "dog")}
+        metadata = {"prompt": unlinkable_graph("cat", "dog")}
         context = {"id": prompt_id, "text": "latest", "fallback": True}
 
         handler = self._handler()
@@ -779,13 +811,18 @@ class TestMonitorRecording(ReviewTestCase):
         self.assertEqual(self._items(), [])
 
         other = self.output_dir / "three.png"
-        write_png(other, prompt=ambiguous_graph("cat", "dog"))
-        handler.review_recorder = self.api.record_review_file
+        write_png(other, prompt=unlinkable_graph("cat", "dog"))
+        calls = []
+
+        def recorder(image_path):
+            calls.append(self.api.record_review_file(image_path))
+            return calls[-1]
+
+        handler.review_recorder = recorder
         handler.link_image_to_prompt(str(other), context, metadata)
         self.assertEqual(self._linked_prompt(other), prompt_id)
-        items = self._items()
-        self.assertEqual(len(items), 1)
-        self.assertEqual(items[0]["sha256"], digest(other))
+        self.assertEqual(calls, ["linked"])
+        self.assertEqual(self._items(), [])
         self.assertEqual(self._capture_files(), [])
 
     async def test_monitor_skips_resolved_and_files_outside_the_roots(self):
@@ -822,7 +859,7 @@ class TestMonitorRecording(ReviewTestCase):
 
         self._roots()
         path = self.output_dir / "held.png"
-        write_png(path, prompt=ambiguous_graph("cat", "dog"))
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
 
         for kind, admission in (
             (RESTORE_JOB_KIND, "records"),
@@ -912,7 +949,7 @@ class TestRecordedSpelling(ReviewTestCase):
         self._require_module()
         self._roots()
         upper = self.output_dir / "Two.PNG"
-        write_png(upper, prompt=ambiguous_graph("cat", "dog"))
+        write_png(upper, prompt=unlinkable_graph("cat", "dog"))
         with _case_folding():
             self.assertEqual(self.api.record_review_file(str(upper)), "recorded")
             (before,) = self._items()
@@ -992,7 +1029,7 @@ class TestCaptureStoreShutdown(ReviewTestCase):
         self._require_module()
         self._roots()
         path = self.output_dir / "two.png"
-        write_png(path, prompt=ambiguous_graph("cat", "dog"))
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
         await self.api.cleanup_annotation_captures(None)  # the shutdown hook
         self.assertIsNone(self.api._captures)
         with patch("utils.media_capture.tempfile.mkdtemp") as mkdtemp:

tests/test_review_outcome.py

@@ -0,0 +1,372 @@
+"""Needs review takes only files whose final outcome is ambiguous and unlinked.
+
+The maintainer's live scan (#284 / #288) queued every file the strict
+resolver found ambiguous, also those the legacy browser-parser fallback then
+linked and those already linked, so the queue (10,232) was far larger than
+the scan's Needs review group. The rule now: a file is queued only when the
+resolver finds it ambiguous, the fallback does not link it and no saved row
+links it. A pending, undecided item whose file is linked (by the fallback, a
+later resolution or the monitor) is removed on the next scan or monitor
+pass; an item that carries a decision is never touched.
+"""
+
+import json
+import logging
+import os
+import sqlite3
+import unittest
+from unittest.mock import patch
+
+try:
+    from tests.test_review import (
+        DELIMITERS,
+        NOW,
+        ReviewTestCase,
+        ambiguous_graph,
+        resolved_graph,
+        unlinkable_graph,
+        write_png,
+    )
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_review import (
+        DELIMITERS,
+        NOW,
+        ReviewTestCase,
+        ambiguous_graph,
+        resolved_graph,
+        unlinkable_graph,
+        write_png,
+    )
+
+from utils.hashing import generate_prompt_hash
+from utils.image_monitor import ImageGenerationHandler
+from utils.prompt_tracker import PromptTracker
+from utils import review_capture
+
+
+def nan_stamped(graph):
+    """ComfyUI's ``is_changed: [NaN]`` stamp (written as a bare NaN literal)."""
+    for node in graph.values():
+        node["is_changed"] = [float("nan")]
+    return graph
+
+
+class OutcomeTestCase(ReviewTestCase):
+    def _pending(self):
+        return self.api.db.library_counts()["review_pending"]
+
+    def _record(self, name):
+        """An item as the pre-fix producers recorded it (any ambiguous file)."""
+        outcome = review_capture.record_ambiguous_file(
+            self.api.db, self.api._captures, self.output_dir, 0, name
+        )
+        self.assertIn(outcome, ("recorded", "limited"))
+
+    def _decide(self, item, outcome):
+        conn = self.api.db.model.get_connection()
+        candidate = None
+        if outcome == "selected":
+            candidate = json.loads(item["evidence"])["candidates"][0]["hash"]
+        with conn:
+            conn.execute(
+                "UPDATE media_review_items SET state=?, revision=1 WHERE id=?",
+                (outcome, item["id"]),
+            )
+            conn.execute(
+                "INSERT INTO media_review_decisions(id,item_id,decision_revision,"
+                "created_at,outcome,candidate_hash,evidence) VALUES(?,?,?,?,?,?,?)",
+                (
+                    f"00000000-0000-4000-8000-{len(self._decisions()) + 1:012d}",
+                    item["id"],
+                    1,
+                    NOW,
+                    outcome,
+                    candidate,
+                    '{"kind":"user-selection"}',
+                ),
+            )
+
+    def _decisions(self):
+        conn = self.api.db.model.get_connection()
+        return [
+            tuple(row)
+            for row in conn.execute(
+                "SELECT id,item_id,outcome FROM media_review_decisions ORDER BY id"
+            )
+        ]
+
+    def _link(self, path, text="already saved"):
+        prompt_id = self.api.db.save_prompt(
+            text=text, prompt_hash=generate_prompt_hash(text)
+        )
+        self.api.db.link_image_to_prompt(prompt_id, str(path))
+        return prompt_id
+
+    @staticmethod
+    def _rows(job, group):
+        return sorted(row["path"] for row in job["outcomes"].get(group, []))
+
+
+class TestOnlyUnlinkedAmbiguousFilesAreQueued(OutcomeTestCase):
+    async def test_nan_stamped_graph_the_fallback_links_is_not_queued(self):
+        self._roots()
+        path = self.output_dir / "nan.png"
+        write_png(path, prompt=nan_stamped(ambiguous_graph("cat", "dog")))
+        self.assertIn(b"NaN", path.read_bytes())
+
+        job = await self._scan()
+
+        self.assertEqual(job["counters"]["added"], 1, job["counters"])
+        self.assertEqual(self._items(), [])
+        self.assertEqual(job["scope"]["review_items"], 0)
+        self.assertEqual(self._pending(), 0)
+
+    async def test_fallback_link_to_a_known_prompt_is_not_queued(self):
+        self._roots()
+        write_png(self.output_dir / "a.png", prompt=ambiguous_graph("cat", "dog"))
+        await self._scan()
+        write_png(self.output_dir / "b.png", prompt=ambiguous_graph("cat", "dog"))
+
+        job = await self._scan()
+
+        self.assertEqual(self._items(), [])
+        self.assertEqual(job["scope"]["review_items"], 0)
+
+    async def test_already_linked_file_is_not_queued(self):
+        self._roots()
+        path = self.output_dir / "linked.png"
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
+        self._link(path)
+
+        job = await self._scan()
+
+        self.assertEqual(self._items(), [])
+        self.assertEqual(job["scope"]["review_items"], 0)
+        self.assertEqual(self._rows(job, "review"), [])
+        self.assertEqual(self._rows(job, "known"), ["linked.png"])
+
+    async def test_genuinely_ambiguous_file_is_queued(self):
+        self._roots()
+        path = self.output_dir / "two.png"
+        write_png(path, prompt=nan_stamped(unlinkable_graph("cat", "dog")))
+        # Beside it, a file the fallback links: only the first is queued.
+        write_png(self.output_dir / "linked.png", prompt=ambiguous_graph("a", "b"))
+
+        job = await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(os.path.basename(item["locator"]), "two.png")
+        self.assertEqual(item["state"], "pending")
+        self.assertEqual(item["candidate_count"], 3)
+        self.assertEqual(job["scope"]["review_items"], 1)
+        self.assertEqual(self._rows(job, "review"), ["two.png"])
+        self.assertEqual(self._pending(), 1)
+        self.assertIn("? 1 image added to Needs review", "\n".join(job["log"]))
+
+    async def test_counter_and_queue_agree_with_the_review_group(self):
+        """``review_items`` and the queue total equal the review group less
+        its unresolved files, on the first scan and on a rescan."""
+        self._roots()
+        for name in ("u1.png", "u2.png"):
+            write_png(self.output_dir / name, prompt=unlinkable_graph("cat", name))
+        write_png(self.output_dir / "delims.png", parameters=DELIMITERS)
+        write_png(self.output_dir / "linked.png", prompt=ambiguous_graph("a", "b"))
+        unresolved = {"1": {"class_type": "KSampler", "inputs": {}}}
+        write_png(self.output_dir / "none.png", prompt=unresolved)
+        write_png(self.output_dir / "one.png", prompt=resolved_graph())
+
+        for _ in range(2):
+            job = await self._scan()
+            review = self._rows(job, "review")
+            unresolved_rows = sorted(
+                row["path"]
+                for row in job["outcomes"]["review"]
+                if "could not be traced" in row["detail"]
+            )
+            queued = [name for name in review if name not in unresolved_rows]
+            self.assertEqual(queued, ["delims.png", "u1.png", "u2.png"], job)
+            self.assertEqual(job["scope"]["review_items"], 3)
+            self.assertEqual(self._pending(), 3)
+        self.assertIn("? 3 images already in Needs review", "\n".join(job["log"]))
+
+
+class TestQueueSelfHeals(OutcomeTestCase):
+    async def test_pending_item_of_a_fallback_linked_file_is_removed(self):
+        """The maintainer's case: queued by the old rule, linked on rescan."""
+        self._roots()
+        path = self.output_dir / "nan.png"
+        write_png(path, prompt=nan_stamped(ambiguous_graph("cat", "dog")))
+        self._record("nan.png")
+        revision = self._review_revision()
+        self.assertEqual(self._pending(), 1)
+
+        job = await self._scan()
+
+        self.assertEqual(self._items(), [])
+        self.assertEqual(self._pending(), 0)
+        self.assertGreater(self._review_revision(), revision)
+        self.assertEqual(job["scope"]["review_settled"], 1)
+        self.assertIn("= 1 image left Needs review · now linked", "\n".join(job["log"]))
+
+    async def test_pending_item_of_an_already_linked_file_is_removed(self):
+        self._roots()
+        path = self.output_dir / "linked.png"
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
+        self._record("linked.png")
+        self._link(path)
+
+        await self._scan()
+
+        self.assertEqual(self._items(), [])
+        self.assertEqual(self._pending(), 0)
+
+    async def test_pending_item_of_a_skipped_known_file_is_removed(self):
+        """A file the "new" scope skips as known still settles its old item."""
+        self._roots()
+        path = self.output_dir / "one.png"
+        write_png(path, prompt=unlinkable_graph("cat", "dog"))
+        self._record("one.png")
+        write_png(path, prompt=resolved_graph())
+        first = await self._scan()
+        self.assertEqual(self._rows(first, "added"), ["one.png"])
+        # An item a pre-fix scan left for the file's old bytes.
+        self.assertEqual(self._items(), [])
+        write_png(path, prompt=unlinkable_graph("cat", "dog"), color=(0, 9, 0))
+        self._record("one.png")
+        write_png(path, prompt=resolved_graph())
+        second = await self._scan()
+        self.assertEqual(self._rows(second, "known"), ["one.png"])
+        self.assertEqual(self._items(), [])
+
+    async def test_a_failed_settle_is_logged_and_counted_not_fatal(self):
+        self._roots()
+        write_png(self.output_dir / "nan.png", prompt=ambiguous_graph("cat", "dog"))
+        self._record("nan.png")
+
+        with (
+            patch.object(
+                self.api.db,
+                "settle_linked_review_items",
+                side_effect=sqlite3.OperationalError("database is locked"),
+            ),
+            self.assertLogs("prompt_manager.api", logging.WARNING) as logs,
+        ):
+            job = await self._scan()
+
+        self.assertEqual(job["state"], "finished")
+        self.assertEqual(job["scope"]["review_errors"], 1)
+        self.assertEqual(job["scope"]["review_settled"], 0)
+        self.assertTrue(any("could not be removed" in line for line in logs.output))
+        self.assertEqual(self._pending(), 1)
+        # The next scan settles it.
+        await self._scan()
+        self.assertEqual(self._pending(), 0)
+
+    async def test_still_unlinked_pending_item_stays(self):
+        self._roots()
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
+        await self._scan()
+        (item,) = self._items()
+
+        job = await self._scan()
+
+        self.assertEqual([i["id"] for i in self._items()], [item["id"]])
+        self.assertEqual(job["scope"].get("review_settled", 0), 0)
+
+    async def test_decided_items_are_never_touched(self):
+        self._roots()
+        for name in ("sel.png", "unres.png"):
+            write_png(self.output_dir / name, prompt=ambiguous_graph("cat", name))
+            self._record(name)
+        items = {os.path.basename(i["locator"]): i for i in self._items()}
+        self._decide(items["sel.png"], "selected")
+        self._decide(items["unres.png"], "unresolved")
+        before = (self._items(), self._decisions())
+        # A pending neighbour the fallback links: it alone leaves the queue.
+        write_png(self.output_dir / "pending.png", prompt=ambiguous_graph("x", "y"))
+        self._record("pending.png")
+        self.assertEqual(self._pending(), 1)
+
+        job = await self._scan("all")
+
+        self.assertEqual((self._items(), self._decisions()), before)
+        self.assertEqual(self._pending(), 0)
+        self.assertEqual(job["scope"]["review_settled"], 1)
+
+
+class TestMonitorRule(OutcomeTestCase):
+    def _handler(self):
+        handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
+        handler.review_recorder = self.api.record_review_file
+        return handler
+
+    def _context(self):
+        prompt_id = self.api.db.save_prompt(
+            text="latest", prompt_hash=generate_prompt_hash("latest")
+        )
+        return {"id": prompt_id, "text": "latest", "fallback": True}
+
+    async def test_monitor_linked_ambiguous_output_is_not_queued(self):
+        self._roots()
+        path = self.output_dir / "two.png"
+        graph = unlinkable_graph("cat", "dog")
+        write_png(path, prompt=graph)
+
+        self._handler().link_image_to_prompt(
+            str(path), self._context(), {"prompt": graph}
+        )
+
+        self.assertEqual(self._items(), [])
+
+    async def test_monitor_link_removes_the_files_pending_item(self):
+        self._roots()
+        path = self.output_dir / "two.png"
+        graph = unlinkable_graph("cat", "dog")
+        write_png(path, prompt=graph)
+        self._record("two.png")
+        self.assertEqual(self._pending(), 1)
+
+        self._handler().link_image_to_prompt(
+            str(path), self._context(), {"prompt": graph}
+        )
+
+        self.assertEqual(self._items(), [])
+        self.assertEqual(self._pending(), 0)
+
+    async def test_monitor_never_touches_a_decided_item(self):
+        self._roots()
+        path = self.output_dir / "two.png"
+        graph = unlinkable_graph("cat", "dog")
+        write_png(path, prompt=graph)
+        self._record("two.png")
+        (item,) = self._items()
+        self._decide(item, "unresolved")
+        before = (self._items(), self._decisions())
+        other = self.output_dir / "other.png"
+        write_png(other, prompt=graph, color=(0, 0, 9))
+        self._record("other.png")
+        handler, context = self._handler(), self._context()
+
+        handler.link_image_to_prompt(str(path), context, {"prompt": graph})
+        handler.link_image_to_prompt(str(other), context, {"prompt": graph})
+
+        self.assertEqual((self._items(), self._decisions()), before)
+        self.assertEqual(self._pending(), 0)
+
+    async def test_record_review_file_applies_the_scan_rule(self):
+        self._roots()
+        write_png(self.output_dir / "fallback.png", prompt=ambiguous_graph("a", "b"))
+        write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
+        linked = self.output_dir / "linked.png"
+        write_png(linked, prompt=unlinkable_graph("cat", "dog"))
+        self._link(linked)
+
+        record = self.api.record_review_file
+        self.assertEqual(record(str(self.output_dir / "fallback.png")), "skipped")
+        self.assertEqual(record(str(linked)), "linked")
+        self.assertEqual(record(str(self.output_dir / "two.png")), "recorded")
+        self.assertEqual(len(self._items()), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/image_monitor.py

@@ -75,8 +75,9 @@ class ImageGenerationHandler(FileSystemEventHandler):
         self.prompt_tracker = prompt_tracker
         self.metadata_extractor = ComfyUIMetadataExtractor()
         self.logger = get_logger("prompt_manager.image_monitor")
-        # Needs review recording for ambiguous outputs (the API's
-        # ``record_review_file``, wired at start-up); None records nothing.
+        # The Needs review pass for ambiguous outputs (the API's
+        # ``record_review_file``, wired at start-up): a linked output is
+        # never queued and its pending items leave the queue. None: no pass.
         self.review_recorder: Optional[Callable[[str], str]] = None
 
         # Read from GalleryConfig if available, otherwise use defaults
@@ -587,10 +588,11 @@ class ImageGenerationHandler(FileSystemEventHandler):
             self._record_review(image_path)
 
     def _record_review(self, image_path: str) -> None:
-        """Queue an ambiguous output for Needs review, best effort.
+        """The Needs review pass for an ambiguous output, best effort.
 
-        The fallback link above stays whatever happens here: a failure is
-        logged, never raised.
+        The output is linked by now, so it is never queued; a pending item
+        an earlier scan left for its path leaves the queue. The link above
+        stays whatever happens here: a failure is logged, never raised.
         """
         recorder = getattr(self, "review_recorder", None)
         if recorder is None:

utils/review_capture.py

@@ -53,15 +53,17 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st
     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
-    bytes an item cites, so it is not even copied) or ``skipped`` (the file
+    bytes an item cites, so it is not even copied, and that item is still
+    pending), ``decided`` (as ``duplicate``, but the item carries a decision,
+    so the file is not in the queue) or ``skipped`` (the file
     changed, vanished, is not ambiguous in its captured bytes, or storage was
     restored meanwhile).
     ``CaptureCapacity`` and database errors propagate to the caller, which
     logs and counts them.
     """
-    spelled_path = _queued_spelling(db, root, relative)
-    if spelled_path is None:
-        return "duplicate"
+    spelled_path, queued = _queued_spelling(db, root, relative)
+    if queued is not None:
+        return _already_queued(db, *queued)
     epoch = db.get_execution_epoch()
     try:
         # Background: the last worker stays free for an interactive request.
@@ -102,12 +104,22 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st
     except LibraryCommandError:
         return "skipped"
     if not inserted:
-        return "duplicate"
+        return _already_queued(db, capture.locator, capture.size, capture.mtime_ns)
     return "limited" if limited else "recorded"
 
 
-def _queued_spelling(db, root, relative: str) -> str | None:
-    """The file's spelling, or None when its current stat is already queued.
+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"
+    return "decided"
+
+
+def _queued_spelling(db, root, relative: str) -> 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
+    queued.
 
     The spelling is the capture's own candidate path (the real root joined
     with the listed relative path), so it folds to the capture's locator. A
@@ -121,7 +133,7 @@ def _queued_spelling(db, root, relative: str) -> str | None:
     try:
         facts = os.lstat(spelled)
     except OSError:
-        return spelled
+        return spelled, None
     locator = os.path.normcase(spelled)
     if db.review_fingerprint_queued(
         locator, facts.st_size, facts.st_mtime_ns, RESOLVER_VERSION
@@ -129,8 +141,8 @@ def _queued_spelling(db, root, relative: str) -> str | None:
         # Not copied again, but a case-only rename keeps the stat: record
         # the current spelling so the item's target still opens.
         db.refresh_review_spelling(locator, spelled)
-        return None
-    return spelled
+        return spelled, (locator, facts.st_size, facts.st_mtime_ns)
+    return spelled, None
 
 
 def captured_review(store, capture) -> tuple[dict, dict] | None: