diff --git a/TESTING.md b/TESTING.md index ec44240..6d377dd 100644 --- a/TESTING.md +++ b/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 diff --git a/py/api/jobs.py b/py/api/jobs.py index 568a99e..de12e8d 100644 --- a/py/api/jobs.py +++ b/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 diff --git a/py/api/review_recording.py b/py/api/review_recording.py index 36a665c..22d0c16 100644 --- a/py/api/review_recording.py +++ b/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( diff --git a/tests/test_review.py b/tests/test_review.py index f00d392..92c471b 100644 --- a/tests/test_review.py +++ b/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