From 034b1c97a08d301e4e06d7e5fa5215c5fe4dc84b Mon Sep 17 00:00:00 2001 From: Seanathon Date: Wed, 19 Aug 2026 14:52:23 -0700 Subject: [PATCH] feat: tell a queued item it's queued, and where it stands in line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Captures run one at a time, so a second add waits on the first. The card said "Capturing the page" the whole time — a claim about work that had not started. It now says "Next up" or "Queued, 3rd in line", counting down as the line moves. Two parts, and they're different sizes. Stopping the lie needed no server change: `status` already distinguishes pending from processing and already flows through both SSE and hydrateItemForUi. The ordinal position is what pulls in the queue, the SSE event, and hydration. The new render state keys on `queuePosition`, NOT on `status`. `pending` with no job behind it is a real, persistent state — manual-upload boards, a missing source, an unregistered ingest_mode, and the 150 legacy imports that sat at pending with complete data — and the in-memory line is empty after a restart. Gating on the position means the queued state appears exactly when real queue information exists and degrades to the old behavior otherwise. The line is tracked in the queue rather than derived from `status='pending'` in SQL, for the same reason: those never-enqueued items would inflate everyone else's position permanently. Position rides on a normal `pending` transition rather than a new `queued` status. An unknown status would fall through itemRenderState's pending/ processing gate and render a waiting card as FINISHED. Both SSE application paths clear the position on absence rather than only assigning when present — the `processing` transition carries none, and a stale one would leave the card claiming to be queued for the whole capture. And `isInFlight` counts queued, or applyFilters would drop the card the user just added, since it has no facets to match on yet. Known imprecision, noted in the code: runSnapshotJob enqueues directly, so an archival snapshot holds the lane without appearing in the line, and "next up" can wait one out. Snapshots are opt-in and status-neutral; not worth plumbing. Verified live: four rapid adds show 1..4, positions survive a hard reload (the hydrate path), and the line advances 4th->3rd->2nd->Next up->Capturing the page->Reading it with no stale label left behind. 598 tests + typecheck green. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01C83mrW9X8zBLY1sSZRgdCa --- public/index.html | 25 +++++++++++++++-- src/collections-ui.js | 21 +++++++++++++-- src/collections-ui.test.ts | 38 ++++++++++++++++++++++++++ src/db/hydrate.ts | 7 +++++ src/db/queue.ts | 55 ++++++++++++++++++++++++++++++++++++++ src/sse.ts | 8 ++++++ 6 files changed, 150 insertions(+), 4 deletions(-) diff --git a/public/index.html b/public/index.html index 7d2f829..a9881f4 100644 --- a/public/index.html +++ b/public/index.html @@ -2617,6 +2617,11 @@

Tags

item.status = event.status; if (event.error_reason !== undefined) item.error_reason = event.error_reason; else delete item.error_reason; + // Cleared on absence, not just set when present: the `processing` transition + // carries no position, and a stale one would leave the card saying it is queued + // for the whole capture (itemRenderState keys the queued state on this). + if (event.queuePosition != null) item.queuePosition = event.queuePosition; + else delete item.queuePosition; if (event.fields && typeof event.fields === 'object') { // `fields` arrives flat and dotted ("meta.tier"); the renderers read it nested. for (const [key, value] of Object.entries(event.fields)) { @@ -3666,7 +3671,22 @@

Tags

// State comes from one tested helper (collections-ui.itemRenderState) so the card, // the list row and the modal can never disagree about what an item is doing. - const STATE_LABEL = { capturing: 'Capturing the page', reading: 'Reading it' }; + const STATE_LABEL = { queued: 'Waiting its turn', capturing: 'Capturing the page', reading: 'Reading it' }; + + /** + * What a waiting card says. Captures run one at a time, so the honest answer is + * where you are in line rather than a claim that anything is happening yet. The + * line is global (one worker across all boards), so the copy deliberately avoids + * implying "on this board". Falls back to the plain label when the position is + * missing — a reload before the first SSE frame, say. + */ + const ORDINALS = ['', 'next', '2nd', '3rd', '4th', '5th', '6th', '7th', '8th', '9th']; + function queueLabel(b) { + const pos = b && b.queuePosition; + if (!pos) return STATE_LABEL.queued; + if (pos === 1) return 'Next up'; + return `Queued, ${ORDINALS[pos] || pos + 'th'} in line`; + } function stateOf(b) { return window.collectionHelpers.itemRenderState(b); } @@ -3685,7 +3705,8 @@

Tags

+ `${esc(reason.charAt(0).toUpperCase() + reason.slice(1))}.` + ``; } - return `
${STATE_LABEL[state]}
`; + const label = state === 'queued' ? queueLabel(b) : STATE_LABEL[state]; + return `
${esc(label)}
`; } /** diff --git a/src/collections-ui.js b/src/collections-ui.js index 071660f..655bf0f 100644 --- a/src/collections-ui.js +++ b/src/collections-ui.js @@ -109,6 +109,11 @@ export function applySseEvent(card, event) { next.fields = { ...(card.fields || {}), ...event.fields }; } if (event.error_reason !== undefined) next.errorReason = event.error_reason; + // Cleared, not just assigned-when-present: the `processing` transition carries no + // position, and a stale one would leave the card claiming to be queued for the whole + // capture (itemRenderState keys the queued state on this). + if (event.queuePosition != null) next.queuePosition = event.queuePosition; + else delete next.queuePosition; return next; } @@ -401,15 +406,27 @@ export function itemRenderState(item) { if (item.status === "error") return "failed"; if (item.status !== "pending" && item.status !== "processing") return "ready"; if (hasAiRead(item)) return "ready"; + // Jobs run one at a time, so an add made while another is capturing sits in line. + // Keyed on the POSITION, not on `status`: `pending` with no job behind it is a real + // persistent state (manual-upload boards, missing source, unregistered ingest_mode, + // legacy imports) and the in-memory line is empty after a restart — in all of those + // there is no position, and we fall through rather than claim a queue that isn't + // there. Ranked below `hasAiRead` so a re-queued item that already has its read + // keeps showing content instead of reverting to a skeleton. + if (item.queuePosition != null) return "queued"; // Capture writes title + screenshot before enrichment runs, so either one means // the page is in hand and only the AI read is outstanding. return item.title || item.screenshot ? "reading" : "capturing"; } -/** True while an item is still being captured or read (skeleton showing). */ +/** + * True while an item is still being captured or read (skeleton showing). `queued` + * counts: an item waiting its turn has no facets to match on yet, and applyFilters + * leans on this to keep the just-added card visible instead of filtering it away. + */ export function isInFlight(item) { const state = itemRenderState(item); - return state === "capturing" || state === "reading"; + return state === "queued" || state === "capturing" || state === "reading"; } // --- Descriptor-driven card summary ---------------------------------------------- diff --git a/src/collections-ui.test.ts b/src/collections-ui.test.ts index 3477ecf..681e5aa 100644 --- a/src/collections-ui.test.ts +++ b/src/collections-ui.test.ts @@ -317,6 +317,21 @@ test("applySseEvent fills the card on a done event (fields from payload)", () => assert.deepEqual(next.fields, { a: 1, b: 2 }, "fields merged from the SSE payload (no refetch)"); }); +// The position must CLEAR when absent, not merely be assigned when present: the +// `processing` transition carries no position, and a stale one left behind would keep +// the card reading as queued for the whole capture (itemRenderState keys on it). +test("applySseEvent carries a queue position and clears it on leaving the line", () => { + const card = { id: "i1", status: "pending" }; + const queued = applySseEvent(card, { itemId: "i1", status: "pending", queuePosition: 2 }); + assert.equal(queued.queuePosition, 2); + + const moved = applySseEvent(queued, { itemId: "i1", status: "pending", queuePosition: 1 }); + assert.equal(moved.queuePosition, 1, "position updates as the line advances"); + + const started = applySseEvent(moved, { itemId: "i1", status: "processing" }); + assert.equal(started.queuePosition, undefined, "no position on the event → cleared"); +}); + test("applySseEvent sets error state on an error event", () => { const card = { id: "i1", status: "processing", fields: {} }; const next = applySseEvent(card, { itemId: "i1", status: "error", error_reason: "timed out" }); @@ -472,6 +487,29 @@ test("itemRenderState: a freshly added item with nothing captured yet is capturi assert.equal(itemRenderState({ status: "processing", title: "", url: "https://x" }), "capturing"); }); +// Jobs run one at a time, so a second add sits in line while the first captures. It +// used to render "Capturing the page" — a claim about work that had not started. +// Keyed on `queuePosition`, NOT on `status`: `pending` with no job behind it is a real +// persistent state (manual-upload boards, a missing source, an unregistered +// ingest_mode, and the legacy imports above), and the in-memory line is empty after a +// restart. No position → fall through to the old behaviour rather than claim a queue +// that isn't there. +test("itemRenderState: an item waiting its turn in the job line is queued", () => { + assert.equal(itemRenderState({ status: "pending", title: "", url: "https://x", queuePosition: 2 }), "queued"); + // position 1 is still WAITING — the job has not started until the status flips + assert.equal(itemRenderState({ status: "pending", title: "", url: "https://x", queuePosition: 1 }), "queued"); + // no position → unchanged, so a stale-pending item never regresses into a fake queue + assert.equal(itemRenderState({ status: "pending", title: "", url: "https://x" }), "capturing"); + // an item that already carries its AI read is still ready — never ghost real content + assert.equal(itemRenderState({ status: "pending", title: "Mastra", meta: { tier: "reference" }, queuePosition: 3 }), "ready"); +}); + +// A queued card must survive the active filters exactly like a capturing one: it has no +// facets to match on yet, so `isInFlight` false would drop the card the user just added. +test("isInFlight: a queued item still counts as in flight", () => { + assert.equal(isInFlight({ status: "pending", title: "", url: "https://x", queuePosition: 2 }), true); +}); + test("itemRenderState: capture landed but the AI read has not is reading", () => { assert.equal(itemRenderState({ status: "processing", title: "eve", screenshot: "s.png" }), "reading"); assert.equal(itemRenderState({ status: "pending", title: "eve" }), "reading"); diff --git a/src/db/hydrate.ts b/src/db/hydrate.ts index c4ece5a..7c58f86 100644 --- a/src/db/hydrate.ts +++ b/src/db/hydrate.ts @@ -1,6 +1,7 @@ import { and, desc, eq, gte, inArray } from 'drizzle-orm'; import { assets, items, type Item, type Asset } from './schema.js'; +import { jobLinePositionOf } from './queue.js'; import type { DbHandle } from './index.js'; // Story 8.x cutover — present a SQLite item in the shape the (polished, prototype) @@ -20,6 +21,12 @@ export function hydrateItemForUi(item: Item, itemAssets: Asset[] = []): Record(fn: () => T | Promise): Promise { // Concurrency 1 is load-bearing: Chromium is ~400-520MB resident, so two concurrent // captures OOM the 512MB-1GB LXC (NFR-1/C1). +// --- The visible job line ------------------------------------------------------- +// +// Item ids whose job is enqueued but has not started, in line order. Purely for +// telling the user where they stand: jobs run one at a time, so a second add waits on +// the first, and the card used to say "Capturing the page" the whole time — a claim +// about work that had not begun. +// +// Tracked HERE rather than derived from `status='pending'` in SQL, because pending is +// not the same thing as queued: an item with no source, a manual-upload board, or an +// unregistered ingest_mode stays pending forever with no job behind it, and would +// inflate everyone else's position permanently. +// +// Caveat: `runSnapshotJob` calls `enqueueJob` directly, so an archival snapshot holds +// the lane without appearing here — "next up" can wait out one. Snapshots are opt-in +// and status-neutral, so this is left as a known imprecision rather than plumbed. +const jobLine: string[] = []; + +function joinJobLine(itemId: string): void { + if (!jobLine.includes(itemId)) jobLine.push(itemId); +} + +function leaveJobLine(itemId: string): void { + const i = jobLine.indexOf(itemId); + if (i !== -1) jobLine.splice(i, 1); +} + +/** 1-based place in the line, or undefined when this item isn't waiting. */ +export function jobLinePositionOf(itemId: string): number | undefined { + const i = jobLine.indexOf(itemId); + return i === -1 ? undefined : i + 1; +} + +/** + * Re-announce every waiting item's position. Called when the line changes (someone + * joins, or someone's turn arrives and the rest move up). The item's real DB status is + * published alongside — `pending` — so the client never has to learn a status value + * that isn't in the schema; the position is the only new information. + */ +function publishJobLine(handle: DbHandle): void { + jobLine.forEach((id, i) => { + const row = handle.db.select().from(items).where(eq(items.id, id)).get(); + if (!row) return; + statusHub.publish({ itemId: id, boardId: row.boardId, status: row.status, queuePosition: i + 1 }); + }); +} + /** A schedulable unit of work. `run` receives an AbortSignal it must honor. */ export interface Job { type: string; @@ -303,6 +349,9 @@ export async function runItemJob(handle: DbHandle, args: RunItemJobArgs): Promis timeoutMs: args.timeoutMs, teardown: args.teardown, run: async (signal) => { + // Our turn: leave the line, then tell everyone behind us they moved up. + leaveJobLine(args.itemId); + publishJobLine(handle); setItemStatusDirect(handle, args.itemId, 'processing', null); try { await args.work(signal); @@ -323,7 +372,13 @@ export async function runItemJob(handle: DbHandle, args: RunItemJobArgs): Promis }, }; + // Join the line BEFORE enqueueing, and announce it, so the card the user just added + // says what it is actually doing instead of claiming to be capturing. + joinJobLine(args.itemId); + publishJobLine(handle); + const result = await enqueueJob(job, { timeoutFn: args.timeoutFn }); + leaveJobLine(args.itemId); // no-op on the normal path; a belt-and-braces cleanup // Timeout: the work was abandoned (possibly still `processing`) — record the // terminal error status through the writer so the item is never stuck. diff --git a/src/sse.ts b/src/sse.ts index 7015ab0..53cf7d3 100644 --- a/src/sse.ts +++ b/src/sse.ts @@ -19,6 +19,14 @@ export interface StatusEvent { */ title?: string; screenshot?: string; + /** + * 1-based place in the job line, on transitions for an item that is enqueued but has + * not started. Absent once the job begins — the client CLEARS it on absence rather + * than only assigning it when present, so a card can't be stranded showing a stale + * position. The line is global (one worker for all boards), so position 2 means two + * captures ahead of you anywhere, not on this board. + */ + queuePosition?: number; } /** A minimal write sink (the SSE response stream). */