diff --git a/TESTING.md b/TESTING.md index 028f81d..2fe3757 100644 --- a/TESTING.md +++ b/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 " 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). diff --git a/__init__.py b/__init__.py index ad4e504..78da47a 100644 --- a/__init__.py +++ b/__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 diff --git a/database/review.py b/database/review.py index 46b6b31..b1269aa 100644 --- a/database/review.py +++ b/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).""" diff --git a/database/review_changes.py b/database/review_changes.py new file mode 100644 index 0000000..31c54b6 --- /dev/null +++ b/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) diff --git a/database/review_decisions.py b/database/review_decisions.py index 3b5b1ff..a4ad767 100644 --- a/database/review_decisions.py +++ b/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"]) diff --git a/py/api/__init__.py b/py/api/__init__.py index 3d92bbc..d4803bb 100644 --- a/py/api/__init__.py +++ b/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) diff --git a/py/api/review_events.py b/py/api/review_events.py new file mode 100644 index 0000000..61d2e0f --- /dev/null +++ b/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 diff --git a/tests/js/workspace-review-state.test.js b/tests/js/workspace-review-state.test.js index 0139af7..7ee0236 100644 --- a/tests/js/workspace-review-state.test.js +++ b/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'); +}); diff --git a/tests/test_api_review_events.py b/tests/test_api_review_events.py new file mode 100644 index 0000000..08e6d3a --- /dev/null +++ b/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]) diff --git a/web/js/workspace/index.mjs b/web/js/workspace/index.mjs index 06d0dc2..5b00388 100644 --- a/web/js/workspace/index.mjs +++ b/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') { diff --git a/web/js/workspace/review-state.mjs b/web/js/workspace/review-state.mjs index c33c40d..7f9fbc4 100644 --- a/web/js/workspace/review-state.mjs +++ b/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(); diff --git a/web/js/workspace/review.mjs b/web/js/workspace/review.mjs index 0dfe44c..13a2a67 100644 --- a/web/js/workspace/review.mjs +++ b/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 }; } diff --git a/web/js/workspace/workspace.css b/web/js/workspace/workspace.css index ec2f1a5..7367d69 100644 --- a/web/js/workspace/workspace.css +++ b/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; }