nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(review): keep the live queue right through applies, Load more and lost streams

Browse Source 
- A change heard while a decision is in flight is owed and read once the
  decision is refused or unanswered (I1).
- Load more has its own read: during a re-read it waits and continues from
  the re-read queue; a re-read started during Load more takes the click
  over, and a failed re-read leaves Load more clickable (I2, M7).
- A stream the browser gave up on (503, proxy error) reopens after 3 s,
  doubling to 30 s, with a quiet "Live updates are paused" line while it is
  down (I3, D-NEW-F2).
- A hidden tab closes its stream and a visible one reopens it; the ready
  version keeps that to at most one read (I4).
- Apply fences on the detail's epoch and a re-read under another epoch reads
  the open item again (M3); leaving during the observe releases its capture
  (M4); an unanswered decision that landed finishes as recorded (M5); cards
  follow a re-read's url, target and availability and show thumbnails
  (M6, M10).
- The unreadable-ready test asserts exactly one re-read (M1); the panel's
  long-file-name wrap carries its D119 marker and TESTING.md entry (M8).
This commit is contained in:
Vito Sansevero
2026-10-22 22:46:11 -07:00
parent afe3733b69
commit 807cb32f58
7 changed files with 498 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+41 -11
View File
@@ -1635,7 +1635,18 @@ pending count (`GET /prompt_manager/review/items?limit=1`, its `total`).
is the file name; the panel's Copy and Send buttons carry no prompt title in their
names; the Tag image dialog opened from the queue heads with the file name. An
unavailable file's empty media button keeps the concept's 1.33 media area, so its
card lines up with the others.
card lines up with the others. A long file name in the panel's `p.small` wraps inside
the panel (`overflow-wrap: anywhere`), where the concept's short sample name never
overflows. A queue card shows the Images listing's thumbnail when the server knows one
(`thumbnail_url`, the same URL the Images page uses), else the original; a re-read
rebuilds a card whose media, target or availability changed (a root came or went), so
Open and Tag follow it. When an unanswered decision's item leaves the queue in a
re-read, the decision was recorded: the page finishes as a recorded decision does (its
toast, back to Images), not with "decided elsewhere".
- **D-NEW-F2** (placeholder number; the concept has no live-update state): while the
change stream is down, the heading carries the quiet status line "Live updates are
paused: changes made elsewhere appear once the connection returns." (`p.small
.review-live`, `role="status"`); the stream's next `ready` hides it.
**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
@@ -1651,25 +1662,44 @@ cannot be read, re-reads); a `ready` heard while the first read runs is compared
read answers. The test server counts a stream as open only once it has sent `ready`, so a
browser test that changes the queue after that count is a `change`, never part of the
version. The page re-reads on every `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. **Every re-read keeps the loaded depth (F3):** it
unanswered decision; re-reads asked for while one runs collapse into one more. A change
heard while a decision is in flight is owed: a refused or unanswered decision reads it
once the panel is idle again (a recorded one leaves the page). Load more asked for during
a re-read waits for it and continues after the queue it read; a re-read started during
Load more takes that click over and reads one page deeper, and a failed re-read leaves
Load more clickable. Apply fences on the epoch of the detail it decides: an observe under
another storage releases its capture and reads the queue and the item again, and any
re-read under another epoch reads the open item again. **Every re-read keeps the loaded depth (F3):** it
reads page after page (40 each, the route's limit) until it holds as many items as Load
more had shown, so the cards stay and the scroll position does not move; a restore met
after the first page reads once more from the first page. The row's count follows the
total the page shows (no count read per change). 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; a re-read of a deep queue costs
one request per 40 items shown.
every 25 s only lets the server drop a closed tab's subscription. The server sends the
first change at once and collapses the notices of the next 0.4 s into one trailing
`change` (`COALESCE_SECONDS`), so a busy scan costs an open page one re-read per 0.4 s at
most. Shutdown (`on_shutdown`) ends every stream and answers a later one 503. A
reconnect the browser gives up on (a 503, or a proxy's error while ComfyUI restarts) is
opened again by the page after 3 s, doubling to 30 s, with D-NEW-F2's line shown while
the stream is down; its `ready` re-syncs the queue. The stream lives only while the tab is
visible: a hidden tab closes it and a visible one opens it again, and that `ready` reads
only when the queue changed meanwhile. A duplicate recording and a refused decision change
nothing and send nothing. Residual: each visible Needs review page holds one HTTP/1.1
connection (browsers allow six per origin, shared with ComfyUI's editor), so several
side-by-side visible windows on the page, during a metadata scan's job stream, can still
queue the origin's other requests; a re-read of a deep queue costs one request per 40
items shown.
Pinned by `tests/test_api_review_events.py`, `tests/js/workspace-review-state.test.js`
("refresh …", "a refresh keeps the loaded depth …", "ready naming …") and
`tests/browser/workspace-review.spec.mjs` ("an open page shows …", "a visit reads its
queue once …", "a live re-read keeps the loaded depth and the scroll position").
queue once …", "a live re-read keeps the loaded depth and the scroll position", "a stream
the browser gives up on …", "a hidden tab closes its stream …", "a re-read refreshes a
card …"); the controller's owed, Load more and epoch cases are "I1: …", "I2: …", "M3:
…", "M5: …" and "M7: …" in `workspace-review-state.test.js`, and the coalescing is "a
burst of changes …" in `test_api_review_events.py`.
Leaving the page mid-decision releases that decision's capture (the server may never see
the request).
the request), and leaving during the observe releases the capture it then answers
("leaving during the observe …").
**Concept comparison (F3).** `tests/browser/review-concept.spec.mjs` measures the whole
page (`reviewInventory` in `tests/browser/tagging-canonical-comparison.mjs`: every shown
box below `main` with its place, fonts, edges and layout, a leaf's text, a button's name,
tests/browser/workspace-review.spec.mjs
+106 -2
View File
@@ -393,11 +393,115 @@ test('a ready whose version cannot be read re-reads the queue once', async ({ pa
// The stream's ready names no readable version (an older server, a proxy that rewrote it): the page reads again.
await seed(request, 1);
const reads = reviewReads(page);
// M1: `retry` keeps the browser from reconnecting the ended stream during the test, so exactly one re-read follows.
await page.route(url => url.pathname === '/prompt_manager/review/events', route => route.fulfill({
status: 200, headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-store' }, body: 'event: ready\ndata: not json\n\n' }));
status: 200, headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-store' },
body: 'retry: 3600000\nevent: ready\ndata: not json\n\n' }));
let reread = 0;
const second = page.waitForResponse(item => item.url().endsWith('/prompt_manager/review/items?limit=40') && ++reread === 2);
await open(page);
await expect.poll(() => reads.filter(search => search === '?limit=40').length).toBeGreaterThanOrEqual(2);
await second;
await expect(cards(page)).toHaveCount(1);
expect(reads.filter(search => search === '?limit=40')).toHaveLength(2);
});
const live = page => main(page).locator('.heading .review-live');
const NOT_LIVE = 'Live updates are paused: changes made elsewhere appear once the connection returns.';
test('a stream the browser gives up on is opened again, and the page says it is not live meanwhile', async ({ page, request }) => {
// I3: a 503 (shutdown, or a proxy while the backend restarts) fails an EventSource for good.
await seed(request, 1);
let attempts = 0;
await page.route(url => url.pathname === '/prompt_manager/review/events', route => {
attempts += 1;
if (attempts === 1) return route.fulfill({ status: 503, contentType: 'application/json', body: '{"success":false}' });
return route.fallback();
});
await open(page);
await expect(live(page)).toHaveText(NOT_LIVE);
await expect(live(page)).toBeVisible();
// The page opens the stream again (after 3 s) and is live once it sends ready.
await expect.poll(async () => (await streams(request)).streams, { timeout: 15000 }).toBe(1);
await expect(live(page)).toBeHidden();
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(cards(page)).toHaveCount(2);
expect(attempts).toBe(2);
});
/** Make the page hidden or visible as a tab switch does. */
async function setVisible(page, visible) {
await page.evaluate(shown => {
Object.defineProperty(document, 'hidden', { configurable: true, get: () => !shown });
Object.defineProperty(document, 'visibilityState', { configurable: true, get: () => (shown ? 'visible' : 'hidden') });
document.dispatchEvent(new Event('visibilitychange'));
}, visible);
}
test('a hidden tab closes its stream and a visible one reads only what changed meanwhile', async ({ page, request }) => {
// I4: background tabs must not hold one of the origin's six connections.
await seed(request, 1);
const reads = reviewReads(page);
await open(page);
await expect.poll(async () => (await streams(request)).streams).toBe(1);
await setVisible(page, false);
// The server drops a closed stream at its next write: this monitor addition is one.
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect.poll(async () => (await streams(request)).streams).toBe(0);
await expect(cards(page)).toHaveCount(1);
expect(reads.filter(search => search === '?limit=40')).toHaveLength(1);
// Visible again: the reopened stream's ready names a newer version, so the page reads once.
const reread = page.waitForResponse(item => item.url().endsWith('/prompt_manager/review/items?limit=40'));
await setVisible(page, true);
await expect.poll(async () => (await streams(request)).streams).toBe(1);
await reread;
await expect(cards(page)).toHaveCount(2);
await expect(live(page)).toBeHidden();
expect(reads.filter(search => search === '?limit=40')).toHaveLength(2);
});
test('a re-read refreshes a card whose media or target changed', async ({ page, request }) => {
// M6 and M10: a root came or went between reads; the card follows, and a known thumbnail is its media.
await seed(request, 2);
let rewrite = false;
await page.route(url => url.pathname === '/prompt_manager/review/items' && url.search === '?limit=40', async route => {
if (!rewrite) return route.fallback();
const response = await route.fetch();
const body = await response.json();
body.items[0] = { ...body.items[0], thumbnail_url: `${body.items[0].url}&thumb=1` };
body.items[1] = { ...body.items[1], url: null, thumbnail_url: null, target: null, unavailable: true };
return route.fulfill({ response, json: body });
});
await open(page);
await expect.poll(async () => (await streams(request)).streams).toBe(1);
const second = cards(page).nth(1);
await expect(second.getByRole('button', { name: 'Open review_0001.png', exact: true })).not.toHaveAttribute('aria-disabled', 'true');
rewrite = true;
await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK();
await expect(cards(page)).toHaveCount(3);
await expect(cards(page).first().locator('img')).toHaveAttribute('src', /&thumb=1$/);
await expect(second.getByRole('button', { name: 'Open review_0001.png', exact: true })).toHaveAttribute('aria-disabled', 'true');
await expect(second.locator('img')).toHaveCount(0);
await expect(second.locator('.image-card-actions button').last()).toHaveAttribute('aria-disabled', 'true');
await expect(cards(page).first()).toHaveClass(/selected/);
});
test('leaving during the observe releases the capture it answers', async ({ page, request }) => {
// M4: the server made the capture; the page left before the answer came.
await seed(request, 1);
await open(page);
let held;
const sent = new Promise(resolve => { held = resolve; });
await page.route(url => url.pathname === '/prompt_manager/image-annotations/observe', async route => {
const response = await route.fetch();
held(() => route.fulfill({ response }));
});
await apply(page).click();
const answer = await sent;
expect((await streams(request)).captures).toBe(1);
await page.evaluate(() => { window.location.hash = '#/tags'; });
await expect(main(page).getByRole('heading', { level: 1 })).toHaveText('Tags & categories');
await answer();
await expect.poll(async () => (await streams(request)).captures).toBe(0);
});
test('a live re-read keeps the loaded depth and the scroll position', async ({ page, request }) => {
tests/js/workspace-review-state.test.js
+180
View File
@@ -606,3 +606,183 @@ test('a change heard during the first read still re-reads after a current ready'
await settle();
assert.equal(lists, 2);
});
// Final review round (I1, I2, M3–M5, M7).
test('I1: a change heard during a refused Apply is re-read once the phase is idle again', async () => {
let queue = [item('one')], releaseDecision;
const { c, h } = await loaded({
[`GET ${ITEMS}`]: () => response(200, page(queue, { review_revision: queue.length })),
[`POST ${ITEMS}/one/decision`]: () => new Promise(resolve => {
releaseDecision = () => resolve(fail(409, 'association-conflict', 'conflict'));
}),
});
const applying = c.apply();
await settle();
assert.equal(c.state.phase, 'applying');
queue = [item('one'), item('two')]; // the monitor records an item while the decision is in flight
c.refresh();
await settle();
assert.equal(h.of('GET', `${ITEMS}?`).length, 1, 'no read while the decision is in flight');
releaseDecision();
await applying;
await settle();
assert.equal(c.state.phase, 'idle');
assert.deepEqual(c.state.items.map(entry => entry.id), ['one', 'two']);
assert.equal(c.state.total, 2);
assert.equal(c.state.warning, 'conflict', 'the refusal stays the warning');
assert.equal(h.of('GET', `${ITEMS}?`).length, 2, 'exactly one owed re-read');
});
test('I1: a change heard during an Apply whose observe is refused is re-read too', async () => {
let queue = [item('one')], releaseObserve;
const { c } = await loaded({
[`GET ${ITEMS}`]: () => response(200, page(queue)),
[`POST ${OBSERVE}`]: () => new Promise(resolve => {
releaseObserve = () => resolve(fail(409, 'capture-capacity', 'busy'));
}),
});
const applying = c.apply();
await settle();
queue = [item('one'), item('two')];
c.refresh();
releaseObserve();
await applying;
await settle();
assert.deepEqual(c.state.items.map(entry => entry.id), ['one', 'two']);
});
const forty = Array.from({ length: 41 }, (_, i) => item(`i${String(i).padStart(2, '0')}`));
function deepRoutes(state) {
return {
[`GET ${ITEMS}`]: call => {
const cursor = new URL(call.url, 'http://x').searchParams.get('cursor');
if (cursor) {
if (state.moreGate) return state.moreGate.then(() => state.more ?? response(200, page([forty[40]], { total: 41 })));
return state.more ?? response(200, page([forty[40]], { total: 41 }));
}
const body = response(200, page(state.first, { total: state.first.length + 1, review_revision: state.revision ?? 1, next_cursor: 'cur' }));
return state.gate ? state.gate.then(() => state.fail ?? body) : body;
},
[`GET ${ITEMS}/i00`]: () => response(200, detailOf(forty[0])),
};
}
test('I2: Load more during a re-read waits for it and continues from the re-read queue', async () => {
const state = { first: forty.slice(0, 40) };
const { c } = await loaded(deepRoutes(state));
// Another tab decides i05; the re-read is slow and the user clicks Load more meanwhile.
let open;
state.first = forty.slice(0, 40).filter(entry => entry.id !== 'i05');
state.gate = new Promise(resolve => { open = resolve; });
const refreshing = c.refresh();
await settle();
const more = c.loadMore();
await settle();
assert.equal(c.state.more, 'loading', 'the click shows as loading while the re-read runs');
open();
await Promise.all([refreshing, more]);
await settle();
const ids = c.state.items.map(entry => entry.id);
assert.ok(!ids.includes('i05'), 'the re-read was applied');
assert.equal(ids.at(-1), 'i40', 'Load more appended the next page after the re-read');
assert.equal(c.state.more, 'idle');
});
test('I2: a re-read started during Load more takes over the click and reads one page deeper', async () => {
const state = { first: forty.slice(0, 40) };
const { c, h } = await loaded(deepRoutes(state));
let openMore;
state.moreGate = new Promise(resolve => { openMore = resolve; });
const more = c.loadMore();
await settle();
state.moreGate = null;
state.first = forty.slice(0, 40).filter(entry => entry.id !== 'i05');
await c.refresh();
openMore();
await more;
await settle();
const ids = c.state.items.map(entry => entry.id);
assert.ok(!ids.includes('i05'));
assert.equal(ids.at(-1), 'i40', 'the aborted Load more page is part of the re-read');
assert.equal(c.state.more, 'idle');
assert.equal(h.of('GET', `${ITEMS}?`).length, 4, 'load, the aborted more, and the re-read of two pages');
});
test('M7: a failed re-read that took over Load more clears its loading state', async () => {
const state = { first: forty.slice(0, 40) };
const { c } = await loaded(deepRoutes(state));
state.moreGate = new Promise(() => {});
c.loadMore();
await settle();
assert.equal(c.state.more, 'loading');
state.moreGate = null;
state.gate = Promise.resolve();
state.fail = fail(500, 'review-failed');
await c.refresh();
await settle();
assert.equal(c.state.more, 'idle', 'Load more can be clicked again');
assert.equal(c.state.warning, COPY_REFRESH_FAILED);
});
const COPY_REFRESH_FAILED = 'The review queue could not be read again. Reload the page to see the latest images.';
test('M3: Apply fences on the epoch of the detail it decides', async () => {
const OTHER = '22222222-2222-4222-8222-222222222222';
const { c, h } = await loaded({ [`POST ${OBSERVE}`]: () => response(200, { ...observed(), library_epoch: OTHER }) });
assert.equal(await c.apply(), false);
await settle();
assert.equal(h.of('POST', `${ITEMS}/one/decision`).length, 0, 'a restored storage sends no decision');
assert.deepEqual(h.of('DELETE', CAPTURE).map(call => call.url), [CAPTURE + 'capture-1']);
assert.equal(h.of('GET', `${ITEMS}?`).length, 2, 'the queue is read again');
assert.equal(h.of('GET', `${ITEMS}/one`).length, 2, 'the open item is read again under the new storage');
assert.equal(c.state.phase, 'idle');
});
test('M3: the decision carries the detail epoch', async () => {
const { c, h } = await loaded();
assert.equal(await c.apply(), true);
assert.equal(h.of('POST', `${ITEMS}/one/decision`)[0].body.expected_epoch, EPOCH);
});
test('M3: a quiet re-read under another epoch reads the open item again', async () => {
const OTHER = '22222222-2222-4222-8222-222222222222';
let epoch = EPOCH;
const { c, h } = await loaded({
[`GET ${ITEMS}`]: () => response(200, page(QUEUE, { library_epoch: epoch })),
[`GET ${ITEMS}/one`]: () => response(200, detailOf(QUEUE[0], { library_epoch: epoch })),
});
epoch = OTHER;
await c.refresh();
await settle();
assert.equal(h.of('GET', `${ITEMS}/one`).length, 2);
assert.equal(c.state.detail.library_epoch, OTHER);
});
test('M4: leaving during the observe releases the capture it then answers', async () => {
let answer;
const { c, h } = await loaded({ [`POST ${OBSERVE}`]: () => new Promise(resolve => { answer = resolve; }) });
const applying = c.apply();
await settle();
c.dispose();
answer(response(200, observed('late')));
assert.equal(await applying, false);
await settle();
assert.deepEqual(h.of('DELETE', CAPTURE).map(call => call.url), [CAPTURE + 'late']);
assert.equal(h.of('POST', `${ITEMS}/one/decision`).length, 0);
});
test('M5: an unanswered decision that landed finishes as recorded, not as decided elsewhere', async () => {
let queue = QUEUE;
const { c, h } = await loaded({
[`GET ${ITEMS}`]: () => response(200, page(queue)),
[`POST ${ITEMS}/one/decision`]: () => Promise.reject(new TypeError('Failed to fetch')),
});
await c.apply();
assert.equal(c.state.canRetry, true);
queue = [QUEUE[1]]; // the server did record it: its change arrives
await c.refresh();
await settle();
assert.equal(c.state.phase, 'done');
assert.equal(c.state.toast, 'Association accepted.');
assert.doesNotMatch(c.state.warning, /decided elsewhere/);
assert.equal(h.of('DELETE', CAPTURE).length, 0, 'the server used the capture');
});
tools/vendor/frontend-coverage-baseline.json
Vendored
+3 -3
View File
@@ -390,15 +390,15 @@
"statements": 95
},
"workspace/review-state.mjs": {
"lines": 95,
"lines": 96,
"branches": 84,
"functions": 98,
"statements": 95
"statements": 96
},
"workspace/review.mjs": {
"lines": 98,
"branches": 89,
"functions": 91,
"functions": 93,
"statements": 98
},
"workspace/saved-views.mjs": {
web/js/workspace/review-state.mjs
+83 -19
View File
@@ -19,6 +19,11 @@
* - `sync(version)` is the stream's `ready`: it re-reads only when the queue read last is older than the version the
* stream names (`{library_epoch, review_revision}`, the pair every list read answers), so a visit reads its queue
* once (F3). An unknown version re-reads.
* - A change heard while a decision is in flight is owed: it is read once the decision is refused or unanswered (a
* recorded one leaves the page). Load more asked for during a re-read waits for it and continues from the queue it
* read; a re-read started during Load more takes the click over and reads that page too.
* - Apply fences on the epoch of the detail it decides: an observe under another storage releases its capture and
* reads the queue and the item again. An unanswered decision that a re-read shows landed finishes as recorded.
*/
import { noun } from './plural.mjs';
@@ -45,6 +50,7 @@ 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.',
storageChanged: 'Prompt storage was restored; reload the review queue.',
// 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.',
@@ -114,6 +120,10 @@ export function createReviewController({ api }) {
const lifetime = new AbortController();
let listRead = null, detailRead = null, pending = null, disposed = false;
let refreshing = null, refreshAgain = false;
// A change heard while a decision was in flight (I1); read when the phase is idle again.
let owedRefresh = false;
// Load more's own read, the one waiting for a re-read, and a click a re-read took over (I2).
let moreRead = null, moreTask = null, moreOwed = false, listing = null;
// What the first read owes once it answers: nothing, a re-read ('always'), or a re-read unless it is this version.
let afterLoad = null;
// The version of the queue read last: {epoch, revision}.
@@ -176,15 +186,26 @@ export function createReviewController({ api }) {
* Read the queue as deep as it is shown; `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 } = {}) {
function load(options) {
const run = listing = loadQueue(options);
run.finally(() => { if (listing === run) listing = null; });
return run;
}
async function loadQueue({ keepWarning = false, stale = false, quiet = false } = {}) {
listRead?.abort();
if (moreRead) { moreRead.abort(); moreRead = null; moreOwed = true; }
const read = listRead = new AbortController();
if (!quiet) set({ status: state.items.length ? state.status : 'loading', ...(keepWarning ? {} : { warning: '' }) });
let readOk = false;
try {
// One page, or as many as Load more had shown.
const queue = await readQueue(read, state.items.length > PAGE_LIMIT ? state.items.length : 0);
// One page, or as many as Load more had shown (and one page more for a click this read took over).
const depth = state.items.length + (moreOwed ? PAGE_LIMIT : 0);
const queue = await readQueue(read, depth > PAGE_LIMIT ? depth : 0);
if (read !== listRead || disposed) return;
moreOwed = false;
// M3: another storage (a restore) makes the open item's detail stale too.
if (state.epoch !== null && queue.first.library_epoch !== state.epoch) stale = true;
const items = Object.freeze(queue.items.map(entry => Object.freeze({ ...entry })));
const left = state.items.find(entry => entry.id === state.selectedId);
shown = Number.isSafeInteger(queue.first.review_revision)
@@ -195,15 +216,23 @@ export function createReviewController({ api }) {
const keep = items.find(entry => entry.id === state.selectedId);
// 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) {
if (quiet && left && !keep && pending?.id === left.id) {
// M5: the decision that got no answer was recorded (the server used its capture).
const choice = pending.body.choice;
pending = null;
set({ phase: 'done', warning: '', canRetry: false,
toast: choice === UNRESOLVED ? COPY.leftUnresolved : COPY.applied });
} else 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;
if (quiet && state.status === 'ready') set({ warning: COPY.refreshFailed });
else set({ status: 'error' });
// M7: a click this read took over can be made again.
moreOwed = false;
if (quiet && state.status === 'ready') set({ warning: COPY.refreshFailed, more: 'idle' });
else set({ status: 'error', more: 'idle' });
} finally {
// A change heard while the first read ran: that read may predate it, so read once more (unless the
// stream's ready named a version this read already holds).
@@ -242,29 +271,45 @@ export function createReviewController({ api }) {
refreshing = (async () => {
do {
refreshAgain = false;
// A decision in flight finishes first: the page leaves when it is recorded.
if (state.phase === 'applying') return;
// A decision in flight finishes first: the page leaves when it is recorded, and a refused or
// unanswered one reads the owed change (I1).
if (state.phase === 'applying') { owedRefresh = true; 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();
function loadMore() {
if (moreTask) return moreTask;
if (!state.nextCursor || state.more === 'loading') return Promise.resolve();
set({ more: 'loading' });
moreTask = readMore().finally(() => { moreTask = null; });
return moreTask;
}
async function readMore() {
// I2: a re-read in flight finishes first; the next page continues after the queue it read.
while ((refreshing || listing) && !disposed) await (refreshing ?? listing);
if (disposed || !state.nextCursor || state.status !== 'ready') {
if (state.more === 'loading') set({ more: 'idle' });
return;
}
if (state.more !== 'loading') set({ more: 'loading' });
const read = moreRead = new AbortController();
try {
const query = new URLSearchParams({ limit: String(PAGE_LIMIT), cursor: state.nextCursor });
const page = await api.get(`${REVIEW_URL}?${query}`, { signal: signalOf(read), cache: 'no-store' });
if (read !== listRead || disposed) return;
if (read !== moreRead || disposed) return;
moreRead = null;
const known = new Set(state.items.map(entry => entry.id));
const added = (page.items ?? []).filter(entry => !known.has(entry.id)).map(entry => Object.freeze({ ...entry }));
set({ items: Object.freeze([...state.items, ...added]), total: page.total, nextCursor: page.next_cursor ?? null,
more: 'idle' });
} catch (error) {
if (read !== listRead || disposed || isAbort(error)) return;
if (RELOAD.has(error?.code)) { await load({ stale: true }); return; }
if (read !== moreRead || disposed || isAbort(error)) return;
moreRead = null;
if (RELOAD.has(error?.code)) { set({ more: 'idle' }); await load({ stale: true }); return; }
set({ more: 'error' });
}
}
@@ -298,6 +343,13 @@ export function createReviewController({ api }) {
set({ choice: value, prompts: promptsOf(option) });
}
/** The phase is idle again after a refused or unanswered decision: read a change heard meanwhile (I1). */
function idleAgain(reloading) {
const owed = owedRefresh;
owedRefresh = false;
if (owed && !reloading && !disposed) refresh();
}
/** Send `body`; true once the decision is recorded. */
async function send(body) {
const id = state.selectedId;
@@ -316,11 +368,13 @@ export function createReviewController({ api }) {
if (isLost(error)) {
pending = { id, body };
set({ phase: 'idle', warning: COPY.lost, canRetry: true });
idleAgain(false);
return false;
}
pending = null;
release(body.capture_token);
set({ phase: 'idle', warning: error?.message || COPY.failed });
idleAgain(RELOAD.has(error?.code));
if (RELOAD.has(error?.code)) await load({ keepWarning: true, stale: true });
return false;
}
@@ -333,22 +387,32 @@ export function createReviewController({ api }) {
if (pending?.id === state.selectedId && pending.body.choice === choice) return retry();
dropPending();
set({ phase: 'applying', warning: '', canRetry: false });
let capture = null, epoch = detail.library_epoch;
let capture = null;
if (detail.target) {
let observed;
try {
const observed = await api.post(OBSERVE_URL, { target: detail.target }, { signal: lifetime.signal });
capture = observed.capture;
epoch = observed.library_epoch;
// M4: not aborted on leave, so the capture it answers can be released.
observed = await api.post(OBSERVE_URL, { target: detail.target });
} catch (error) {
if (disposed || isAbort(error)) return false;
set({ phase: 'idle', warning: error?.message || COPY.failed });
idleAgain(RELOAD.has(error?.code));
if (RELOAD.has(error?.code)) await load({ keepWarning: true, stale: true });
return false;
}
capture = observed.capture;
if (disposed) { release(capture?.token); return false; }
// M3: the file was observed under another storage than the detail was read from.
if (observed.library_epoch !== detail.library_epoch) {
release(capture?.token);
set({ phase: 'idle', warning: COPY.storageChanged });
idleAgain(true);
await load({ keepWarning: true, stale: true });
return false;
}
}
return send({
expected_epoch: epoch, expected_revision: detail.revision, capture_token: capture?.token ?? null,
expected_epoch: detail.library_epoch, expected_revision: detail.revision, capture_token: capture?.token ?? null,
sha256: capture?.sha256 ?? detail.sha256, dependency_fingerprint: detail.dependency_fingerprint, choice,
});
}
web/js/workspace/review.mjs
+83 -14
View File
@@ -15,6 +15,12 @@
* 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) names the queue's version, and the page re-reads only when the
* queue it shows is older (F3: one read per visit). A re-read keeps the depth Load more had reached. Nothing polls.
* - The stream lives only while the tab is visible: a hidden tab closes it and a visible one opens it again, so
* background tabs hold no connection (I4); the reopened stream's `ready` reads only when the queue changed meanwhile.
* A stream the browser gave up on (a 503 or a proxy error on its reconnect) is opened again after 3 s, doubling to
* 30 s; while the stream is down the page says so in a quiet status line (D-NEW-F2, I3).
* - A re-read refreshes each card whose media, target or availability changed (M6). Cards show the Images listing's
* thumbnail when the server knows one (M10).
* - The SAVED VIEWS row's count follows the total the page shows (`onTotal`), so a change costs no count read.
* - 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
@@ -36,6 +42,11 @@ import { showWorkspaceToast } from './toast.mjs';
import { imageKey } from './images-state.mjs';
export const REVIEW_EVENTS_URL = '/prompt_manager/review/events';
// I3: reopening a stream the browser closed for good waits this long, doubling up to the maximum.
export const REOPEN_FIRST_MS = 3000;
export const REOPEN_MAX_MS = 30000;
// D-NEW-F2: the page's line while the change stream is down (the concept has no such state).
export const NOT_LIVE = 'Live updates are paused: changes made elsewhere appear once the connection returns.';
const openEvents = url => new EventSource(window.ApiClient.resolveUrl(url));
function el(tag, text, className) {
@@ -57,9 +68,14 @@ const isDisabled = button => button.getAttribute('aria-disabled') === 'true';
export function imageOf(item) {
const target = item.target ?? {};
return { filename: item.filename, root_index: target.root_index, root_scope: target.root_scope, path: target.path,
url: item.url, is_video: item.is_video === true, media_type: item.is_video ? 'video' : 'image' };
url: item.url, thumbnail_url: item.thumbnail_url ?? null, is_video: item.is_video === true,
media_type: item.is_video ? 'video' : 'image' };
}
/** True when a re-read changed what a card shows or can open (M6). */
const cardChanged = (before, after) => before.url !== after.url || before.thumbnail_url !== after.thumbnail_url ||
before.unavailable !== after.unavailable || JSON.stringify(before.target) !== JSON.stringify(after.target);
/** The queue version a `ready` event names, or null (the page then re-reads). */
export function readyVersion(event) {
try {
@@ -120,6 +136,11 @@ export function mountReview(main, { api, scope, onDecision = () => {}, onTotal,
panel.append(panelTitle, filename, tagRow, contextField, warning, blocks, actions);
layout.append(grid, panel);
const live = el('p', NOT_LIVE, 'small review-live'); // D-NEW-F2: the change stream is down
live.setAttribute('role', 'status');
live.hidden = true;
heading.description.after(live);
const continuation = el('div', undefined, 'infinite-continuation'); // D119: Load more past 40 items
const moreText = el('p', '', 'small');
const more = createButton('Load more images');
@@ -136,6 +157,17 @@ export function mountReview(main, { api, scope, onDecision = () => {}, onTotal,
const cards = new Map();
let shownItems = null, shownSelected, shownOptions = null, shownPrompts, shownDetail;
function createCard(item) {
// D119: no selection checkbox and no star on a queue card.
const card = createImageCard(imageOf(item), { media: Boolean(item.url) });
card.dataset.reviewId = item.id;
card.item = item;
if (!item.url) setDisabled(card.querySelector('.image-open'), true);
if (!item.target) setDisabled(card.querySelector('.image-card-actions button:last-child'), true);
cards.set(item.id, card);
return card;
}
function renderCards(state) {
if (state.items !== shownItems) {
shownItems = state.items;
@@ -143,15 +175,13 @@ export function mountReview(main, { api, scope, onDecision = () => {}, onTotal,
for (const [id, card] of cards) if (!ids.has(id)) { card.remove(); cards.delete(id); }
state.items.forEach((item, index) => {
let card = cards.get(item.id);
if (!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;
if (!item.url) setDisabled(card.querySelector('.image-open'), true);
if (!item.target) setDisabled(card.querySelector('.image-card-actions button:last-child'), true);
cards.set(item.id, card);
}
if (card && cardChanged(card.item, item)) {
// M6: a root came or went: the card's media, Open and Tag follow the re-read.
const fresh = createCard(item);
card.replaceWith(fresh);
card = fresh;
} else if (card) card.item = item;
if (!card) card = createCard(item);
if (grid.children[index] !== card) grid.insertBefore(card, grid.children[index] ?? null);
});
shownSelected = undefined;
@@ -270,9 +300,48 @@ export function mountReview(main, { api, scope, onDecision = () => {}, onTotal,
scope.cleanup(controller.subscribe(reportTotal));
render(controller.state);
controller.load();
const events = connect(REVIEW_EVENTS_URL);
scope.cleanup(() => events.close());
events.addEventListener('ready', event => { if (!scope.signal.aborted) controller.sync(readyVersion(event)); });
events.addEventListener('change', () => { if (!scope.signal.aborted) controller.refresh(); });
connectEvents({ scope, connect, controller, live });
return { controller };
}
/**
* The change stream while the page is mounted: open while the tab is visible (I4), reopened with a bounded backoff
* after the browser gives up on it (I3), with `live` shown while it is down.
*/
function connectEvents({ scope, connect, controller, live }) {
let events = null, timer = null, delay = REOPEN_FIRST_MS;
const setLive = up => { live.hidden = up; };
function close() {
clearTimeout(timer);
timer = null;
events?.close();
events = null;
}
function open() {
if (events || scope.signal.aborted || document.hidden) return;
const source = events = connect(REVIEW_EVENTS_URL);
source.addEventListener('ready', event => {
if (source !== events || scope.signal.aborted) return;
delay = REOPEN_FIRST_MS;
setLive(true);
controller.sync(readyVersion(event));
});
source.addEventListener('change', () => { if (source === events && !scope.signal.aborted) controller.refresh(); });
source.addEventListener('error', () => {
if (source !== events || scope.signal.aborted) return;
setLive(false);
// The browser retries a dropped connection itself; one it closed for good is reopened here.
if (source.readyState !== 2) return; // EventSource.CLOSED
events = null;
source.close();
timer = setTimeout(() => { timer = null; open(); }, delay);
delay = Math.min(delay * 2, REOPEN_MAX_MS);
});
}
scope.listen(document, 'visibilitychange', () => {
if (document.hidden) close();
else open();
});
scope.cleanup(close);
open();
}
web/js/workspace/workspace.css
+2 -1
View File
@@ -386,4 +386,5 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
/* D119: a file that cannot be served (no current root) has an inert, empty media button that keeps the concept's
1.33 media area (app.js `.image-open img`), so its card lines up with the queue's other cards. */
.review-queue .image-open[aria-disabled="true"] { cursor: not-allowed; aspect-ratio: 1.33; /* D119 */ }
.review-panel > p.small { min-inline-size: 0; overflow-wrap: anywhere; }
/* D119: a long file name in the panel wraps inside the panel, where the concept's sample name never overflows. */
.review-panel > p.small { min-inline-size: 0; overflow-wrap: anywhere; /* D119 */ }