feat(review): re-read an open Needs review page when its queue changes

An open Needs review page now follows the queue without polling. A new
server-sent event stream, GET /prompt_manager/review/events, sends
"ready" when it opens and "change" after every committed queue change:
an item the scan or the monitor recorded, a spelling refresh, a decision
from any tab, and a restore. database/review_changes.py notifies after
the commit from whichever thread wrote it. The page re-reads its first
page on each event, keeps the open item, its choice and an unanswered
decision, and re-reads the SAVED VIEWS count with it. When the open item
left the queue, the next one opens with a note. A keepalive comment
every 25 s lets the server drop a closed tab; shutdown ends every stream.

Leaving the page mid-decision now releases that decision's capture.

D-NEW-F1 is renumbered D119 in markers and TESTING.md.

TESTING.md

@@ -1618,7 +1618,7 @@ pending count (`GET /prompt_manager/review/items?limit=1`, its `total`).
   carries the concept's **Review metadata** (app.js a738 `go-review`): in its Activity
   item after the item's line, and in its job view's heading after Download report. The
   item's chip still names its state (D54), where the concept's says "Needs attention".
-- **D-NEW-F1** (number to be assigned): the queue cards have no selection checkbox and
+- **D119**: the queue cards have no selection checkbox and
   no star (the page has no selection bar and the queue carries no favorite state); a
   file under no current root has an empty, `aria-disabled` image button and Tag; more
   than 40 items add the Images page's `.infinite-continuation` ("40 of 45 images",
@@ -1626,7 +1626,31 @@ pending count (`GET /prompt_manager/review/items?limit=1`, its `total`).
   line (with Retry); a limited or unavailable file and the decision states are text in
   the panel's `p.warning-note`; the toasts "Association accepted." and "Left
   unresolved." replace the concept's prototype toast. The concept's "Sample review
-  queue" chip and "sample" wording are prototype copy (D29).
+  queue" chip and "sample" wording are prototype copy (D29). When a re-read (below)
+  finds the open item gone, the next one opens with "<file> was decided elsewhere, so
+  the next image is open."; a failed re-read keeps the queue and says "The review queue
+  could not be read again. Reload the page to see the latest images."
+
+**Live queue (F2).** An open Needs review page holds one server-sent event stream,
+`GET /prompt_manager/review/events` (`py/api/review_events.py`): `event: ready` when it
+opens, then `event: change` after each committed queue change (an item the scan or the
+monitor recorded, a spelling refresh, a decision from any tab; `database/review_changes.py`
+notifies after the commit, from whichever thread wrote it) and after a restore. Events
+carry no data. The page re-reads its first page on every `ready` and `change`, keeping
+the open item, its choice and an unanswered decision; re-reads asked for while one runs
+collapse into one more, and one waits while a decision is in flight. The row's count is
+re-read with it. Nothing polls: a `: keepalive` comment every 25 s only lets the server
+drop a closed tab's subscription. Shutdown (`on_shutdown`) ends every stream and answers
+a later one 503; the browser's own EventSource reconnection then stops. A duplicate
+recording and a refused decision change nothing and send nothing. Residual: the stream
+is one HTTP/1.1 connection per open Needs review page (browsers allow six per origin), so
+it lives only while the page is open; the re-read replaces Load more's extra pages with
+the first 40.
+
+Pinned by `tests/test_api_review_events.py`, `tests/js/workspace-review-state.test.js`
+("refresh …") and `tests/browser/workspace-review.spec.mjs` ("an open page shows …").
+Leaving the page mid-decision releases that decision's capture (the server may never see
+the request).
 - **D3** and **D20** apply to the queue grid (its cards stay in the grid's flow; it is
   not virtualized) and to the panel's prompt blocks (their mount point is transparent).
 

__init__.py

@@ -91,6 +91,9 @@ try:
     if host_app is not None and hasattr(host_app, "on_cleanup"):
         host_app.on_cleanup.append(api.cleanup_history_reads)
         host_app.on_cleanup.append(api.cleanup_annotation_captures)
+    if host_app is not None and hasattr(host_app, "on_shutdown"):
+        # Open Needs review change streams end before the server waits on them.
+        host_app.on_shutdown.append(api.close_review_events)
 
 except Exception as e:
     # Log to internal logging system without console spam

database/review.py

@@ -31,6 +31,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_changes import notify_review_change
 from .review_schema import (
     ITEMS_TABLE,
     MAX_CANDIDATES,
@@ -286,7 +287,9 @@ class ReviewOperations:
             refreshed = _refresh_spelling(conn, locator, spelled_path)
             if inserted or refreshed:
                 conn.execute(_BUMP_REVIEW_INDEX)
-            return inserted
+        if inserted or refreshed:
+            notify_review_change(self.model.db_path)
+        return inserted
 
     def refresh_review_spelling(self, locator: str, spelled_path: str) -> bool:
         """Record ``spelled_path`` as the spelling of every item at ``locator``.
@@ -303,7 +306,9 @@ class ReviewOperations:
             refreshed = _refresh_spelling(conn, locator, spelled_path)
             if refreshed:
                 conn.execute(_BUMP_REVIEW_INDEX)
-            return refreshed
+        if refreshed:
+            notify_review_change(self.model.db_path)
+        return refreshed
 
     def review_index_revision(self) -> int:
         """The queue's revision (moved by every queue change)."""

database/review_changes.py

@@ -0,0 +1,64 @@
+"""Needs review queue change notices (#249 PR F2).
+
+Every committed change to the pending queue (an item recorded by the scan or
+the monitor, a spelling refresh, a decision) calls ``notify_review_change``
+after its transaction, from whichever thread wrote it. The review event
+stream (``py/api/review_events.py``) subscribes, so an open Needs review page
+re-reads its queue when it changes instead of polling. Observers are keyed by
+the database file, like ``subscribe_history_restore``; an observer that fails
+is logged and never fails the write.
+"""
+
+import os
+import threading
+from typing import Callable
+
+try:
+    from ..utils.logging_config import get_logger
+except ImportError:
+    from utils.logging_config import get_logger
+
+_logger = get_logger("prompt_manager.database.review_changes")
+_lock = threading.Lock()
+_observers: dict[str, dict[object, Callable[[], None]]] = {}
+
+
+def _identity(db_path: str) -> str:
+    return os.path.normcase(os.path.realpath(db_path))
+
+
+def subscribe_review_changes(
+    db_path: str, callback: Callable[[], None]
+) -> Callable[[], None]:
+    """Call ``callback()`` after each committed queue change; return the unsubscribe."""
+    identity = _identity(db_path)
+    token = object()
+    with _lock:
+        _observers.setdefault(identity, {})[token] = callback
+
+    def unsubscribe() -> None:
+        with _lock:
+            observers = _observers.get(identity)
+            if observers is not None:
+                observers.pop(token, None)
+                if not observers:
+                    _observers.pop(identity, None)
+
+    return unsubscribe
+
+
+def observer_count(db_path: str) -> int:
+    """How many observers watch ``db_path`` (tests and diagnostics)."""
+    with _lock:
+        return len(_observers.get(_identity(db_path), {}))
+
+
+def notify_review_change(db_path: str) -> None:
+    """Tell every observer of ``db_path``; call it after the commit."""
+    with _lock:
+        callbacks = tuple(_observers.get(_identity(db_path), {}).values())
+    for callback in callbacks:
+        try:
+            callback()
+        except Exception:
+            _logger.warning("Needs review change observer failed", exc_info=True)

database/review_decisions.py

@@ -29,6 +29,7 @@ from .review import (
     dependent_images,
     same_name_rows,
 )
+from .review_changes import notify_review_change
 from .review_schema import DECISIONS_TABLE, ITEMS_TABLE, canonical_evidence
 
 DECISION_KIND = "user-selection"
@@ -119,7 +120,7 @@ class ReviewDecisionOperations:
             raise ValueError("A selected decision needs a candidate and a file")
         try:
             with self._pair_transaction() as conn:
-                return self._decide_in(
+                decided = self._decide_in(
                     conn,
                     item_id=item_id,
                     expected_epoch=expected_epoch,
@@ -135,6 +136,9 @@ class ReviewDecisionOperations:
                 )
         except _Rollback as rollback:
             raise rollback.error from None
+        # After the commit: open Needs review pages re-read their queue.
+        notify_review_change(self.model.db_path)
+        return decided
 
     def _decide_in(self, conn: sqlite3.Connection, **command) -> dict:
         epoch = self._library_epoch(conn, command["expected_epoch"])

py/api/__init__.py

@@ -9,6 +9,7 @@ Split into domain-specific mixins for maintainability:
 - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py)
 - ReviewRoutesMixin: the Needs review queue, item detail and decisions
 - ReviewRecordingMixin: Needs review recording for the scan and the monitor
+- ReviewEventsMixin: the Needs review change stream an open page re-reads on
 """
 
 import asyncio
@@ -31,6 +32,7 @@ from .vocabulary import VocabularyRoutesMixin
 from .annotations import AnnotationRoutesMixin
 from .review import ReviewRoutesMixin
 from .review_recording import ReviewRecordingMixin
+from .review_events import ReviewEventsMixin
 from .saved_views import SavedViewRoutesMixin
 from .history import HistoryRoutesMixin
 from .images import ImageRoutesMixin
@@ -260,6 +262,7 @@ class PromptManagerAPI(
     AnnotationRoutesMixin,
     ReviewRoutesMixin,
     ReviewRecordingMixin,
+    ReviewEventsMixin,
     SavedViewRoutesMixin,
     HistoryRoutesMixin,
     PromptRoutesMixin,
@@ -457,6 +460,7 @@ class PromptManagerAPI(
         self._register_vocabulary_routes(routes)
         self._register_annotation_routes(routes)
         self._register_review_routes(routes)
+        self._register_review_event_routes(routes)
         self._register_saved_view_routes(routes)
         self._register_history_routes(routes)
         self._register_prompt_routes(routes)

py/api/review_events.py

@@ -0,0 +1,130 @@
+"""Needs review change stream (#249 PR F2).
+
+``GET /prompt_manager/review/events`` is a server-sent event stream for an
+open Needs review page: ``event: ready`` when it opens, then
+``event: change`` after each committed queue change (an item the scan or the
+monitor recorded, a decision from any tab, a spelling refresh) and after a
+restore. Notices that arrive while one is being written collapse into one.
+The page re-reads its queue on each; nothing polls. The events carry no data
+(``{}``), so nothing about the queue is pushed.
+
+A ``: keepalive`` comment every ``HEARTBEAT_SECONDS`` lets the server notice
+a closed tab (its write fails) and drop the subscription; clients ignore
+comments. Shutdown (``close_review_events``, an ``on_shutdown`` hook) ends
+every stream, and a stream asked for afterwards is a 503.
+"""
+
+import asyncio
+
+from aiohttp import web
+
+try:
+    from ...database.execution_history import subscribe_history_restore
+    from ...database.review_changes import subscribe_review_changes
+except ImportError:
+    from database.execution_history import subscribe_history_restore
+    from database.review_changes import subscribe_review_changes
+
+EVENTS_URL = "/prompt_manager/review/events"
+HEARTBEAT_SECONDS = 25
+_READY = b"event: ready\ndata: {}\n\n"
+_CHANGE = b"event: change\ndata: {}\n\n"
+_KEEPALIVE = b": keepalive\n\n"
+
+
+class _Stream:
+    """One open stream: a change flag set from any thread, and a close flag."""
+
+    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
+        self._loop = loop
+        self.changed = asyncio.Event()
+        self.closed = asyncio.Event()
+
+    def wake(self) -> None:
+        """Thread-safe: mark a change (the loop may already be gone)."""
+        try:
+            self._loop.call_soon_threadsafe(self.changed.set)
+        except RuntimeError:  # the loop closed
+            pass
+
+    async def next(self, timeout: float) -> str:
+        """``change``, ``closed`` or ``idle`` (nothing within ``timeout``)."""
+        waits = {
+            asyncio.ensure_future(self.changed.wait()): "change",
+            asyncio.ensure_future(self.closed.wait()): "closed",
+        }
+        try:
+            done, _ = await asyncio.wait(
+                waits, timeout=timeout, return_when=asyncio.FIRST_COMPLETED
+            )
+        finally:
+            for waiter in waits:
+                waiter.cancel()
+        if self.closed.is_set():
+            return "closed"
+        if done:
+            self.changed.clear()
+            return "change"
+        return "idle"
+
+
+class ReviewEventsMixin:
+    """The Needs review change stream."""
+
+    def _register_review_event_routes(self, routes: web.RouteTableDef) -> None:
+        routes.get(EVENTS_URL)(self.review_events)
+
+    def _review_streams(self) -> set:
+        streams = getattr(self, "_open_review_streams", None)
+        if streams is None:
+            streams = self._open_review_streams = set()
+        return streams
+
+    def review_stream_count(self) -> int:
+        """How many change streams are open (tests and diagnostics)."""
+        return len(self._review_streams())
+
+    async def close_review_events(self, app) -> None:
+        """End every open stream; later requests get 503 (shutdown hook)."""
+        self._review_events_closed = True
+        for stream in list(self._review_streams()):
+            stream.closed.set()
+
+    async def review_events(self, request: web.Request) -> web.StreamResponse:
+        if getattr(self, "_review_events_closed", False):
+            return web.json_response(
+                {"success": False, "error": "The server is shutting down."},
+                status=503,
+            )
+        stream = _Stream(asyncio.get_running_loop())
+        db_path = self.db.model.db_path
+        unsubscribers = (
+            subscribe_review_changes(db_path, stream.wake),
+            subscribe_history_restore(db_path, stream.wake),
+        )
+        streams = self._review_streams()
+        streams.add(stream)
+        response = web.StreamResponse(
+            status=200,
+            headers={
+                "Content-Type": "text/event-stream",
+                "Cache-Control": "no-store",
+                "X-Accel-Buffering": "no",
+            },
+        )
+        try:
+            await response.prepare(request)
+            await response.write(_READY)
+            while True:
+                outcome = await stream.next(HEARTBEAT_SECONDS)
+                if outcome == "closed":
+                    break
+                await response.write(_CHANGE if outcome == "change" else _KEEPALIVE)
+            await response.write_eof()
+        except (ConnectionError, RuntimeError) as exc:  # the tab went away
+            self.logger.debug(f"Needs review stream closed ({exc})")
+        finally:
+            streams.discard(stream)
+            for unsubscribe in unsubscribers:
+                unsubscribe()
+        return response

tests/js/workspace-review-state.test.js

@@ -341,3 +341,126 @@ test('dispose releases a kept capture and stops reads', async () => {
     assert.equal(released[0].url, CAPTURE + 'capture-1');
     assert.equal(await c.retry(), false);
 });
+
+// F2: an open page re-reads its queue when the server says it changed (the review change stream).
+function refreshOf(c) {
+    assert.equal(typeof c.refresh, 'function', 'refresh is not implemented');
+    return c.refresh();
+}
+test('refresh re-reads the queue quietly and keeps the selection and choice', async () => {
+    let queue = QUEUE;
+    const { c, h } = await loaded({ [`GET ${ITEMS}`]: () => response(200, page(queue)) });
+    c.choose(HASH_B);
+    const statuses = [];
+    c.subscribe(state => statuses.push(state.status));
+    queue = [...QUEUE, item('three')];
+    await refreshOf(c);
+    await settle();
+    assert.deepEqual(c.state.items.map(entry => entry.id), ['one', 'two', 'three']);
+    assert.equal(c.state.total, 3);
+    assert.equal(c.state.selectedId, 'one');
+    assert.equal(c.state.choice, HASH_B);
+    assert.equal(c.state.warning, '');
+    assert.ok(statuses.every(status => status === 'ready'), `no loading state: ${statuses}`);
+    assert.equal(h.of('GET', `${ITEMS}/one`).length, 1, 'the open item is not read again');
+});
+
+test('refreshes asked for during a refresh collapse into one more read', async () => {
+    let release;
+    const held = new Promise(resolve => { release = resolve; });
+    let lists = 0;
+    const { c, h } = await loaded({ [`GET ${ITEMS}`]: async () => {
+        lists += 1;
+        if (lists === 2) await held;
+        return response(200, page(QUEUE));
+    } });
+    const first = refreshOf(c);
+    await settle();
+    refreshOf(c);
+    refreshOf(c);
+    release();
+    await first;
+    await settle();
+    assert.equal(h.of('GET', `${ITEMS}?`).length, 3, 'the load, the refresh and one collapsed re-read');
+    assert.equal(c.state.status, 'ready');
+});
+
+test('a refresh after the open item was decided elsewhere opens the next one and says so', async () => {
+    let queue = QUEUE;
+    const { c, h } = await loaded({
+        [`GET ${ITEMS}`]: () => response(200, page(queue)),
+        [`GET ${ITEMS}/three`]: () => response(200, detailOf(item('three'))),
+    });
+    queue = [item('three')];
+    await refreshOf(c);
+    await settle();
+    assert.equal(c.state.selectedId, 'three');
+    assert.equal(c.state.detailStatus, 'ready');
+    assert.equal(c.state.warning, 'one.png was decided elsewhere, so the next image is open.');
+    assert.equal(h.of('GET', `${ITEMS}/three`).length, 1);
+});
+
+test('a refresh to an empty queue clears the selection', async () => {
+    let queue = QUEUE;
+    const { c } = await loaded({ [`GET ${ITEMS}`]: () => response(200, page(queue)) });
+    queue = [];
+    await refreshOf(c);
+    await settle();
+    assert.equal(c.state.total, 0);
+    assert.deepEqual(c.state.items, []);
+    assert.equal(c.state.selectedId, null);
+    assert.equal(c.state.status, 'ready');
+});
+
+test('a failed refresh keeps the queue and says so', async () => {
+    let lists = 0;
+    const { c } = await loaded({ [`GET ${ITEMS}`]: () => { lists += 1; return lists === 1 ? response(200, page(QUEUE)) : fail(500, 'review-failed'); } });
+    await refreshOf(c);
+    await settle();
+    assert.equal(c.state.status, 'ready');
+    assert.deepEqual(c.state.items.map(entry => entry.id), ['one', 'two']);
+    assert.equal(c.state.warning, 'The review queue could not be read again. Reload the page to see the latest images.');
+});
+
+test('dispose during a decision releases its capture', async () => {
+    const { c, h } = await loaded({ [`POST ${ITEMS}/one/decision`]: () => new Promise(() => {}) });
+    const applying = c.apply();
+    await settle();
+    assert.equal(c.state.phase, 'applying');
+    c.dispose();
+    assert.equal(await applying, false);
+    await settle();
+    assert.deepEqual(h.of('DELETE', CAPTURE).map(call => call.url), [CAPTURE + 'capture-1']);
+    await refreshOf(c);
+    assert.equal(h.of('GET', `${ITEMS}?`).length, 1, 'a disposed controller reads nothing');
+});
+
+test('a change heard during the first read re-reads once after it, and a failed first read stays an error', async () => {
+    const { createReviewController } = await load();
+    let release;
+    let lists = 0;
+    const h = harness(routes({ [`GET ${ITEMS}`]: async () => {
+        lists += 1;
+        if (lists === 1) await new Promise(resolve => { release = resolve; });
+        return response(200, page(QUEUE));
+    } }));
+    const c = createReviewController({ api: h.api });
+    const first = c.load();
+    await settle();
+    refreshOf(c);
+    assert.equal(lists, 1, 'the first read is not interrupted');
+    release();
+    await first;
+    await settle();
+    assert.equal(lists, 2, 'one re-read after the first read');
+    assert.equal(c.state.status, 'ready');
+
+    const failing = harness(routes({ [`GET ${ITEMS}`]: () => fail(500, 'review-failed') }));
+    const broken = createReviewController({ api: failing.api });
+    const reading = broken.load();
+    refreshOf(broken);
+    await reading;
+    await settle();
+    assert.equal(broken.state.status, 'error');
+    assert.equal(failing.of('GET', `${ITEMS}?`).length, 1, 'Retry is the way back, not a re-read');
+});

tests/test_api_review_events.py

@@ -0,0 +1,215 @@
+"""Needs review change notices (#249 PR F2).
+
+``GET /prompt_manager/review/events`` is a server-sent event stream: one
+``ready`` event when it opens, then one ``change`` event after every
+committed queue change (an item recorded by the scan or the monitor, a
+spelling refresh, a decision from any tab) and after a restore. An open
+Needs review page re-reads its queue on ``change``; nothing polls. A refused
+decision and a duplicate recording change nothing and send nothing.
+"""
+
+import asyncio
+import sqlite3
+import threading
+
+try:
+    from tests.test_api_review import ReviewRoutesTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_review import ReviewRoutesTestCase
+
+try:  # guarded so a missing module fails assertions, not test loading
+    from database import review_changes
+except ImportError:
+    review_changes = None
+try:
+    from py.api import review_events
+except ImportError:
+    review_events = None
+
+EVENTS_URL = "/prompt_manager/review/events"
+# A safety bound for one read; every wait below ends on an event first.
+READ_SECONDS = 10
+
+
+class ReviewEventsTestCase(ReviewRoutesTestCase):
+    def _require_events(self):
+        self.assertIsNotNone(review_changes, "database.review_changes is missing")
+        self.assertIsNotNone(review_events, "py.api.review_events is missing")
+
+    async def _stream(self):
+        resp = await self.client.get(EVENTS_URL)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Content-Type"], "text/event-stream")
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        self.addAsyncCleanup(self._close, resp)
+        self.assertEqual(await self._event(resp), "ready")
+        return resp
+
+    @staticmethod
+    async def _close(resp):
+        resp.close()
+
+    async def _event(self, resp):
+        """The next event's name (keepalive comments are skipped)."""
+        name = None
+        while True:
+            line = await asyncio.wait_for(resp.content.readline(), READ_SECONDS)
+            self.assertTrue(line, "the stream ended")
+            text = line.decode("utf-8").rstrip("\n")
+            if text.startswith("event: "):
+                name = text[len("event: ") :]
+            elif text == "" and name is not None:
+                return name
+
+    async def _record_in_thread(self, path):
+        """Record as the monitor does: from a worker thread, not the loop."""
+        self._roots()
+        outcomes = []
+        thread = threading.Thread(
+            target=lambda: outcomes.append(self.api.record_review_file(str(path)))
+        )
+        thread.start()
+        await asyncio.get_running_loop().run_in_executor(None, thread.join)
+        return outcomes[0]
+
+
+class TestReviewEvents(ReviewEventsTestCase):
+    async def test_monitor_recording_sends_one_change(self):
+        self._require_events()
+        stream = await self._stream()
+        self.assertEqual(await self._record_in_thread(self._ambiguous()), "recorded")
+        self.assertEqual(await self._event(stream), "change")
+
+    async def test_duplicate_recording_and_refused_decision_send_nothing(self):
+        self._require_events()
+        path = self._ambiguous("two.png")
+        self._record(path)
+        stream = await self._stream()
+        # The same bytes again: nothing changes.
+        self.assertEqual(await self._record_in_thread(path), "duplicate")
+        item, detail, observed = await self._open()
+        # A stale revision: refused with zero writes.
+        body = {**self._body(detail, observed, "unresolved"), "expected_revision": 9}
+        status, data = await self._decide(item["id"], body)
+        self.assertEqual(status, 409, data)
+        # The next notice is the next real change, so nothing came before it.
+        self._record(self._ambiguous("other.png", color=(0, 0, 9)))
+        self.assertEqual(await self._event(stream), "change")
+        self.assertEqual(len(self._items()), 2)
+
+    async def test_decision_from_any_tab_sends_a_change(self):
+        self._require_events()
+        self._record(self._ambiguous())
+        stream = await self._stream()
+        item, detail, observed = await self._open()
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, "unresolved")
+        )
+        self.assertEqual(status, 200, data)
+        self.assertEqual(await self._event(stream), "change")
+
+    async def test_restore_sends_a_change(self):
+        self._require_events()
+        stream = await self._stream()
+        from database.execution_history import notify_history_restore
+
+        notify_history_restore(self.api.db.model.db_path)
+        self.assertEqual(await self._event(stream), "change")
+
+    async def test_scan_recording_sends_a_change(self):
+        self._require_events()
+        self._roots()
+        self._ambiguous("sub/two.png")
+        stream = await self._stream()
+        job = await self._scan("all")
+        self.assertEqual(job["scope"]["review_items"], 1)
+        self.assertEqual(await self._event(stream), "change")
+
+    async def test_closing_the_stream_unsubscribes(self):
+        self._require_events()
+        path = self.api.db.model.db_path
+        stream = await self._stream()
+        self.assertEqual(review_changes.observer_count(path), 1)
+        self.assertEqual(self.api.review_stream_count(), 1)
+        stream.close()
+        # The server notices the closed socket at its next write: a change.
+        self._record(self._ambiguous())
+        for _ in range(100):
+            if not self.api.review_stream_count():
+                break
+            await asyncio.sleep(0.02)
+        self.assertEqual(self.api.review_stream_count(), 0)
+        self.assertEqual(review_changes.observer_count(path), 0)
+
+    async def test_shutdown_ends_every_stream(self):
+        self._require_events()
+        stream = await self._stream()
+        await self.api.close_review_events(None)
+        line = await asyncio.wait_for(stream.content.read(), READ_SECONDS)
+        self.assertEqual(line, b"")
+        self.assertEqual(self.api.review_stream_count(), 0)
+        # A stream opened after shutdown ends at once.
+        resp = await self.client.get(EVENTS_URL)
+        self.assertEqual(resp.status, 503)
+        resp.close()
+
+    async def test_keepalive_comment_is_skipped_by_clients(self):
+        self._require_events()
+        with self.patch_heartbeat(0.05):
+            stream = await self._stream()
+            line = await asyncio.wait_for(stream.content.readline(), READ_SECONDS)
+            self.assertEqual(line, b": keepalive\n")
+
+    def patch_heartbeat(self, seconds):
+        from unittest.mock import patch
+
+        return patch.object(review_events, "HEARTBEAT_SECONDS", seconds)
+
+
+class TestReviewChangeRegistry(ReviewEventsTestCase):
+    def test_a_failing_observer_never_fails_a_write(self):
+        self._require_events()
+        path = self.api.db.model.db_path
+        seen = []
+
+        def broken():
+            raise RuntimeError("observer failed")
+
+        unsubscribe_broken = review_changes.subscribe_review_changes(path, broken)
+        unsubscribe = review_changes.subscribe_review_changes(
+            path, lambda: seen.append(1)
+        )
+        try:
+            with self.assertLogs(
+                "prompt_manager.database.review_changes", level="WARNING"
+            ):
+                self._record(self._ambiguous())
+            self.assertEqual(seen, [1])
+        finally:
+            unsubscribe_broken()
+            unsubscribe()
+        self.assertEqual(review_changes.observer_count(path), 0)
+
+    def test_notice_comes_after_the_commit(self):
+        self._require_events()
+        path = self.api.db.model.db_path
+        seen = []
+
+        def committed():
+            # A separate connection sees only committed rows.
+            other = sqlite3.connect(path)
+            try:
+                seen.append(
+                    other.execute("SELECT COUNT(*) FROM media_review_items").fetchone()[
+                        0
+                    ]
+                )
+            finally:
+                other.close()
+
+        unsubscribe = review_changes.subscribe_review_changes(path, committed)
+        try:
+            self._record(self._ambiguous())
+        finally:
+            unsubscribe()
+        self.assertEqual(seen, [1])

web/js/workspace/index.mjs

@@ -251,8 +251,9 @@ function mountWorkspace() {
             return;
         }
         if (route.id === 'review') {
-            // Needs review (ruling R5): a recorded decision re-reads the row's count.
-            mountReview(main, { api, scope: view, onDecision: () => refreshCounts(REVIEW_WRITE_COUNTS) });
+            // Needs review (ruling R5): a recorded decision, and a change the open page hears of, re-read the row's count.
+            mountReview(main, { api, scope: view, onDecision: () => refreshCounts(REVIEW_WRITE_COUNTS),
+                onQueueChange: () => refreshCounts(REVIEW_WRITE_COUNTS) });
             return;
         }
         if (route.id === 'inspect') {

web/js/workspace/review-state.mjs

@@ -11,6 +11,10 @@
  *   a restored storage or a changed queue reads the queue again. A request that got no answer keeps its capture and
  *   its exact body, and the next Apply of the same choice (`retry()`) resends that body once.
  * - Leave for later writes nothing. Tagging the image (the shared Tag dialog) never changes the item.
+ * - `refresh()` re-reads the first page quietly when the server says the queue changed (F2: the review change stream:
+ *   a scan or the monitor added items, another tab decided one). The open item, its choice and an unanswered decision
+ *   stay; when the open item has left the queue the next one opens with a note. Refreshes asked for while one runs
+ *   collapse into one more read.
  */
 import { noun } from './plural.mjs';
 
@@ -28,7 +32,7 @@ export const COPY = Object.freeze({
     empty: 'Ambiguous images are added when a metadata scan or a new generation finds more than one sampler path.',
     panelTitle: 'Resolve the association',
     leaveUnresolved: 'Leave unresolved',
-    // The states the concept lacks are text only (Global Constraints); D-NEW-F1.
+    // The states the concept lacks are text only (Global Constraints); D119.
     loading: 'Loading the review queue…',
     loadFailed: 'The review queue could not be read.',
     detailLoading: 'Loading sampler paths…',
@@ -37,6 +41,9 @@ export const COPY = Object.freeze({
     limited: 'This file has too many sampler paths to list. You can leave it unresolved.',
     unavailable: 'This file is not under a current image source, so it cannot be opened. You can leave it unresolved.',
     lost: 'The decision may not have been saved. Apply again to send the same decision once more.',
+    // D119 (F2): the open item left the queue in a re-read, and a failed re-read.
+    decidedElsewhere: name => `${name} was decided elsewhere, so the next image is open.`,
+    refreshFailed: 'The review queue could not be read again. Reload the page to see the latest images.',
     failed: 'The decision could not be saved. Try again.',
     applied: 'Association accepted.',
     leftUnresolved: 'Left unresolved.',
@@ -94,7 +101,7 @@ const initial = () => ({
 
 /**
  * @param {{api: {get: Function, post: Function, del: Function}}} options
- * @returns the controller: `state`, `subscribe(listener)`, `load()`, `loadMore()`, `select(id)`, `choose(value)`,
+ * @returns the controller: `state`, `subscribe(listener)`, `load()`, `refresh()`, `loadMore()`, `select(id)`, `choose(value)`,
  *   `apply()`, `retry()`, `leave()`, `dispose()`.
  */
 export function createReviewController({ api }) {
@@ -102,6 +109,7 @@ export function createReviewController({ api }) {
     const listeners = new Set();
     const lifetime = new AbortController();
     let listRead = null, detailRead = null, pending = null, disposed = false;
+    let refreshing = null, refreshAgain = false, refreshAfterLoad = false;
 
     function set(update) {
         if (disposed) return;
@@ -119,27 +127,67 @@ export function createReviewController({ api }) {
         pending = null;
     }
 
-    /** Read the first page; `stale` re-reads the selected item's detail too (its revision or epoch moved). */
-    async function load({ keepWarning = false, stale = false } = {}) {
+    /**
+     * Read the first page; `stale` re-reads the selected item's detail too (its revision or epoch moved). `quiet` (a
+     * refresh) keeps the page as it is while reading and after a failed read.
+     */
+    async function load({ keepWarning = false, stale = false, quiet = false } = {}) {
         listRead?.abort();
         const read = listRead = new AbortController();
-        set({ status: state.items.length ? state.status : 'loading', ...(keepWarning ? {} : { warning: '' }) });
+        if (!quiet) set({ status: state.items.length ? state.status : 'loading', ...(keepWarning ? {} : { warning: '' }) });
+        let readOk = false;
         try {
             const page = await api.get(`${REVIEW_URL}?limit=${PAGE_LIMIT}`, { signal: signalOf(read), cache: 'no-store' });
             if (read !== listRead || disposed) return;
             const items = Object.freeze((page.items ?? []).map(entry => Object.freeze({ ...entry })));
+            const left = state.items.find(entry => entry.id === state.selectedId);
             set({ status: 'ready', epoch: page.library_epoch, total: page.total, items, nextCursor: page.next_cursor ?? null,
                 more: 'idle' });
+            readOk = true;
             const keep = items.find(entry => entry.id === state.selectedId);
-            if (keep && state.detail && !stale) return;
-            if (items.length) await select((keep ?? items[0]).id, { keepWarning });
+            // A re-read keeps the open item as it is (even a failed read of it, which has its own Retry).
+            if (keep && !stale && (quiet || state.detail)) return;
+            if (quiet && left && !keep && items.length) {
+                set({ warning: COPY.decidedElsewhere(left.filename) });
+                await select(items[0].id, { keepWarning: true });
+            } else if (items.length) await select((keep ?? items[0]).id, { keepWarning });
             else set({ selectedId: null, detail: null, detailStatus: 'idle', options: Object.freeze([]), choice: null, prompts: null });
         } catch (error) {
             if (read !== listRead || disposed || isAbort(error)) return;
-            set({ status: 'error' });
+            if (quiet && state.status === 'ready') set({ warning: COPY.refreshFailed });
+            else set({ status: 'error' });
+        } finally {
+            // A change heard while the first read ran: that read may predate it, so read once more.
+            if (readOk && refreshAfterLoad && !disposed) {
+                refreshAfterLoad = false;
+                refresh();
+            }
         }
     }
 
+    /** Re-read the queue after a change elsewhere; calls made while one runs collapse into one more read. */
+    function refresh() {
+        if (disposed) return Promise.resolve();
+        // Before the queue has been read there is nothing to keep: the first read (or Retry) reads the latest.
+        if (state.status !== 'ready') {
+            if (state.status === 'loading') refreshAfterLoad = true;
+            return Promise.resolve();
+        }
+        if (refreshing) {
+            refreshAgain = true;
+            return refreshing;
+        }
+        refreshing = (async () => {
+            do {
+                refreshAgain = false;
+                // A decision in flight finishes first: the page leaves when it is recorded.
+                if (state.phase === 'applying') return;
+                await load({ keepWarning: true, quiet: true });
+            } while (refreshAgain && !disposed);
+        })().finally(() => { refreshing = null; });
+        return refreshing;
+    }
+
     async function loadMore() {
         if (!state.nextCursor || state.more === 'loading') return;
         const read = listRead = new AbortController();
@@ -200,7 +248,9 @@ export function createReviewController({ api }) {
                 toast: choice === UNRESOLVED ? COPY.leftUnresolved : COPY.applied });
             return true;
         } catch (error) {
-            if (disposed || isAbort(error)) return false;
+            // Leaving the page mid-request: the server may never see it, so the capture is released here (a release
+            // of a capture the server already used is a harmless 404).
+            if (disposed || isAbort(error)) { release(body.capture_token); return false; }
             if (isLost(error)) {
                 pending = { id, body };
                 set({ phase: 'idle', warning: COPY.lost, canRetry: true });
@@ -261,7 +311,7 @@ export function createReviewController({ api }) {
             listeners.add(listener);
             return () => listeners.delete(listener);
         },
-        load: () => load(), loadMore, select: id => select(id), choose, apply, retry, leave,
+        load: () => load(), refresh, loadMore, select: id => select(id), choose, apply, retry, leave,
         dispose() {
             if (disposed) return;
             dropPending();

web/js/workspace/review.mjs

@@ -10,8 +10,11 @@
  *   association": filename, Tag image, Sampler context, the chosen path's prompt blocks and the actions Leave for later
  *   and Apply association. Both return to Images (ruling R7).
  * - States the concept lacks (loading, a failed read, a refused or unanswered decision, a limited or unavailable file)
- *   are text: `p.warning-note` in the panel, or the page's status line (D-NEW-F1).
- * - D-NEW-F1: the queue cards carry no selection checkbox and no star (this page has no selection bar, and the queue
+ *   are text: `p.warning-note` in the panel, or the page's status line (D119).
+ * - The open page re-reads its queue on each `change` of the review change stream (`/prompt_manager/review/events`,
+ *   F2): items a scan or the monitor added, and decisions made in other tabs. Each `ready` (the first one, and one
+ *   after the browser reconnects a dropped stream) re-reads too, since a change may have come before it. Nothing polls.
+ * - D119: the queue cards carry no selection checkbox and no star (this page has no selection bar, and the queue
  *   carries no favorite state); more than 40 items add the Images page's `.infinite-continuation` Load more after the
  *   layout; the toasts "Association accepted." and "Left unresolved." stand in for the concept's prototype toast.
  */
@@ -30,6 +33,9 @@ import { navigateTo } from './state.mjs';
 import { showWorkspaceToast } from './toast.mjs';
 import { imageKey } from './images-state.mjs';
 
+export const REVIEW_EVENTS_URL = '/prompt_manager/review/events';
+const openEvents = url => new EventSource(window.ApiClient.resolveUrl(url));
+
 function el(tag, text, className) {
     const node = document.createElement(tag);
     if (text !== undefined) node.textContent = text;
@@ -53,10 +59,12 @@ export function imageOf(item) {
 }
 
 /**
- * Mount the page under the shell's heading. `onDecision()` runs after a recorded decision (the shell re-reads the
- * Needs review count); `controller` is injectable for tests.
+ * Mount the page under the shell's heading. `onDecision()` runs after a recorded decision and `onQueueChange()` after
+ * a change elsewhere (the shell re-reads the Needs review count); `controller` and `connect` (the change stream's
+ * EventSource) are injectable for tests.
  */
-export function mountReview(main, { api, scope, onDecision = () => {}, controller = createReviewController({ api }) }) {
+export function mountReview(main, { api, scope, onDecision = () => {}, onQueueChange = () => {},
+    connect = openEvents, controller = createReviewController({ api }) }) {
     scope.cleanup(() => controller.dispose());
     const heading = pageHeading(main);
     heading.h1.textContent = COPY.title;
@@ -90,7 +98,7 @@ export function mountReview(main, { api, scope, onDecision = () => {}, controlle
     tagRow.append(tagButton);
     const select = el('select');
     const contextField = field('Sampler context', select, 'review-context');
-    const warning = el('p', '', 'warning-note'); // D-NEW-F1: the states the concept lacks
+    const warning = el('p', '', 'warning-note'); // D119: the states the concept lacks
     warning.setAttribute('role', 'status');
     const blocks = el('div');
     const actions = el('div', undefined, 'dialog-actions');
@@ -100,7 +108,7 @@ export function mountReview(main, { api, scope, onDecision = () => {}, controlle
     panel.append(panelTitle, filename, tagRow, contextField, warning, blocks, actions);
     layout.append(grid, panel);
 
-    const continuation = el('div', undefined, 'infinite-continuation'); // D-NEW-F1: Load more past 40 items
+    const continuation = el('div', undefined, 'infinite-continuation'); // D119: Load more past 40 items
     const moreText = el('p', '', 'small');
     const more = createButton('Load more images');
     continuation.append(moreText, more);
@@ -124,7 +132,7 @@ export function mountReview(main, { api, scope, onDecision = () => {}, controlle
             state.items.forEach((item, index) => {
                 let card = cards.get(item.id);
                 if (!card) {
-                    // D-NEW-F1: no selection checkbox and no star on a queue card.
+                    // D119: no selection checkbox and no star on a queue card.
                     card = createImageCard(imageOf(item), { media: Boolean(item.url) });
                     card.dataset.reviewId = item.id;
                     card.item = item;
@@ -243,5 +251,14 @@ export function mountReview(main, { api, scope, onDecision = () => {}, controlle
     scope.cleanup(controller.subscribe(render));
     render(controller.state);
     controller.load();
+    const events = connect(REVIEW_EVENTS_URL);
+    scope.cleanup(() => events.close());
+    const changed = () => {
+        if (scope.signal.aborted) return;
+        controller.refresh();
+        onQueueChange();
+    };
+    events.addEventListener('ready', changed);
+    events.addEventListener('change', changed);
     return { controller };
 }

web/js/workspace/workspace.css

@@ -383,6 +383,6 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
 /* #249 D20: the prompt blocks' mount point is transparent, so the chosen path's sections sit in the panel as the
    concept's do. */
 .review-panel > .prompt-inspector { display: contents; /* D20 */ }
-/* D-NEW-F1: a file that cannot be served (no current root) has an inert, empty media button. */
+/* D119: a file that cannot be served (no current root) has an inert, empty media button. */
 .review-queue .image-open[aria-disabled="true"] { cursor: not-allowed; }
 .review-panel > p.small { min-inline-size: 0; overflow-wrap: anywhere; }