fix(review): refuse monitor recording beside record-changing jobs

System S1 (#289) gives every record change a registry admission class. The
scan already records Needs review items under its own records job. The
monitor's recording lasts one file and is not a job, so it now checks the
records admission: while a records or guarded job or hold runs it records
nothing, logs that the file waits for the next scan, and answers deferred.
The fallback link stands and the next "new" scan reads the ambiguous file
again, because a fallback link is not a resolved one.

JobRegistry.blocker and active iterate a snapshot of the jobs, since the
monitor's worker thread now asks them too.

TESTING.md

@@ -2014,7 +2014,12 @@ capacity). 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. A new item moves `review_index_revision`. A
+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
+next scan, and answers `deferred`; the next "new" scan reads the ambiguous file
+again (`test_monitor_recording_is_refused_beside_record_changing_jobs`). A new item moves `review_index_revision`. A
 rescan does not copy or hash an ambiguous file whose current size and
 `mtime_ns` equal the verified fingerprint of bytes an item already cites (a
 stat compare); the file still stays in the default "new" scope, because it is

py/api/jobs.py

@@ -432,7 +432,8 @@ class JobRegistry:
         if same_kind is not None:
             return same_kind
         refused = _REFUSED_BESIDE[resolve_admission(admission, record_changing)]
-        running = [job for job in self._jobs.values() if job.running]
+        # A snapshot: the monitor's worker thread asks too (Needs review).
+        running = [job for job in tuple(self._jobs.values()) if job.running]
         running.sort(key=lambda job: job._started_seq)
         for job in running:
             if job.admission in refused:
@@ -463,8 +464,8 @@ class JobRegistry:
         return self._jobs.get(job_id)
 
     def active(self, kind):
-        """The running job of ``kind``, or None."""
-        for job in self._jobs.values():
+        """The running job of ``kind``, or None (safe off the event loop)."""
+        for job in tuple(self._jobs.values()):
             if job.kind == kind and job.running:
                 return job
         return None

py/api/review_recording.py

@@ -5,6 +5,12 @@ The scan hands every file the ingestion resolver finds ambiguous to
 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.
+
+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
+``records`` or ``guarded`` job runs (``deferred``), and the next scan reads
+the ambiguous file again.
 """
 
 import os
@@ -12,6 +18,9 @@ from pathlib import Path
 
 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"
+
 try:
     from ...utils.review_capture import record_ambiguous_file
 except ImportError:
@@ -50,9 +59,18 @@ class ReviewRecordingMixin:
         """Needs review recording for the live monitor (its worker thread).
 
         Maps ``image_path`` to the gallery root that lists it (the listing's
-        ``root_index``); a file under no current root is ``skipped``. Errors
-        propagate: the monitor logs them and keeps its link.
+        ``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.
         """
+        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"
+            )
+            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(

tests/test_review.py

@@ -756,6 +756,45 @@ class TestMonitorRecording(ReviewTestCase):
         self.assertEqual(self.api.record_review_file(str(outside)), "skipped")
         self.assertEqual(self._items(), [])
 
+    async def test_monitor_recording_is_refused_beside_record_changing_jobs(self):
+        """S1 admission: a ``records`` recording waits for no job; it is refused.
+
+        While a ``records`` or ``guarded`` job runs, the monitor's recording
+        records nothing and answers ``deferred`` (its fallback link stands and
+        a later scan reads the ambiguous file again); a ``read`` job does not
+        refuse it, and it records once the blocking job has ended.
+        """
+        self._require_module()
+        from py.api.jobs import RESTORE_JOB_KIND, THUMBNAILS_JOB_KIND
+
+        self._roots()
+        path = self.output_dir / "held.png"
+        write_png(path, prompt=ambiguous_graph("cat", "dog"))
+
+        for kind, admission in (
+            (RESTORE_JOB_KIND, "records"),
+            (THUMBNAILS_JOB_KIND, "guarded"),
+        ):
+            with self.subTest(admission=admission):
+                async with self.api._jobs.hold(
+                    kind, f"{kind} (admin page)", admission=admission
+                ):
+                    with self.assertLogs("prompt_manager", logging.INFO) as logs:
+                        outcome = await asyncio.to_thread(
+                            self.api.record_review_file, str(path)
+                        )
+                self.assertEqual(outcome, "deferred")
+                self.assertEqual(self._items(), [])
+                self.assertTrue(
+                    any(f"{kind} (admin page)" in line for line in logs.output),
+                    logs.output,
+                )
+
+        async with self.api._jobs.hold("probe-read", "Read probe", admission="read"):
+            outcome = await asyncio.to_thread(self.api.record_review_file, str(path))
+        self.assertEqual(outcome, "recorded")
+        self.assertEqual(len(self._items()), 1)
+
     async def test_monitor_wiring_reaches_new_and_existing_handlers(self):
         self._require_module()
         from utils.image_monitor import ImageMonitor