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 @@
`;
}
/**
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). */