diff --git a/src/daemon/dispatch.ts b/src/daemon/dispatch.ts index a6c6ef0..8b00669 100644 --- a/src/daemon/dispatch.ts +++ b/src/daemon/dispatch.ts @@ -99,8 +99,20 @@ export interface DispatcherHost { audit(action: string, detail: string): void; } -/** Terminal statuses — a member in one of these will never run again. */ -const TERMINAL: ReadonlySet = new Set(["done", "failed", "blocked"]); +/** + * Terminal statuses — a task in one of these will never run again. + * + * Exported because the barrier and the client-facing panel view must agree on + * it: the barrier joins when every member is terminal, and the UI renders + * "settled" from the same rule. They were two separate literals in two files, + * so adding a status would have left the sidebar quietly disagreeing with the + * barrier about whether a fan-out had finished. + */ +export const TERMINAL_TASK_STATUS: ReadonlySet = new Set([ + "done", + "failed", + "blocked", +]); /** Statuses that mean "the worker's current turn is still in flight". */ const WORKER_ACTIVE: ReadonlySet = new Set([ @@ -663,9 +675,9 @@ export class Dispatcher { // one place: a non-terminal task has not completed, so it is not the // barrier's business. const self = members.find((m) => m.id === task.id); - if (!self || !TERMINAL.has(self.status)) return false; + if (!self || !TERMINAL_TASK_STATUS.has(self.status)) return false; - const pending = members.filter((m) => !TERMINAL.has(m.status)); + const pending = members.filter((m) => !TERMINAL_TASK_STATUS.has(m.status)); if (pending.length > 0) { this.#host.audit( "dispatch.group_waiting", diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index eb979ff..bec016f 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -74,6 +74,7 @@ import { import { Dispatcher, NonRetryableDispatchError, + TERMINAL_TASK_STATUS, type DispatcherHost, } from "./dispatch.js"; import { createPipelineManagerFromConfig } from "./pipeline/wiring.js"; @@ -200,13 +201,6 @@ function normalizeWorkdir(input: string): string | null { */ const PANEL_HISTORY_LIMIT = 5; -/** Task statuses that will never change again — mirrors the barrier's rule. */ -const TERMINAL_TASK_STATUS: ReadonlySet = new Set([ - "done", - "failed", - "blocked", -]); - const RESUME_MAX_SESSIONS = 50; const RESUME_DEADLINE_MS = 20_000; /** Per-session transcript read budget on resume. Scrollback keeps at most diff --git a/src/daemon/store.ts b/src/daemon/store.ts index 383d632..c9cbffe 100644 --- a/src/daemon/store.ts +++ b/src/daemon/store.ts @@ -460,6 +460,15 @@ export class Store { this.#db.exec(` CREATE INDEX IF NOT EXISTS idx_dispatch_group ON dispatch_tasks(group_id) WHERE group_id IS NOT NULL; + -- The client-facing panel poll (dispatchRecentGroups) runs every few + -- seconds while a collaboration is focused, and without this it could only + -- use the (account, project) index and then filter created_by row by row + -- over the tenant's ENTIRE task history — measured at 17.8ms per call over + -- 20k tasks, growing with history rather than with panel count. Partial on + -- group_id so it stays small: only grouped rows are ever polled this way. + CREATE INDEX IF NOT EXISTS idx_dispatch_panels + ON dispatch_tasks(account_id, project_id, created_by, created_at DESC) + WHERE group_id IS NOT NULL; `); // Pre-release single-row predecessor of provider_model_catalogs — never diff --git a/src/tests/dispatch-store.test.ts b/src/tests/dispatch-store.test.ts index ebb14ad..ffaf8b3 100644 --- a/src/tests/dispatch-store.test.ts +++ b/src/tests/dispatch-store.test.ts @@ -279,7 +279,11 @@ describe("migration — opening a database written before dispatch groups", () = /** Strip the group columns + index, simulating a pre-panel database file. */ function downgrade(dbPath: string): void { const db = new Database(dbPath); + // Both group indexes reference the columns below; SQLite refuses to drop a + // column an index depends on, so they go first. The REAL migration only ever + // ADDS, so this ordering is an artifact of simulating a downgrade. db.exec("DROP INDEX IF EXISTS idx_dispatch_group"); + db.exec("DROP INDEX IF EXISTS idx_dispatch_panels"); db.exec("ALTER TABLE dispatch_tasks DROP COLUMN group_ordinal"); db.exec("ALTER TABLE dispatch_tasks DROP COLUMN group_id"); db.close(); @@ -359,3 +363,81 @@ describe("migration — opening a database written before dispatch groups", () = expect(members.map((m) => m.groupOrdinal)).toEqual([1, 2]); }); }); + +// ── Index coverage for the polled panel query ─────────────────────────────── + +// `dispatchRecentGroups` backs a client poll that runs every few seconds while a +// collaboration is focused. Without a supporting index SQLite could only use the +// (account, project) index and then filter `created_by` row by row across the +// tenant's ENTIRE task history — measured at 17.8ms per call over 20k tasks, and +// growing with history rather than with panel count. Asserting the PLAN rather +// than a duration, because a timing assertion in CI is a flake generator. +describe("dispatchRecentGroups is index-covered", () => { + test("seeks on (account, project, created_by) instead of scanning history", () => { + const dbPath = join(tmp, "plan.db"); + const s = new Store(dbPath); + // A realistic long-lived daemon: mostly ungrouped conductor tasks, a few panels. + for (let i = 0; i < 300; i++) { + s.dispatchEnqueue({ + id: `bulk-${i}`, ...TENANT, kind: "send", shape: "ship", + targetSession: "t", prompt: "p", failureLimit: 2, + createdBy: "conductor:acc-a/proj-a", now: i, + }); + } + s.dispatchEnqueueGroup( + [1, 2, 3].map((n) => ({ + id: `grp-${n}`, ...TENANT, kind: "send" as const, shape: "scout" as const, + targetSession: `kid-${n}`, prompt: "review", failureLimit: 2, + createdBy: "orchestrator:goal-1", groupId: "g1", groupOrdinal: n, now: 9_000, + })), + ); + + const db = new Database(dbPath, { readonly: true }); + const plan = ( + db + .prepare( + `EXPLAIN QUERY PLAN + SELECT * FROM dispatch_tasks + WHERE account_id = ? AND project_id = ? AND created_by = ? + AND group_id IN ( + SELECT group_id FROM dispatch_tasks + WHERE account_id = ? AND project_id = ? AND created_by = ? + AND group_id IS NOT NULL + GROUP BY group_id ORDER BY MAX(created_at) DESC LIMIT ?) + ORDER BY created_at DESC, group_ordinal ASC, id ASC`, + ) + .all( + TENANT.accountId, TENANT.projectId, "orchestrator:goal-1", + TENANT.accountId, TENANT.projectId, "orchestrator:goal-1", 5, + ) as Array<{ detail: string }> + ).map((r) => r.detail); + db.close(); + + // Both the outer query and the group subquery must use it — the subquery is + // the one that would otherwise walk every task the tenant has ever queued. + const seeks = plan.filter((d) => d.includes("idx_dispatch_panels")); + expect(seeks.length).toBeGreaterThanOrEqual(2); + expect(plan.some((d) => /SCAN dispatch_tasks(?!.*USING)/.test(d))).toBe(false); + + // ...and it still returns the right rows, in fan-out order. + const members = s.dispatchRecentGroups( + TENANT.accountId, TENANT.projectId, "orchestrator:goal-1", 5, + ); + expect(members.map((m) => m.groupOrdinal)).toEqual([1, 2, 3]); + expect(members.every((m) => m.createdBy === "orchestrator:goal-1")).toBe(true); + }); + + test("never returns another dispatcher's groups", () => { + const s = new Store(join(tmp, "scoped.db")); + s.dispatchEnqueueGroup([ + { id: "a1", ...TENANT, kind: "send", shape: "scout", targetSession: "k", prompt: "p", + failureLimit: 2, createdBy: "orchestrator:goal-A", groupId: "gA", groupOrdinal: 1, now: 1 }, + { id: "b1", ...TENANT, kind: "send", shape: "scout", targetSession: "k", prompt: "p", + failureLimit: 2, createdBy: "orchestrator:goal-B", groupId: "gB", groupOrdinal: 1, now: 2 }, + ]); + const mine = s.dispatchRecentGroups(TENANT.accountId, TENANT.projectId, "orchestrator:goal-A"); + expect(mine.map((m) => m.id)).toEqual(["a1"]); + // And a different tenant sees nothing, group id or not. + expect(s.dispatchRecentGroups("acc-x", "proj-x", "orchestrator:goal-A")).toEqual([]); + }); +}); diff --git a/web/src/components/SessionListPane.test.tsx b/web/src/components/SessionListPane.test.tsx index ce42bf0..15c6be4 100644 --- a/web/src/components/SessionListPane.test.tsx +++ b/web/src/components/SessionListPane.test.tsx @@ -14,12 +14,12 @@ vi.mock("./AnalyticsPanel", () => ({ default: () => null })); vi.mock("./NewSessionModal", () => ({ openNewSessionModal: vi.fn() })); // Panel state is daemon-fed; drive it directly so the render is under test // rather than the polling transport. -const livePanelMock = vi.hoisted(() => vi.fn<() => unknown>(() => null)); +const liveProgressMock = vi.hoisted(() => vi.fn<() => unknown>(() => null)); const livePanelMemberMock = vi.hoisted(() => vi.fn<(id: string) => unknown>(() => null)); vi.mock("../state/panels", () => ({ fetchPanels: vi.fn(() => Promise.resolve()), resetPanels: vi.fn(), - livePanel: livePanelMock, + liveProgress: liveProgressMock, livePanelMember: livePanelMemberMock, })); @@ -215,7 +215,7 @@ describe("SessionListPane — live panel state", () => { ]; afterEach(() => { - livePanelMock.mockReturnValue(null); + liveProgressMock.mockReturnValue(null); livePanelMemberMock.mockReturnValue(null); }); @@ -226,20 +226,9 @@ describe("SessionListPane — live panel state", () => { }); it("shows settled-of-total on the orchestrator while a panel runs", () => { - livePanelMock.mockReturnValue({ - groupId: "g1", - createdAt: 0, - settled: 2, - joined: false, - members: [ - { sessionId: "p:review", ordinal: 1, status: "done" }, - { sessionId: "p:review-2", ordinal: 2, status: "blocked" }, - { sessionId: "p:search", ordinal: 3, status: "running" }, - ], - }); + liveProgressMock.mockReturnValue({ settled: 2, total: 3, panels: 1 }); ingestSessionList(FLEET); const { getByText, getByRole } = render(() => ); - expect(getByText("2/3")).toBeTruthy(); const bar = getByRole("progressbar"); expect(bar.getAttribute("aria-valuenow")).toBe("2"); @@ -249,25 +238,26 @@ describe("SessionListPane — live panel state", () => { it("counts SETTLED members, so a failed member does not stall the bar", () => { // The barrier joins on all-terminal. Counting successes would leave a // finished panel showing 2/3 forever — the exact confusion this removes. - livePanelMock.mockReturnValue({ - groupId: "g1", createdAt: 0, settled: 3, joined: false, - members: [ - { sessionId: "p:review", ordinal: 1, status: "done" }, - { sessionId: "p:review-2", ordinal: 2, status: "failed" }, - { sessionId: "p:search", ordinal: 3, status: "blocked" }, - ], - }); + liveProgressMock.mockReturnValue({ settled: 3, total: 3, panels: 1 }); ingestSessionList(FLEET); const { getByText, getByTitle } = render(() => ); expect(getByText("3/3")).toBeTruthy(); expect(getByTitle(/including failures/)).toBeTruthy(); }); + it("aggregates when TWO fan-outs are live at once", () => { + // Concurrent panels are legal — an orchestrator can start a second while the + // first resolves. Showing one panel's numbers understates the outstanding work. + liveProgressMock.mockReturnValue({ settled: 1, total: 5, panels: 2 }); + ingestSessionList(FLEET); + const { getByText, getByTitle } = render(() => ); + expect(getByText(/panels ×2/)).toBeTruthy(); + expect(getByText("1/5")).toBeTruthy(); + expect(getByTitle(/2 panels in flight/)).toBeTruthy(); + }); + it("badges each participating child with its position and state", () => { - livePanelMock.mockReturnValue({ - groupId: "g1", createdAt: 0, settled: 1, joined: false, - members: [{ sessionId: "x", ordinal: 1, status: "running" }], - }); + liveProgressMock.mockReturnValue({ settled: 1, total: 2, panels: 1 }); livePanelMemberMock.mockImplementation((id: string) => id === "p:review" ? { ordinal: 1, status: "running" } : id === "p:review-2" ? { ordinal: 2, status: "failed" } @@ -275,11 +265,9 @@ describe("SessionListPane — live panel state", () => { ); ingestSessionList(FLEET); const { getByTitle, queryAllByTitle } = render(() => ); - expect(getByTitle("Panel member 1 — working")).toBeTruthy(); // A failed member is SHOWN, not hidden — same rule as the joined digest. expect(getByTitle("Panel member 2 — failed")).toBeTruthy(); - // The non-participating child carries no badge. expect(queryAllByTitle(/Panel member/)).toHaveLength(2); }); }); diff --git a/web/src/components/SessionListPane.tsx b/web/src/components/SessionListPane.tsx index 6c8c4a4..63f90ff 100644 --- a/web/src/components/SessionListPane.tsx +++ b/web/src/components/SessionListPane.tsx @@ -16,7 +16,7 @@ import { type FilteredFleetGroup, } from "../lib/fleet"; import { sessionAgentLabel, shortSub } from "../lib/identity"; -import { fetchPanels, livePanel, livePanelMember, resetPanels } from "../state/panels"; +import { fetchPanels, liveProgress, livePanelMember, resetPanels } from "../state/panels"; import { nowTick } from "../state/clock"; import { focusedSession, @@ -91,17 +91,29 @@ const SessionListPane: Component = () => { filterFleet(groupFleet(sessionList()), filter()), ); - // Poll panel state for the focused collaboration only. There is no push - // channel for dispatch state, and a fan-out changes over minutes — a static - // indicator would show a finished panel as still running. Scoped to the - // focused goal rather than every visible fleet so the cost stays one cheap - // metadata query per tick regardless of how many collaborations exist. - createEffect(() => { + // The goal whose panels we poll, as a plain STRING. + // + // A memo over a primitive, deliberately. Reading `focusedSession()?.collaboration` + // directly inside the effect subscribed it to that store property — and + // `mergeSession` reassigns every field on each `session.info_update` with no + // equality guard, so `collaboration` gets a fresh object reference on every + // status flip. The effect then tore down and rebuilt its interval, firing an + // immediate poll each time: during a tool-heavy turn that turned a 3s cadence + // into a burst of queries. A memo returning a string only notifies when the id + // actually changes. + const polledGoalId = createMemo(() => { const focused = focusedSession(); - const goalId = - focused?.collaboration !== undefined - ? focused.id - : focused?.collaborationRole?.parentSessionId; + if (!focused) return null; + if (focused.collaboration !== undefined) return focused.id; + return focused.collaborationRole?.parentSessionId ?? null; + }); + + // There is no push channel for dispatch state, and a fan-out changes over + // minutes — a static indicator would show a finished panel as still running. + // Scoped to the focused goal so the cost stays one cheap metadata query per + // tick regardless of how many collaborations exist. + createEffect(() => { + const goalId = polledGoalId(); if (!goalId) { resetPanels(); return; @@ -443,11 +455,12 @@ const SessionRow: Component<{ {/* The fan-out, while it runs. This is the whole point of a panel and was invisible until the daemon started reporting it. */} - + {(p) => ( )} @@ -543,14 +556,22 @@ const SessionRow: Component<{ * leave the bar short of full on a panel that is already done, which is exactly * the confusion the indicator exists to remove. */ -const PanelProgress: Component<{ settled: number; total: number }> = (props) => { +const PanelProgress: Component<{ + settled: number; + total: number; + /** How many fan-outs are live. More than one is legal — an orchestrator can + * run a second panel while the first is still resolving. */ + panels: number; +}> = (props) => { const pct = () => (props.total === 0 ? 0 : (props.settled / props.total) * 100); return (
- ⇉ panel + + ⇉ {props.panels === 1 ? "panel" : `panels ×${props.panels}`} + { expect(livePanelMember("kid-1")).toBeNull(); }); }); + +describe("concurrent fan-outs", () => { + // Two live panels at once is legal — an orchestrator can start a second while + // the first resolves, which is why the dispatcher watches a SET of tasks per + // session. Reporting only the first understated the parallelism this UI exists + // to show, and left a child in the other panel with no badge. + const twoLive = () => + result([ + panel("newer", ["running", "queued"]), + panel("older", ["done", "running", "running"]), + panel("finished", ["done", "done"]), + ]); + + beforeEach(async () => { + requestMock.mockResolvedValueOnce(twoLive()); + await fetchPanels("goal-1"); + requestMock.mockReset(); + }); + + it("reports every live fan-out, not just the newest", () => { + expect(livePanels().map((p) => p.groupId)).toEqual(["newer", "older"]); + expect(livePanel()?.groupId).toBe("newer"); + }); + + it("aggregates progress across all of them", () => { + // newer: 0 of 2 settled; older: 1 of 3. A single panel's numbers would + // misstate how much work is outstanding. + expect(liveProgress()).toEqual({ settled: 1, total: 5, panels: 2 }); + }); + + it("finds a member of an OLDER live fan-out", () => { + // `older` has three members; kid-3 exists only there. + expect(livePanelMember("kid-3")).toEqual({ ordinal: 3, status: "running" }); + }); + + it("liveProgress is null when everything has joined", async () => { + requestMock.mockResolvedValueOnce(result([panel("finished", ["done", "done"])])); + await fetchPanels("goal-1"); + expect(liveProgress()).toBeNull(); + expect(livePanels()).toEqual([]); + }); +}); diff --git a/web/src/state/panels.ts b/web/src/state/panels.ts index 2119336..f276da0 100644 --- a/web/src/state/panels.ts +++ b/web/src/state/panels.ts @@ -64,16 +64,45 @@ export async function fetchPanels(sessionId: string): Promise { } /** - * The fan-out currently in flight, if any. + * Every fan-out currently in flight. * - * "Live" means not yet joined — deliberately keyed off `joined` rather than off - * member statuses, so this can never disagree with the barrier's own rule about - * when a panel is finished. + * PLURAL, because concurrent panels are legal: an orchestrator can start a + * second one while the first is still resolving, which is exactly why the + * dispatcher tracks a SET of watchers per session. Reporting only the first + * unjoined panel under-counted the parallelism this UI exists to show — a child + * in the other live panel got no badge at all. + * + * "Live" means not yet joined — keyed off `joined` rather than off member + * statuses, so this can never disagree with the barrier's own rule about when a + * panel is finished. */ +export const livePanels = createMemo(() => + state().panels.filter((p) => !p.joined), +); + +/** The first live fan-out, for callers that only need one. */ export const livePanel = createMemo( - () => state().panels.find((p) => !p.joined) ?? null, + () => livePanels()[0] ?? null, ); +/** + * Aggregate progress across every live fan-out, or null when none is running. + * + * Summed rather than showing one panel's numbers: with two panels in flight a + * single panel's "1/3" is a lie about how much work is outstanding. + */ +export const liveProgress = createMemo< + { settled: number; total: number; panels: number } | null +>(() => { + const live = livePanels(); + if (live.length === 0) return null; + return { + settled: live.reduce((n, p) => n + p.settled, 0), + total: live.reduce((n, p) => n + p.members.length, 0), + panels: live.length, + }; +}); + /** The most recently joined fan-out — context once the live one is gone. */ export const lastJoinedPanel = createMemo( () => state().panels.find((p) => p.joined) ?? null, @@ -88,10 +117,13 @@ export const lastJoinedPanel = createMemo( export function livePanelMember( sessionId: string, ): { ordinal: number; status: CollaborationPanel["members"][number]["status"] } | null { - const p = livePanel(); - if (!p) return null; - const m = p.members.find((x) => x.sessionId === sessionId); - return m ? { ordinal: m.ordinal, status: m.status } : null; + // Searches EVERY live fan-out, not just the newest: a child participating only + // in an older still-running panel showed no badge before this. + for (const p of livePanels()) { + const m = p.members.find((x) => x.sessionId === sessionId); + if (m) return { ordinal: m.ordinal, status: m.status }; + } + return null; } export function resetPanels(): void {