From a5f9444b2a5775ab6f3ab40e524f0698c51efd85 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 25 Oct 2026 07:59:05 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20attempt=20generations=20on=20dispat?= =?UTF-8?q?ch=20=E2=80=94=20fence=20stale=20worker=20results,=20held=20tas?= =?UTF-8?q?ks,=20loop=20attempt=20routing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every claim and reclaim of a dispatch task now bumps a generation, and every state change a worker result can cause (complete, fail, mark-running, hold, lease renewal) is guarded on that generation and a live claim. Before this, dispatchComplete updated by id alone, so a worker replaced after a reclaim could overwrite its successor's outcome or resurrect a task already terminal. A fenced result is a no-op, logged and audited as dispatch.stale_result. Adds a non-terminal `held` status that reclaim never selects, with hold/unhold on the store and dispatcher, and a `attempt` task kind carrying run_id/attempt_id for Loops (docs/loop-engineering-design.md §5, §6, §13 B). Attempts route to optional host hooks (spawnAttemptWorker, continueAttempt, onAttemptSettled, onAttemptWaiting), fail closed on a host without them, emit no fleet events, are never torn down by the dispatcher, report an approval wedge to the loop, and are never re-queued: a reclaim settles as `reclaimed` and any failure as `failed`. Additive migration: run_id, attempt_id, generation (default 0) via #addColumnIfMissing, plus a partial run_id index after the ALTERs. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/protocol/src/types.ts | 14 +- src/daemon/dispatch.ts | 374 ++++++++++++++-- src/daemon/fleet.ts | 2 +- src/daemon/session-manager.ts | 8 +- src/daemon/store.ts | 211 +++++++-- src/tests/collaboration.test.ts | 27 +- src/tests/dispatch-generations.test.ts | 589 +++++++++++++++++++++++++ src/tests/dispatch-host.test.ts | 19 +- src/tests/dispatch-store.test.ts | 36 +- 9 files changed, 1183 insertions(+), 97 deletions(-) create mode 100644 src/tests/dispatch-generations.test.ts diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index e48e0325..49adf1e5 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -2346,9 +2346,14 @@ export interface CollaborationPanelsResultMsg { */ export interface FleetTaskWire { id: string; - kind: "send" | "spawn"; + /** `attempt` = one attempt of a loop run (docs/loop-engineering-design.md §5). */ + kind: "send" | "spawn" | "attempt"; shape: "ship" | "scout"; - status: "queued" | "claimed" | "running" | "done" | "failed" | "blocked"; + /** + * `held` is non-terminal: the task is parked on something outside its worker + * (an approval a loop's profile does not cover) and is never reclaimed. + */ + status: "queued" | "claimed" | "running" | "held" | "done" | "failed" | "blocked"; attempts: number; /** Epoch ms. */ createdAt: number; @@ -2363,6 +2368,9 @@ export interface FleetTaskWire { createdBy: string; /** Dispatch group (fan-out barrier); absent = a standalone task. */ groupId?: string; + /** attempt: the loop run and attempt this task executes. Absent on send/spawn. */ + runId?: string; + attemptId?: string; /** * RESERVED and never populated in P5. The daemon has no dependency model — * the conductor sequences in prose. Present so typed fan-in/join edges are a @@ -2384,7 +2392,7 @@ export interface FleetEventWire { /** Fleet-wide rollup, normalized across backends so one number spans vendors. */ export interface FleetUsage { - /** Tasks not yet terminal (queued + claimed + running). */ + /** Tasks not yet terminal (queued + claimed + running + held). */ activeTasks: number; /** Tasks in `blocked` — the anti-spin failure cap tripped; needs a human. */ blockedTasks: number; diff --git a/src/daemon/dispatch.ts b/src/daemon/dispatch.ts index 3f1b34ac..8b74510a 100644 --- a/src/daemon/dispatch.ts +++ b/src/daemon/dispatch.ts @@ -21,10 +21,25 @@ * lease expiry reclaims the task. * - Events are written before delivery and marked delivered after — a * crash between "worker finished" and "conductor saw it" re-delivers. + * - Every claim and reclaim bumps the task's generation. A worker is watched + * under the generation it was started from, and only that generation can + * complete, fail, or hold the task — a replaced worker finishing late is a + * no-op (audited as `dispatch.stale_result`), never an overwrite. + * + * Loop attempts (`kind: "attempt"`, docs/loop-engineering-design.md §5) ride + * the same queue — the dispatcher stays the sole scheduler of worker attempts — + * but settle through the host's attempt hooks instead of fleet events: no + * conductor turn, no teardown on finish, an approval wedge reports to the loop, + * and a reclaim is terminal (`reclaimed`), because retrying is the loop's call. */ import { randomUUID } from "node:crypto"; -import type { DispatchEventRow, DispatchTaskRow, Store } from "./store.js"; +import type { + DispatchEventRow, + DispatchTaskKind, + DispatchTaskRow, + Store, +} from "./store.js"; import type { SessionStatus } from "../protocol/types.js"; export interface DispatchConfig { @@ -75,6 +90,12 @@ export const DEFAULT_DISPATCH_CONFIG: DispatchConfig = { /** Thrown by host methods when retrying can never succeed (target gone, bad input). */ export class NonRetryableDispatchError extends Error {} +/** The hooks a host must implement before an attempt task may run on it. */ +const ATTEMPT_HOOKS = "spawnAttemptWorker / onAttemptSettled / onAttemptWaiting are not implemented"; + +/** How an attempt task ended, as reported to `DispatcherHost.onAttemptSettled`. */ +export type AttemptSettledStatus = "done" | "failed" | "reclaimed"; + /** * The execution surface the SessionManager implements. Deliberately narrow: * the dispatcher owns lifecycle/ordering; the host owns sessions. @@ -96,6 +117,35 @@ export interface DispatcherHost { buildWorkerDigest(task: DispatchTaskRow): string; /** Tear down a finished worker session (best-effort). */ destroyWorker(sessionId: string, reason: string): Promise; + /** + * Loop attempts (§5). All four are optional so a host without a loop service + * keeps compiling unchanged — and an attempt task reaching such a host fails + * closed rather than running as an ordinary spawn, which would drop the + * loop's permission profile and budgets on the floor. + * + * Start a worker for an attempt task. Returns its session id. The host owns + * that session for the attempt's whole life: the dispatcher never destroys it. + */ + spawnAttemptWorker?(task: DispatchTaskRow): Promise<{ sessionId: string }>; + /** + * Re-attach an attempt to its surviving worker session. False = the session + * is gone; the dispatcher then starts a fresh one through `spawnAttemptWorker`. + * Optional even for a loop host: without it, continuation is always fresh. + */ + continueAttempt?(task: DispatchTaskRow): Promise; + /** + * An attempt task reached a terminal state. `reclaimed` means its claim died + * (crash or lease expiry) and the dispatcher did NOT retry it — whether to + * start another attempt is the loop's decision, against the loop's budget. + * `task` is the row as stored after the transition. + */ + onAttemptSettled?(task: DispatchTaskRow, status: AttemptSettledStatus): void; + /** + * An attempt's worker is waiting on an approval. The loop decides whether + * it falls inside the run's profile or parks the task with `hold()`; the + * dispatcher neither fails it nor tells a conductor. + */ + onAttemptWaiting?(task: DispatchTaskRow): void; /** * Inject pending events into whichever session each one belongs to, batched * per target — the tenant's conductor for its own dispatches, and the @@ -166,14 +216,17 @@ export class Dispatcher { * running one). With one id per key the second registration silently evicted * the first, whose task then sat in `running` until the lease expired — and * its barrier hung for the whole lease with it. + * + * Each task id maps to the generation it was started under — the one its + * completion must present to the store. */ - #watched = new Map>(); + #watched = new Map>(); - /** Register a task as watching `sessionId`. */ - #watch(sessionId: string, taskId: string): void { + /** Register a task (at the generation it was started under) as watching `sessionId`. */ + #watch(sessionId: string, taskId: string, generation: number): void { const existing = this.#watched.get(sessionId); - if (existing) existing.add(taskId); - else this.#watched.set(sessionId, new Set([taskId])); + if (existing) existing.set(taskId, generation); + else this.#watched.set(sessionId, new Map([[taskId, generation]])); } /** Stop watching one task; drops the key when it was the last watcher. */ @@ -187,6 +240,10 @@ export class Dispatcher { set.delete(taskId); if (set.size === 0) this.#watched.delete(sessionId); } + /** Drop every watch held by one task, whichever session it is on. */ + #unwatchTask(taskId: string): void { + for (const sessionId of [...this.#watched.keys()]) this.#unwatch(sessionId, taskId); + } /** Re-entrancy guard — a slow tick must not overlap the next. */ #ticking = false; #deliveringEvents = false; @@ -207,12 +264,12 @@ export class Dispatcher { // a shared send target can have several, and this accessor exists for the // spawn case — see `tasksForSession` when you need all of them. const set = this.#watched.get(sessionId); - return set ? set.values().next().value : undefined; + return set ? set.keys().next().value : undefined; } /** Every task currently watching `sessionId`. */ tasksForSession(sessionId: string): string[] { - return [...(this.#watched.get(sessionId) ?? [])]; + return [...(this.#watched.get(sessionId)?.keys() ?? [])]; } start(): void { @@ -239,11 +296,14 @@ export class Dispatcher { enqueue(input: { accountId: string; projectId: string; - kind: "send" | "spawn"; + kind: DispatchTaskKind; shape: "ship" | "scout"; targetSession?: string; workdir?: string; prompt: string; + /** Loop run + attempt this task executes. Required for `kind: "attempt"`. */ + runId?: string; + attemptId?: string; /** * Backend the spawned worker runs on. Omit for the daemon default. * Validated by the caller (the fleet tool / SessionManager) against the @@ -256,18 +316,31 @@ export class Dispatcher { model?: string; createdBy: string; }): string { + const isAttempt = input.kind === "attempt"; + // An attempt row without its loop ids could never be reconciled to its + // run, and loop ids on a send/spawn would make a fleet task look like one. + if (isAttempt !== (input.runId !== undefined && input.attemptId !== undefined)) { + throw new Error( + isAttempt + ? "an attempt task needs both runId and attemptId" + : `runId/attemptId are only valid on an attempt task, not ${input.kind}`, + ); + } const id = randomUUID(); this.#store.dispatchEnqueue({ id, ...input, - failureLimit: this.#config.failureLimit, + // One claim per attempt: a reclaim is terminal and the loop decides what + // happens next, so the queue's own retry budget must never re-run one. + failureLimit: isAttempt ? 1 : this.#config.failureLimit, ownerDaemon: this.#config.daemonId, now: Date.now(), }); this.#host.audit( "dispatch.enqueued", `task=${id} kind=${input.kind} shape=${input.shape} target=${input.targetSession ?? input.workdir ?? "-"}` + - `${input.provider ? ` provider=${input.provider}` : ""}${input.model ? ` model=${input.model}` : ""}`, + `${input.provider ? ` provider=${input.provider}` : ""}${input.model ? ` model=${input.model}` : ""}` + + `${isAttempt ? ` run=${input.runId} attempt=${input.attemptId}` : ""}`, ); this.#signalBoardChange(); return id; @@ -342,6 +415,62 @@ export class Dispatcher { return { groupId, taskIds }; } + /** + * Park a live task as `held` — waiting on something outside its worker (an + * approval outside a loop's permission profile, say). Reclaim never touches + * a held task, so the wait costs no attempt however long the person takes. + * + * `generation` is the one the caller was handed with the task (e.g. via + * `onAttemptWaiting`); a stale generation is refused, so a replaced worker's + * request cannot park its successor. The watch is dropped while held — a + * status change is not this dispatcher's to act on until `unhold`, which + * re-reads the worker's state rather than relying on a transition it missed. + */ + hold(taskId: string, generation: number, reason: string): boolean { + const held = this.#store.dispatchHold(taskId, generation, reason, Date.now()); + if (!held) { + this.#staleResult(taskId, generation, "hold"); + return false; + } + const task = this.#store.dispatchGet(taskId); + if (task?.workerSessionId) this.#unwatch(task.workerSessionId, taskId); + this.#host.audit( + "dispatch.held", + `task=${taskId} generation=${generation} reason=${reason.slice(0, 200)}`, + ); + this.#signalBoardChange(); + return true; + } + + /** + * Resume a held task as `running` under this boot, with a fresh lease, at the + * same generation — the same worker carries on, so its result still applies. + * + * The worker may have moved while held (or the daemon restarted), so its + * CURRENT status decides what happens next: still working → watch it; + * already settled → finish the task now; gone → leave it unwatched, and the + * lease reclaims it like any other dead worker. + */ + async unhold(taskId: string, generation: number): Promise { + const resumed = this.#store.dispatchUnhold(taskId, generation, this.bootId, Date.now()); + if (!resumed) { + this.#staleResult(taskId, generation, "unhold"); + return false; + } + this.#host.audit("dispatch.unheld", `task=${taskId} generation=${generation}`); + this.#signalBoardChange(); + const task = this.#store.dispatchGet(taskId); + const sessionId = task?.workerSessionId; + if (!sessionId) return true; + const status = this.#host.workerStatus(sessionId); + if (status === "idle" || status === "error") { + await this.#finishWorkerTask(taskId, generation, sessionId, status); + } else if (status) { + this.#watch(sessionId, taskId, generation); + } + return true; + } + /** * One dispatcher pass: reclaim stale claims, renew live leases, claim + * execute ready tasks, deliver pending conductor events. Public so tests @@ -386,14 +515,14 @@ export class Dispatcher { * worker's turn completion becomes a digest without polling. */ onSessionStatus(sessionId: string, status: SessionStatus): void { - const taskIds = this.tasksForSession(sessionId); - if (taskIds.length === 0) return; + const watchers = [...(this.#watched.get(sessionId) ?? [])]; + if (watchers.length === 0) return; if (status === "idle" || status === "error") { // Every task watching this session completes on the same transition — // dropping all but one is what left a panel member's task orphaned. - for (const taskId of taskIds) { + for (const [taskId, generation] of watchers) { this.#unwatch(sessionId, taskId); - void this.#finishWorkerTask(taskId, sessionId, status); + void this.#finishWorkerTask(taskId, generation, sessionId, status); } return; } @@ -403,9 +532,18 @@ export class Dispatcher { // and STOP renewing the lease; expiry reclaims (attempts++) and either // retries fresh or auto-blocks. The owner can also attach and approve // before the lease runs out — then the turn simply continues. - for (const taskId of taskIds) { + for (const [taskId, generation] of watchers) { const task = this.#store.dispatchGet(taskId); if (!task) continue; + // The wedge is a worker's report like any other: only the generation + // that worker was started under may raise it. + if (task.generation !== generation) continue; + if (task.kind === "attempt") { + // A loop decides what an approval means for its run — inside the + // profile, or `hold()` and wait for a person. Never a fleet event. + this.#callAttemptHook("onAttemptWaiting", () => this.#host.onAttemptWaiting?.(task)); + continue; + } this.#emitEvent(task, "task_failed", // type refined below if it recovers `worker for task ${task.id.slice(0, 8)} (${task.shape}) is WAITING FOR APPROVAL in session ${sessionId.slice(0, 8)} — its autonomous tool budget is exhausted or it hit a gated tool. Attach and approve to let it continue, or it will be reclaimed when the lease expires.`, { keepPending: true }, @@ -432,8 +570,17 @@ export class Dispatcher { if (task.workerSessionId) this.#unwatch(task.workerSessionId, task.id); this.#host.audit( "dispatch.reclaimed", - `task=${task.id} attempts=${task.attempts} status=${task.status}`, + `task=${task.id} attempts=${task.attempts} status=${task.status} generation=${task.generation}`, ); + if (task.kind === "attempt") { + // Terminal at its first reclaim (the store never re-queues one), and + // reported as `reclaimed` rather than as an ordinary failure: the + // worker may still be alive somewhere, which is exactly what a loop's + // restart reconciliation needs to know before it starts another. The + // worker is the host's — not destroyed here. + this.#settleAttempt(task, "reclaimed"); + continue; + } if (task.status === "blocked") { this.#emitEvent( task, @@ -455,12 +602,14 @@ export class Dispatcher { /** Renew leases only for workers that are verifiably alive AND working. */ #renewLiveLeases(): void { - const alive: string[] = []; - for (const [sessionId, taskIds] of this.#watched) { + const alive: Array<{ id: string; generation: number }> = []; + for (const [sessionId, watchers] of this.#watched) { const status = this.#host.workerStatus(sessionId); // Every task watching a live session renews — a shared target keeps all // of its dispatches leased, not just whichever registered first. - if (status && WORKER_ACTIVE.has(status)) alive.push(...taskIds); + if (status && WORKER_ACTIVE.has(status)) { + for (const [id, generation] of watchers) alive.push({ id, generation }); + } // idle/error are handled by onSessionStatus; waiting_approval and a // vanished session deliberately do NOT renew — the lease reclaims them. } @@ -487,7 +636,7 @@ export class Dispatcher { if (!task) return; touched.push(task.id); if ( - task.kind === "spawn" && + (task.kind === "spawn" || task.kind === "attempt") && this.#store.dispatchActiveSpawnCount(task.accountId, task.projectId) > this.#config.maxConcurrentWorkers ) { @@ -515,21 +664,27 @@ export class Dispatcher { return await this.#startGroupedSend(task, now); } await this.#host.sendToSession(task); - this.#store.dispatchComplete( + const applied = this.#store.dispatchComplete( task.id, + task.generation, `delivered to session ${task.targetSession?.slice(0, 8) ?? "?"}`, Date.now(), ); + if (!applied) return this.#staleResult(task.id, task.generation, "send"); this.#host.audit("dispatch.sent", `task=${task.id} target=${task.targetSession}`); return; } + if (task.kind === "attempt") return await this.#startAttempt(task); + // spawn: continue a surviving worker (post-restart) or create fresh. if (task.workerSessionId) { const continued = await this.#host.continueWorker(task); if (continued) { - this.#store.dispatchMarkRunning(task.id, task.workerSessionId, now); - this.#watch(task.workerSessionId, task.id); + if (!this.#store.dispatchMarkRunning(task.id, task.generation, task.workerSessionId, now)) { + return await this.#staleStart(task, task.workerSessionId); + } + this.#watch(task.workerSessionId, task.id, task.generation); this.#host.audit( "dispatch.continued", `task=${task.id} worker=${task.workerSessionId}`, @@ -539,21 +694,36 @@ export class Dispatcher { // Worker didn't survive the restart — fall through to a fresh spawn. } const { sessionId } = await this.#host.spawnWorker(task); - this.#store.dispatchMarkRunning(task.id, sessionId, Date.now()); - this.#watch(sessionId, task.id); + if (!this.#store.dispatchMarkRunning(task.id, task.generation, sessionId, Date.now())) { + return await this.#staleStart(task, sessionId); + } + this.#watch(sessionId, task.id, task.generation); this.#host.audit("dispatch.spawned", `task=${task.id} worker=${sessionId}`); } catch (err) { const message = err instanceof Error ? err.message : String(err); const retryable = !(err instanceof NonRetryableDispatchError); const now = Date.now(); - const status = this.#store.dispatchFail(task.id, message, now, { + const status = this.#store.dispatchFail(task.id, task.generation, message, now, { retryable, notBefore: retryable ? this.#retryAt(task.attempts, now) : undefined, }); + if (status === null) return this.#staleResult(task.id, task.generation, "execute"); + // Whatever this execution registered is now dead: a grouped send watches + // its target BEFORE delivering, so a delivery that threw left a watcher + // behind, and the target's next idle arrived as a result for a task that + // had already failed. A re-queued task re-watches when it runs again. + this.#unwatchTask(task.id); this.#host.audit( "dispatch.failed", `task=${task.id} retryable=${retryable} status=${status} error=${message.slice(0, 200)}`, ); + if (task.kind === "attempt") { + // Always terminal (the store never re-queues an attempt). No event and + // no teardown — both are the loop's. + const refreshed = this.#store.dispatchGet(task.id) ?? task; + this.#settleAttempt(refreshed, "failed"); + return; + } if (status === "failed" || status === "blocked") { const refreshed = this.#store.dispatchGet(task.id) ?? task; this.#emitEvent( @@ -575,7 +745,92 @@ export class Dispatcher { } } - /** Worker turn ended — digest, complete/fail, notify, tear down. */ + /** + * Start (or re-attach) the worker for a loop attempt. + * + * Fails CLOSED when the host has no loop support. Falling back to an ordinary + * spawn would run the attempt without the loop's permission profile or + * budgets, and with nobody listening for its result — a worker doing + * unbounded work for a run that can never hear back. `continueAttempt` is + * the one optional hook: without it, continuation is a fresh worker. + */ + async #startAttempt(task: DispatchTaskRow): Promise { + const host = this.#host; + if (!host.spawnAttemptWorker || !host.onAttemptSettled || !host.onAttemptWaiting) { + throw new NonRetryableDispatchError( + `attempt task ${task.id.slice(0, 8)} (run ${task.runId ?? "?"}) cannot run: this daemon has no loop host (${ATTEMPT_HOOKS})`, + ); + } + if (task.workerSessionId && host.continueAttempt && (await host.continueAttempt(task))) { + if (!this.#store.dispatchMarkRunning(task.id, task.generation, task.workerSessionId, Date.now())) { + return await this.#staleStart(task, task.workerSessionId); + } + this.#watch(task.workerSessionId, task.id, task.generation); + this.#host.audit( + "dispatch.attempt_continued", + `task=${task.id} run=${task.runId} attempt=${task.attemptId} worker=${task.workerSessionId}`, + ); + return; + } + const { sessionId } = await host.spawnAttemptWorker(task); + if (!this.#store.dispatchMarkRunning(task.id, task.generation, sessionId, Date.now())) { + return await this.#staleStart(task, sessionId); + } + this.#watch(sessionId, task.id, task.generation); + this.#host.audit( + "dispatch.attempt_started", + `task=${task.id} run=${task.runId} attempt=${task.attemptId} worker=${sessionId} generation=${task.generation}`, + ); + } + + /** + * A worker came up for a claim that was reclaimed while it was starting — + * the start outlived the lease. The successor owns the task now, so this + * worker must not be recorded or watched. A spawned worker is ours and is + * torn down; an attempt's belongs to the loop host, which the reclaim + * already told (`reclaimed`). + */ + async #staleStart(task: DispatchTaskRow, sessionId: string): Promise { + this.#staleResult(task.id, task.generation, `start worker=${sessionId}`); + if (task.kind === "spawn") { + await this.#host.destroyWorker(sessionId, `task ${task.id} claim was replaced while starting`); + } + } + + /** + * A result (or state change) from a generation that no longer owns the task — + * a replaced worker finishing late, or a task already terminal. Applying it + * would overwrite the successor's outcome or resurrect a settled task, so it + * is dropped; logged and audited so a fenced result is visible, not silent. + */ + #staleResult(taskId: string, generation: number, what: string): void { + const current = this.#store.dispatchGet(taskId); + const detail = + `task=${taskId} generation=${generation} current=${current?.generation ?? "-"} ` + + `status=${current?.status ?? "missing"} what=${what}`; + console.warn(`[codeoid/dispatch] ignored stale result: ${detail}`); + this.#host.audit("dispatch.stale_result", detail); + } + + /** Report an attempt's terminal state to the loop host. Never throws into the dispatcher. */ + #settleAttempt(task: DispatchTaskRow, status: AttemptSettledStatus): void { + this.#host.audit( + "dispatch.attempt_settled", + `task=${task.id} run=${task.runId} attempt=${task.attemptId} status=${status} generation=${task.generation}`, + ); + this.#callAttemptHook("onAttemptSettled", () => this.#host.onAttemptSettled?.(task, status)); + } + + #callAttemptHook(name: string, fn: () => void): void { + try { + fn(); + } catch (err) { + console.error( + `[codeoid/dispatch] ${name} failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + /** * Deliver (or re-attach to) a grouped send and watch its target to completion. * @@ -606,13 +861,20 @@ export class Dispatcher { `panel target ${target.slice(0, 8)} no longer exists`, ); } - this.#store.dispatchMarkRunning(task.id, target, now); + if (!this.#store.dispatchMarkRunning(task.id, task.generation, target, now)) { + return this.#staleResult(task.id, task.generation, "group_rewatch"); + } if (!WORKER_ACTIVE.has(status)) { // Settled while we were away — complete from what it left behind. - await this.#finishWorkerTask(task.id, target, status === "error" ? "error" : "idle"); + await this.#finishWorkerTask( + task.id, + task.generation, + target, + status === "error" ? "error" : "idle", + ); return; } - this.#watch(target, task.id); + this.#watch(target, task.id, task.generation); this.#host.audit( "dispatch.group_rewatched", `task=${task.id} group=${task.groupId} target=${target} status=${status}`, @@ -625,8 +887,10 @@ export class Dispatcher { // it after leaves a window where a crash between delivery and the marker // makes the retry re-send, and the reviewer works its brief twice while the // digest describes only the second pass. - this.#store.dispatchMarkRunning(task.id, target, now); - this.#watch(target, task.id); + if (!this.#store.dispatchMarkRunning(task.id, task.generation, target, now)) { + return this.#staleResult(task.id, task.generation, "group_send"); + } + this.#watch(target, task.id, task.generation); await this.#host.sendToSession(task); this.#host.audit( "dispatch.group_sent", @@ -634,19 +898,45 @@ export class Dispatcher { ); } + /** + * Worker turn ended — digest, complete/fail, notify, tear down. + * + * `generation` is the one the worker was started under. A task that has since + * been reclaimed (and possibly re-claimed by a successor) carries a newer + * one, and this worker's result is dropped rather than recorded over it. + */ async #finishWorkerTask( taskId: string, + generation: number, sessionId: string, status: "idle" | "error", ): Promise { const task = this.#store.dispatchGet(taskId); - if (!task || (task.status !== "running" && task.status !== "claimed")) return; + if (!task) return; + if ( + task.generation !== generation || + (task.status !== "running" && task.status !== "claimed") + ) { + this.#staleResult(taskId, generation, `worker=${sessionId} turn=${status}`); + return; + } try { // Digest BEFORE teardown — it reads the live session + memory. const digest = this.#host.buildWorkerDigest(task); if (status === "idle") { - this.#store.dispatchComplete(task.id, digest, Date.now()); + // The read above and this write are not one transaction; the store's + // generation + status guard is what actually fences the result. + if (!this.#store.dispatchComplete(task.id, generation, digest, Date.now())) { + this.#staleResult(taskId, generation, `worker=${sessionId} turn=idle`); + return; + } + if (task.kind === "attempt") { + // The worker stays: an attempt's session is the loop's to keep for + // verification, inspection, or the next attempt's continuation. + this.#settleAttempt(this.#store.dispatchGet(task.id) ?? task, "done"); + return; + } this.#emitEvent(task, "task_done", digest); this.#host.audit("dispatch.done", `task=${task.id} worker=${sessionId}`); // Disposable children (design R2): the work products live in the @@ -665,10 +955,20 @@ export class Dispatcher { const failNow = Date.now(); const failStatus = this.#store.dispatchFail( task.id, + generation, "worker turn ended in error", failNow, { retryable: true, notBefore: this.#retryAt(task.attempts, failNow) }, ); + if (failStatus === null) { + this.#staleResult(taskId, generation, `worker=${sessionId} turn=error`); + return; + } + if (task.kind === "attempt") { + // Terminal `failed` — the store never re-queues an attempt. + this.#settleAttempt(this.#store.dispatchGet(task.id) ?? task, "failed"); + return; + } if (failStatus === "blocked") { this.#emitEvent( task, @@ -799,6 +1099,10 @@ ${body ?? "no digest"}`; digest: string, opts?: { keepPending?: boolean }, ): void { + // A loop attempt never wakes a conductor (§12.6): its outcome goes to the + // loop host. Every attempt path settles before reaching here; this is the + // backstop so a future path cannot start a model turn by accident. + if (task.kind === "attempt") return; // A grouped task's completion is the barrier's business, not its own. if (this.#barrierAbsorb(task)) return; this.#store.dispatchEventAdd({ diff --git a/src/daemon/fleet.ts b/src/daemon/fleet.ts index b8a30ff7..a6622952 100644 --- a/src/daemon/fleet.ts +++ b/src/daemon/fleet.ts @@ -51,7 +51,7 @@ export interface FleetSessionView { /** A dispatch task as the conductor sees it on the fleet_tasks board. */ export interface FleetTaskView { id: string; - kind: "send" | "spawn"; + kind: "send" | "spawn" | "attempt"; shape: "ship" | "scout"; status: string; attempts: number; diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index 305ceaac..ce6e462d 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -305,6 +305,8 @@ function toFleetTaskWire(row: DispatchTaskRow): FleetTaskWire { error: row.error ?? undefined, createdBy: row.createdBy, groupId: row.groupId ?? undefined, + runId: row.runId ?? undefined, + attemptId: row.attemptId ?? undefined, }; } @@ -3529,7 +3531,11 @@ mcpHub: this.#mcpHub, panel.members.push({ sessionId: t.targetSession, ordinal: t.groupOrdinal ?? panel.members.length + 1, - status: t.status, + // A held member is unsettled with its worker still assigned — what a + // panel draws as `running`. Nothing holds a panel member today (only + // loop attempts are held, and attempts are never grouped), so the + // panel's wire union stays as clients already render it. + status: t.status === "held" ? "running" : t.status, }); if (TERMINAL_TASK_STATUS.has(t.status)) panel.settled++; } diff --git a/src/daemon/store.ts b/src/daemon/store.ts index 3db97ed8..f1bcd1cf 100644 --- a/src/daemon/store.ts +++ b/src/daemon/store.ts @@ -14,15 +14,30 @@ export type DispatchTaskStatus = | "queued" | "claimed" | "running" + /** + * Parked on something outside the worker — an approval outside a loop's + * permission profile, say. Non-terminal, and deliberately invisible to + * reclaim: a held task is not a dead claim, it is one waiting on a person, + * and reclaiming it would burn the attempt for the human's response time. + */ + | "held" | "done" | "failed" | "blocked"; +/** + * `attempt` is one bounded attempt of a loop run (docs/loop-engineering-design.md + * §5). It rides the same queue as send/spawn — the dispatcher stays the sole + * scheduler of worker attempts — but its lifecycle belongs to the loop host: + * no fleet events, no teardown on finish, and a reclaim is terminal. + */ +export type DispatchTaskKind = "send" | "spawn" | "attempt"; + export interface DispatchTaskRow { id: string; accountId: string; projectId: string; - kind: "send" | "spawn"; + kind: DispatchTaskKind; shape: "ship" | "scout"; targetSession: string | null; workdir: string | null; @@ -59,6 +74,18 @@ export interface DispatchTaskRow { * "member 1 of 3" labels in a joined digest would shuffle between reads. */ groupOrdinal: number | null; + /** Loop run this attempt belongs to. NULL for every non-attempt task. */ + runId: string | null; + /** The loop's own attempt record. NULL for every non-attempt task. */ + attemptId: string | null; + /** + * Claim generation — bumped on every claim and every reclaim. A worker's + * result is accepted only for the generation it was started under, so a + * worker replaced after a reclaim cannot complete or fail the task its + * successor now owns (§6 "Restart guarantees"). 0 = never claimed, which is + * what every row written before generations existed reads as. + */ + generation: number; createdAt: number; updatedAt: number; } @@ -82,7 +109,7 @@ interface RawDispatchRow { id: string; account_id: string; project_id: string; - kind: "send" | "spawn"; + kind: DispatchTaskKind; shape: "ship" | "scout"; target_session: string | null; workdir: string | null; @@ -102,6 +129,9 @@ interface RawDispatchRow { created_by: string; group_id: string | null; group_ordinal: number | null; + run_id: string | null; + attempt_id: string | null; + generation: number; created_at: number; updated_at: number; } @@ -132,6 +162,9 @@ function rowToDispatchTask(r: RawDispatchRow): DispatchTaskRow { createdBy: r.created_by, groupId: r.group_id ?? null, groupOrdinal: r.group_ordinal ?? null, + runId: r.run_id ?? null, + attemptId: r.attempt_id ?? null, + generation: r.generation ?? 0, createdAt: r.created_at, updatedAt: r.updated_at, }; @@ -452,14 +485,14 @@ export class Store { id TEXT PRIMARY KEY, account_id TEXT NOT NULL, project_id TEXT NOT NULL, - kind TEXT NOT NULL, -- 'send' | 'spawn' + kind TEXT NOT NULL, -- 'send' | 'spawn' | 'attempt' shape TEXT NOT NULL, -- 'ship' | 'scout' target_session TEXT, -- send: existing session id workdir TEXT, -- spawn: worker workdir prompt TEXT NOT NULL, provider TEXT, -- spawn: per-child backend id; NULL = daemon default model TEXT, -- spawn: per-child model, resolved against the provider above - status TEXT NOT NULL DEFAULT 'queued', + status TEXT NOT NULL DEFAULT 'queued', -- queued|claimed|running|held|done|failed|blocked attempts INTEGER NOT NULL DEFAULT 0, failure_limit INTEGER NOT NULL DEFAULT 2, claim_owner TEXT, -- daemon boot id @@ -471,6 +504,9 @@ export class Store { created_by TEXT NOT NULL, -- conductor WIMSE URI / sub group_id TEXT, -- dispatch group (barrier); NULL = a standalone task group_ordinal INTEGER, -- 1-based position within the group; NULL when ungrouped + run_id TEXT, -- attempt: loop run id; NULL otherwise + attempt_id TEXT, -- attempt: loop attempt id; NULL otherwise + generation INTEGER NOT NULL DEFAULT 0, -- claim generation; fences stale worker results created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); @@ -575,6 +611,14 @@ export class Store { `CREATE INDEX IF NOT EXISTS idx_dispatch_owner ON dispatch_tasks(owner_daemon, status, created_at);`, ); + // Loop attempts (docs/loop-engineering-design.md §5) and claim generations + // (§6). NULL run/attempt ids = a send or spawn task, which is every row + // written before loops existed. generation 0 = "never claimed under + // generations": the first claim after the upgrade bumps it to 1, so a + // worker started by the old binary is fenced like any other stale one. + this.#addColumnIfMissing("dispatch_tasks", "run_id", "TEXT"); + this.#addColumnIfMissing("dispatch_tasks", "attempt_id", "TEXT"); + this.#addColumnIfMissing("dispatch_tasks", "generation", "INTEGER NOT NULL DEFAULT 0"); // AFTER the ALTERs, never inside the CREATE block above. On an existing // database `CREATE TABLE IF NOT EXISTS` is a no-op, so an index declared // there would run against a table that does not have the column yet and @@ -595,6 +639,10 @@ export class Store { 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; + -- A loop reconciles its attempts by run on restart. Partial: only attempt + -- rows carry a run id, so send/spawn history never bloats it. + CREATE INDEX IF NOT EXISTS idx_dispatch_run + ON dispatch_tasks(run_id) WHERE run_id IS NOT NULL; `); // Pre-release single-row predecessor of provider_model_catalogs — never @@ -1022,11 +1070,14 @@ export class Store { id: string; accountId: string; projectId: string; - kind: "send" | "spawn"; + kind: DispatchTaskKind; shape: "ship" | "scout"; targetSession?: string; workdir?: string; prompt: string; + /** Loop run + attempt ids — required for `kind: "attempt"`, absent otherwise. */ + runId?: string; + attemptId?: string; /** Per-child backend for a spawn task; omit for the daemon default. */ provider?: string; /** Per-child model, already resolved against `provider`. */ @@ -1050,8 +1101,8 @@ export class Store { `INSERT INTO dispatch_tasks (id, account_id, project_id, kind, shape, target_session, workdir, prompt, provider, model, failure_limit, created_by, group_id, - group_ordinal, owner_daemon, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + group_ordinal, owner_daemon, run_id, attempt_id, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( task.id, @@ -1069,6 +1120,8 @@ export class Store { task.groupId ?? null, task.groupOrdinal ?? null, task.ownerDaemon ?? null, + task.runId ?? null, + task.attemptId ?? null, task.now, task.now, ); @@ -1082,6 +1135,10 @@ export class Store { * failed retries) WITHOUT global head-of-line blocking: only the specific * deferred tasks are skipped, so neither sends nor other tenants' spawns * are starved by one capped tenant. + * + * Every claim bumps `generation`; the returned row carries the new value, + * which is the one the worker started from this claim must present when it + * completes or fails the task. */ dispatchClaimNext( bootId: string, @@ -1092,7 +1149,8 @@ export class Store { const row = this.#db .prepare( `UPDATE dispatch_tasks - SET status = 'claimed', claim_owner = ?, claimed_at = ?, updated_at = ? + SET status = 'claimed', claim_owner = ?, claimed_at = ?, updated_at = ?, + generation = generation + 1 WHERE id = ( SELECT id FROM dispatch_tasks WHERE status = 'queued' @@ -1114,46 +1172,81 @@ export class Store { return row ? rowToDispatchTask(row) : null; } - /** Transition a claimed task to running (worker session known, if spawn). */ - dispatchMarkRunning(id: string, workerSessionId: string | null, now: number): void { - this.#db + /** + * Transition a claimed task to running (worker session known, if spawn). + * + * Fenced like a result: false when the claim it was started under has been + * reclaimed in the meantime (a spawn that outlived the lease). Recording a + * worker on a claim that is no longer ours would hand the successor a + * session it never started. + */ + dispatchMarkRunning( + id: string, + generation: number, + workerSessionId: string | null, + now: number, + ): boolean { + const res = this.#db .prepare( `UPDATE dispatch_tasks SET status = 'running', worker_session_id = COALESCE(?, worker_session_id), updated_at = ? - WHERE id = ?`, + WHERE id = ? AND generation = ? AND status IN ('claimed', 'running')`, ) - .run(workerSessionId, now, id); + .run(workerSessionId, now, id, generation); + return res.changes > 0; } - /** Renew the lease on tasks whose workers are verifiably still alive. */ - dispatchTouch(ids: string[], now: number): void { - if (ids.length === 0) return; + /** + * Renew the lease on tasks whose workers are verifiably still alive — only + * for the generation each was watched under, so a watcher left over from a + * replaced worker cannot keep its successor's claim artificially fresh. + */ + dispatchTouch( + entries: ReadonlyArray<{ id: string; generation: number }>, + now: number, + ): void { + if (entries.length === 0) return; const stmt = this.#db.prepare( - "UPDATE dispatch_tasks SET claimed_at = ? WHERE id = ?", + "UPDATE dispatch_tasks SET claimed_at = ? WHERE id = ? AND generation = ?", ); // One transaction: the batch is atomic (all-or-nothing) and commits once. this.#db.transaction(() => { - for (const id of ids) stmt.run(now, id); + for (const e of entries) stmt.run(now, e.id, e.generation); })(); } - dispatchComplete(id: string, digest: string, now: number): void { - this.#db + /** + * Record a worker's success. Returns whether it applied. + * + * Guarded on the generation the worker was started under AND on a live + * claim. Without the guard a worker replaced after a reclaim — still + * running, finishing late — overwrote whatever its successor had recorded, + * or resurrected a task that had already gone terminal. + */ + dispatchComplete(id: string, generation: number, digest: string, now: number): boolean { + const res = this.#db .prepare( `UPDATE dispatch_tasks SET status = 'done', result_digest = ?, error = NULL, updated_at = ? - WHERE id = ?`, + WHERE id = ? AND generation = ? AND status IN ('claimed', 'running')`, ) - .run(digest, now, id); + .run(digest, now, id, generation); + return res.changes > 0; } /** * Record a failure. Retryable failures re-queue until `failure_limit` * attempts, then auto-block (hermes anti-spin); non-retryable ones (target - * gone, invalid input) go terminal immediately. Returns the final status. + * gone, invalid input) go terminal immediately. Returns the final status, or + * null when the update did not apply — the generation is stale or the claim + * is no longer live (same fence as `dispatchComplete`). + * + * A loop attempt never re-queues: retrying is the loop's decision (a new + * attempt with its own record), so any failure of one is terminal `failed`. */ dispatchFail( id: string, + generation: number, error: string, now: number, opts: { retryable: boolean; notBefore?: number }, @@ -1163,24 +1256,64 @@ export class Store { opts.retryable ? `UPDATE dispatch_tasks SET attempts = attempts + 1, - status = CASE WHEN attempts + 1 >= failure_limit THEN 'blocked' ELSE 'queued' END, + status = CASE + WHEN kind = 'attempt' THEN 'failed' + WHEN attempts + 1 >= failure_limit THEN 'blocked' + ELSE 'queued' END, claim_owner = NULL, claimed_at = NULL, not_before = ?, error = ?, updated_at = ? - WHERE id = ? + WHERE id = ? AND generation = ? AND status IN ('claimed', 'running') RETURNING status` : `UPDATE dispatch_tasks SET attempts = attempts + 1, status = 'failed', claim_owner = NULL, claimed_at = NULL, error = ?, updated_at = ? - WHERE id = ? + WHERE id = ? AND generation = ? AND status IN ('claimed', 'running') RETURNING status`, ) .get( ...(opts.retryable - ? [opts.notBefore ?? null, error, now, id] - : [error, now, id]), + ? [opts.notBefore ?? null, error, now, id, generation] + : [error, now, id, generation]), ) as { status: "queued" | "blocked" | "failed" } | null; return row?.status ?? null; } + /** + * Park a live task on something outside its worker (`held`). Fenced on the + * generation like a result, since it is one: a stale worker's approval + * request must not park the claim its successor holds. Returns whether it + * applied. The claim owner and worker are kept — the task is still this + * daemon's, just not progressing — and reclaim never selects a held row. + */ + dispatchHold(id: string, generation: number, reason: string, now: number): boolean { + const res = this.#db + .prepare( + `UPDATE dispatch_tasks + SET status = 'held', error = ?, updated_at = ? + WHERE id = ? AND generation = ? AND status IN ('claimed', 'running')`, + ) + .run(reason, now, id, generation); + return res.changes > 0; + } + + /** + * Resume a held task as `running` under THIS boot with a fresh lease. + * + * Re-owned by the caller's boot id because the hold may have outlived the + * daemon that placed it; keeping the old boot id would have the very next + * tick reclaim the task as a crashed claim. The generation is unchanged — + * the same worker carries on — so its eventual result still applies. + */ + dispatchUnhold(id: string, generation: number, bootId: string, now: number): boolean { + const res = this.#db + .prepare( + `UPDATE dispatch_tasks + SET status = 'running', claim_owner = ?, claimed_at = ?, error = NULL, updated_at = ? + WHERE id = ? AND generation = ? AND status = 'held'`, + ) + .run(bootId, now, now, id, generation); + return res.changes > 0; + } + /** * Reclaim tasks whose claim is dead: held by another boot (daemon crashed) * or past the lease (worker hung and the dispatcher stopped renewing). @@ -1188,6 +1321,12 @@ export class Store { * restarts burns through failure_limit and lands in 'blocked': the * stuck-loop escalation, in queue form. worker_session_id is preserved so * a re-claimed spawn can continue its (resumed) worker session. + * + * A reclaim bumps `generation`, fencing the replaced worker the moment its + * claim is declared dead rather than only at the next claim. `held` rows are + * never selected — they are parked, not dead. A loop attempt is never + * re-queued: a reclaimed attempt goes terminal `failed`, and the loop decides + * whether to start another. */ /** * Reclaim claims this daemon can prove are dead. @@ -1208,8 +1347,11 @@ export class Store { .prepare( `UPDATE dispatch_tasks SET attempts = attempts + 1, - status = CASE WHEN attempts + 1 >= failure_limit THEN 'blocked' ELSE 'queued' END, - claim_owner = NULL, claimed_at = NULL, + status = CASE + WHEN kind = 'attempt' THEN 'failed' + WHEN attempts + 1 >= failure_limit THEN 'blocked' + ELSE 'queued' END, + claim_owner = NULL, claimed_at = NULL, generation = generation + 1, error = COALESCE(error, 'reclaimed: stale claim'), updated_at = ? WHERE status IN ('claimed', 'running') AND ( @@ -1411,7 +1553,7 @@ export class Store { const row = this.#db .prepare( `SELECT - SUM(CASE WHEN status IN ('queued','claimed','running') THEN 1 ELSE 0 END) AS active, + SUM(CASE WHEN status IN ('queued','claimed','running','held') THEN 1 ELSE 0 END) AS active, SUM(CASE WHEN status = 'blocked' THEN 1 ELSE 0 END) AS blocked FROM dispatch_tasks WHERE account_id = ? AND project_id = ?`, @@ -1465,13 +1607,16 @@ export class Store { .all(accountId, projectId, sinceId) as DispatchEventRow[]; } - /** Count of live (claimed/running) spawn tasks — the concurrency-cap read. */ + /** + * Count of live worker tasks — the concurrency-cap read. Loop attempts are + * workers too, and a held one still holds its session, so both count. + */ dispatchActiveSpawnCount(accountId: string, projectId: string): number { const row = this.#db .prepare( `SELECT COUNT(*) AS n FROM dispatch_tasks WHERE account_id = ? AND project_id = ? - AND kind = 'spawn' AND status IN ('claimed', 'running')`, + AND kind IN ('spawn', 'attempt') AND status IN ('claimed', 'running', 'held')`, ) .get(accountId, projectId) as { n: number }; return row.n; diff --git a/src/tests/collaboration.test.ts b/src/tests/collaboration.test.ts index 47deca01..9e0ff239 100644 --- a/src/tests/collaboration.test.ts +++ b/src/tests/collaboration.test.ts @@ -437,6 +437,21 @@ const textTurn = (text: string): ProviderEvent[] => [ { type: "turn_done", result: mockResult() } as ProviderEvent, ]; +/** + * Force a dispatch task terminal, bypassing the claim lifecycle. These tests + * exercise what is drawn or delivered once a task has settled, not how it got + * there — and `dispatchComplete` / `dispatchFail` now refuse a row that was + * never claimed (the generation fence), which is exactly right for a worker + * result and exactly wrong for staging one. + */ +function settleTask(id: string, status: "done" | "failed", digest: string): void { + store.database + .prepare( + "UPDATE dispatch_tasks SET status = ?, result_digest = ?, error = ?, updated_at = ? WHERE id = ?", + ) + .run(status, status === "done" ? digest : null, status === "done" ? null : digest, Date.now(), id); +} + function mkConfig(over: Partial = {}): CodeoidConfig { return { daemonUrl: "ws://127.0.0.1:7400", @@ -3169,13 +3184,13 @@ describe("collaboration.panels", () => { }); const members = store.dispatchGroupMembers(AUTH.accountId, AUTH.projectId, groupId); - store.dispatchComplete(members[0]!.id, "one", Date.now()); + settleTask(members[0]!.id, "done", "one"); let p = (await panelsOf(goal.id)).panels[0]!; expect(p.settled).toBe(1); expect(p.joined).toBe(false); // 1/3 — a UI shows progress, not completion - store.dispatchComplete(members[1]!.id, "two", Date.now()); - store.dispatchComplete(members[2]!.id, "three", Date.now()); + settleTask(members[1]!.id, "done", "two"); + settleTask(members[2]!.id, "done", "three"); p = (await panelsOf(goal.id)).panels[0]!; expect(p.settled).toBe(3); expect(p.joined).toBe(true); @@ -3193,10 +3208,10 @@ describe("collaboration.panels", () => { targets: kids.map((k) => k.id), prompt: "review", shape: "scout", }); const members = store.dispatchGroupMembers(AUTH.accountId, AUTH.projectId, groupId); - store.dispatchComplete(members[0]!.id, "ok", Date.now()); - store.dispatchComplete(members[1]!.id, "ok", Date.now()); + settleTask(members[0]!.id, "done", "ok"); + settleTask(members[1]!.id, "done", "ok"); // Burn the third to blocked. - for (let i = 0; i < 3; i++) store.dispatchFail(members[2]!.id, "nope", Date.now(), { retryable: false }); + settleTask(members[2]!.id, "failed", "nope"); const p = (await panelsOf(goal.id)).panels[0]!; expect(p.members.some((m) => m.status === "failed" || m.status === "blocked")).toBe(true); diff --git a/src/tests/dispatch-generations.test.ts b/src/tests/dispatch-generations.test.ts new file mode 100644 index 00000000..19a0e565 --- /dev/null +++ b/src/tests/dispatch-generations.test.ts @@ -0,0 +1,589 @@ +/** + * Claim generations, `held` tasks, and loop-attempt routing on the dispatch + * queue (docs/loop-engineering-design.md §5, §6 "Restart guarantees"). + * + * The bug these fence: `dispatchComplete` updated by id alone, so a worker + * replaced after a reclaim — still alive, finishing late — overwrote the + * outcome its successor recorded, or resurrected a task that had already gone + * terminal. Every worker result now presents the generation it was started + * under, and only that generation of a live claim applies. + */ + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { Database } from "bun:sqlite"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + type AttemptSettledStatus, + Dispatcher, + type DispatcherHost, +} from "../daemon/dispatch.js"; +import { type DispatchEventRow, type DispatchTaskRow, Store } from "../daemon/store.js"; +import type { SessionStatus } from "../protocol/types.js"; + +const TENANT = { accountId: "acc-a", projectId: "proj-a" }; +const LEASE_MS = 60_000; + +let tmp: string; +let store: Store; + +beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-dispatch-gen-")); + store = new Store(join(tmp, "codeoid.db")); +}); + +afterEach(() => { + store.close(); + rmSync(tmp, { recursive: true, force: true }); +}); + +let seq = 0; +function enqueueRow(overrides: Partial[0]> = {}): string { + seq += 1; + const id = overrides.id ?? `task-${seq}`; + store.dispatchEnqueue({ + id, + ...TENANT, + kind: "spawn", + shape: "scout", + workdir: "/tmp/w", + prompt: "do the thing", + failureLimit: 3, + createdBy: "wimse://test/conductor", + now: seq, + ...overrides, + }); + return id; +} + +/** Age a task's lease so the next reclaim treats it as dead. */ +function expireLease(id: string): void { + store.database.prepare("UPDATE dispatch_tasks SET claimed_at = 1 WHERE id = ?").run(id); +} + +// ── Store ───────────────────────────────────────────────────────────────── + +describe("dispatch store — claim generations", () => { + test("every claim and every reclaim bumps the generation", () => { + const id = enqueueRow(); + expect(store.dispatchGet(id)!.generation).toBe(0); + expect(store.dispatchClaimNext("boot-1", 100)!.generation).toBe(1); + // A reclaim fences the dead worker at once, not only at the next claim. + expect(store.dispatchReclaimStale("boot-2", LEASE_MS, 200)[0]!.generation).toBe(2); + expect(store.dispatchClaimNext("boot-2", 300)!.generation).toBe(3); + }); + + test("a stale worker's completion after reclaim + re-claim is a no-op; the successor's applies", () => { + const id = enqueueRow(); + const old = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchMarkRunning(id, old.generation, "worker-1", 100)).toBe(true); + store.dispatchReclaimStale("boot-2", LEASE_MS, 200); + const fresh = store.dispatchClaimNext("boot-2", 300)!; + expect(store.dispatchMarkRunning(id, fresh.generation, "worker-2", 300)).toBe(true); + + // The replaced worker finishes late — every result path refuses it. + expect(store.dispatchComplete(id, old.generation, "stale digest", 400)).toBe(false); + expect(store.dispatchFail(id, old.generation, "stale error", 400, { retryable: true })).toBeNull(); + expect(store.dispatchMarkRunning(id, old.generation, "worker-1", 400)).toBe(false); + expect(store.dispatchHold(id, old.generation, "stale approval", 400)).toBe(false); + let row = store.dispatchGet(id)!; + expect(row.status).toBe("running"); + expect(row.workerSessionId).toBe("worker-2"); + expect(row.resultDigest).toBeNull(); + + expect(store.dispatchComplete(id, fresh.generation, "real digest", 500)).toBe(true); + row = store.dispatchGet(id)!; + expect(row.status).toBe("done"); + expect(row.resultDigest).toBe("real digest"); + }); + + test("a terminal task cannot be completed or failed again, even at its own generation", () => { + const id = enqueueRow(); + const claim = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchFail(id, claim.generation, "gone", 101, { retryable: false })).toBe("failed"); + // Previously this resurrected a failed task as `done`. + expect(store.dispatchComplete(id, claim.generation, "late", 102)).toBe(false); + expect(store.dispatchGet(id)!.status).toBe("failed"); + }); + + test("a stale lease renewal does not keep the successor's claim fresh", () => { + const id = enqueueRow(); + const old = store.dispatchClaimNext("boot-1", 1_000)!; + store.dispatchReclaimStale("boot-2", LEASE_MS, 2_000); + store.dispatchClaimNext("boot-2", 3_000); + store.dispatchTouch([{ id, generation: old.generation }], 50_000); + expect(store.dispatchGet(id)!.claimedAt).toBe(3_000); + }); +}); + +describe("dispatch store — held tasks", () => { + test("a held task is never reclaimed, by boot mismatch or by lease expiry", () => { + const id = enqueueRow(); + const claim = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchHold(id, claim.generation, "waiting on approval", 150)).toBe(true); + expect(store.dispatchReclaimStale("boot-2", LEASE_MS, 10_000_000)).toEqual([]); + const row = store.dispatchGet(id)!; + expect(row.status).toBe("held"); + expect(row.attempts).toBe(0); + expect(row.generation).toBe(claim.generation); + }); + + test("unhold resumes under the caller's boot with a fresh lease and the same generation", () => { + const id = enqueueRow(); + const claim = store.dispatchClaimNext("boot-OLD", 100)!; + store.dispatchHold(id, claim.generation, "approval", 150); + expect(store.dispatchUnhold(id, claim.generation + 1, "boot-NEW", 5_000)).toBe(false); + expect(store.dispatchUnhold(id, claim.generation, "boot-NEW", 5_000)).toBe(true); + const row = store.dispatchGet(id)!; + expect(row.status).toBe("running"); + expect(row.claimOwner).toBe("boot-NEW"); + expect(row.claimedAt).toBe(5_000); + expect(row.generation).toBe(claim.generation); + // Re-owned, so the new boot's fast path does not take it for a crash. + expect(store.dispatchReclaimStale("boot-NEW", LEASE_MS, 6_000)).toEqual([]); + // And the worker's result, at the unchanged generation, still applies. + expect(store.dispatchComplete(id, claim.generation, "ok", 7_000)).toBe(true); + }); + + test("only a live claim can be held, and a held task is still active and still a worker", () => { + const queued = enqueueRow(); + expect(store.dispatchHold(queued, 0, "nope", 100)).toBe(false); + const claim = store.dispatchClaimNext("boot-1", 100)!; + store.dispatchHold(claim.id, claim.generation, "approval", 101); + expect(store.dispatchStatusCounts(TENANT.accountId, TENANT.projectId).active).toBe(1); + expect(store.dispatchActiveSpawnCount(TENANT.accountId, TENANT.projectId)).toBe(1); + }); +}); + +describe("dispatch store — attempt tasks", () => { + test("run and attempt ids round-trip", () => { + const id = enqueueRow({ kind: "attempt", runId: "run-1", attemptId: "att-1", failureLimit: 1 }); + const row = store.dispatchGet(id)!; + expect(row.kind).toBe("attempt"); + expect(row.runId).toBe("run-1"); + expect(row.attemptId).toBe("att-1"); + }); + + test("a reclaimed attempt goes terminal `failed` instead of re-queueing, whatever its limit", () => { + // failureLimit 3 on purpose: the store's rule, not the row's budget, is what + // stops the queue retrying an attempt behind the loop's back. + const id = enqueueRow({ kind: "attempt", runId: "r", attemptId: "a" }); + store.dispatchClaimNext("boot-1", 100); + const [reclaimed] = store.dispatchReclaimStale("boot-2", LEASE_MS, 200); + expect(reclaimed!.status).toBe("failed"); + expect(store.dispatchClaimNext("boot-2", 300)).toBeNull(); + expect(store.dispatchGet(id)!.error).toContain("reclaimed"); + }); + + test("a retryable failure of an attempt is terminal too", () => { + const id = enqueueRow({ kind: "attempt", runId: "r", attemptId: "a" }); + const claim = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchFail(id, claim.generation, "x", 101, { retryable: true })).toBe("failed"); + }); +}); + +describe("migration — opening a database written before generations", () => { + function downgrade(dbPath: string): void { + const db = new Database(dbPath); + // The run index references run_id; SQLite refuses to drop an indexed column. + db.exec("DROP INDEX IF EXISTS idx_dispatch_run"); + db.exec("ALTER TABLE dispatch_tasks DROP COLUMN generation"); + db.exec("ALTER TABLE dispatch_tasks DROP COLUMN attempt_id"); + db.exec("ALTER TABLE dispatch_tasks DROP COLUMN run_id"); + db.close(); + } + + const columns = (dbPath: string): string[] => { + const db = new Database(dbPath, { readonly: true }); + const cols = ( + db.prepare("PRAGMA table_info(dispatch_tasks)").all() as Array<{ name: string }> + ).map((c) => c.name); + db.close(); + return cols; + }; + + test("adds the columns in place; existing rows read as generation 0 and fence normally", () => { + const dbPath = join(tmp, "upgrade.db"); + const old = new Store(dbPath); + old.dispatchEnqueue({ + id: "pre-upgrade", + ...TENANT, + kind: "spawn", + shape: "scout", + workdir: "/tmp/w", + prompt: "queued before generations", + failureLimit: 2, + createdBy: "conductor", + now: 1, + }); + // A claim in flight when the old binary went down. + old.dispatchClaimNext("boot-OLD", 2); + old.close(); + downgrade(dbPath); + expect(columns(dbPath)).not.toContain("generation"); + + const upgraded = new Store(dbPath); + try { + expect(columns(dbPath)).toEqual( + expect.arrayContaining(["run_id", "attempt_id", "generation"]), + ); + const row = upgraded.dispatchGet("pre-upgrade")!; + expect(row.generation).toBe(0); + expect(row.runId).toBeNull(); + expect(row.attemptId).toBeNull(); + expect(row.status).toBe("claimed"); + + // The old binary's worker presents generation 0; once the new boot + // reclaims and re-claims, that result no longer applies. + upgraded.dispatchReclaimStale("boot-NEW", LEASE_MS, 3); + const fresh = upgraded.dispatchClaimNext("boot-NEW", 4)!; + expect(fresh.generation).toBe(2); + expect(upgraded.dispatchComplete("pre-upgrade", 0, "stale", 5)).toBe(false); + expect(upgraded.dispatchComplete("pre-upgrade", fresh.generation, "ok", 6)).toBe(true); + } finally { + upgraded.close(); + } + }); +}); + +// ── Dispatcher ──────────────────────────────────────────────────────────── + +/** Minimal scriptable host — no attempt hooks (a daemon without a loop service). */ +class FakeHost implements DispatcherHost { + spawned: DispatchTaskRow[] = []; + destroyed: string[] = []; + delivered: DispatchEventRow[][] = []; + audits: string[] = []; + statuses = new Map(); + continueSucceeds = true; + #next = 0; + + async sendToSession(): Promise {} + + async spawnWorker(task: DispatchTaskRow): Promise<{ sessionId: string }> { + this.spawned.push(task); + const sessionId = `worker-${++this.#next}`; + this.statuses.set(sessionId, "thinking"); + return { sessionId }; + } + + async continueWorker(task: DispatchTaskRow): Promise { + if (this.continueSucceeds && task.workerSessionId) { + this.statuses.set(task.workerSessionId, "thinking"); + return true; + } + return false; + } + + workerStatus(sessionId: string): SessionStatus | null { + return this.statuses.get(sessionId) ?? null; + } + + buildWorkerDigest(task: DispatchTaskRow): string { + return `digest for ${task.id}`; + } + + async destroyWorker(sessionId: string): Promise { + this.destroyed.push(sessionId); + this.statuses.delete(sessionId); + } + + async deliverEvents( + _a: string, + _p: string, + events: DispatchEventRow[], + ): Promise { + this.delivered.push(events); + return events.map((e) => e.id); + } + + audit(action: string, detail: string): void { + this.audits.push(`${action} ${detail}`); + } +} + +/** A host with a loop service behind it. */ +class LoopHost extends FakeHost { + attemptSpawned: DispatchTaskRow[] = []; + settled: Array<{ task: DispatchTaskRow; status: AttemptSettledStatus }> = []; + waiting: DispatchTaskRow[] = []; + #nextAttempt = 0; + + async spawnAttemptWorker(task: DispatchTaskRow): Promise<{ sessionId: string }> { + this.attemptSpawned.push(task); + const sessionId = `attempt-worker-${++this.#nextAttempt}`; + this.statuses.set(sessionId, "thinking"); + return { sessionId }; + } + + onAttemptSettled(task: DispatchTaskRow, status: AttemptSettledStatus): void { + this.settled.push({ task, status }); + } + + onAttemptWaiting(task: DispatchTaskRow): void { + this.waiting.push(task); + } +} + +function makeDispatcher(host: DispatcherHost): Dispatcher { + return new Dispatcher(store, host, { + tickMs: 999_999, + leaseMs: LEASE_MS, + failureLimit: 3, + maxConcurrentWorkers: 4, + workerToolBudget: 10, + retryBaseMs: 0, + }); +} + +function enqueueSpawn(d: Dispatcher): string { + return d.enqueue({ + ...TENANT, + kind: "spawn", + shape: "scout", + workdir: "/tmp/w", + prompt: "investigate", + createdBy: "wimse://test/conductor", + }); +} + +function enqueueAttempt(d: Dispatcher): string { + return d.enqueue({ + ...TENANT, + kind: "attempt", + shape: "ship", + workdir: "/tmp/w", + prompt: "make the failing test pass", + runId: "run-1", + attemptId: "attempt-1", + createdBy: "wimse://test/loop", + }); +} + +const allEvents = (host: FakeHost) => host.delivered.flat(); + +describe("dispatcher — stale workers are fenced", () => { + test("a replaced worker finishing late changes nothing; its successor's completion applies", async () => { + const host = new FakeHost(); + // Boot A starts worker-1, then "crashes" — but the object (and the worker + // it watches) is still around, which is what a slow-dying process or a + // worker outliving its daemon looks like. + const bootA = makeDispatcher(host); + const id = enqueueSpawn(bootA); + await bootA.tick(); + expect(store.dispatchGet(id)!.workerSessionId).toBe("worker-1"); + + // Boot B reclaims and re-claims; worker-1 did not survive + // as far as B can tell, so it spawns worker-2. + host.continueSucceeds = false; + expireLease(id); // unowned rows (no daemonId) are reclaimed on lease expiry only + const bootB = makeDispatcher(host); + await bootB.tick(); + let row = store.dispatchGet(id)!; + expect(row.workerSessionId).toBe("worker-2"); + expect(row.generation).toBe(3); + + // worker-1 finishes: boot A still watches it, at generation 1. + bootA.onSessionStatus("worker-1", "idle"); + await Bun.sleep(10); + row = store.dispatchGet(id)!; + expect(row.status).toBe("running"); + expect(row.resultDigest).toBeNull(); + expect(host.audits.some((a) => a.startsWith("dispatch.stale_result") && a.includes("generation=1"))).toBe(true); + expect(allEvents(host).some((e) => e.type === "task_done")).toBe(false); + // The stale path tears nothing down — worker-2 is someone else's now. + expect(host.destroyed).toEqual([]); + + bootB.onSessionStatus("worker-2", "idle"); + await Bun.sleep(10); + row = store.dispatchGet(id)!; + expect(row.status).toBe("done"); + expect(allEvents(host).filter((e) => e.type === "task_done")).toHaveLength(1); + expect(host.destroyed).toEqual(["worker-2"]); + bootA.stop(); + bootB.stop(); + }); + + test("a lease-expired worker is unwatched at reclaim, so its late turn is never routed", async () => { + const host = new FakeHost(); + const d = makeDispatcher(host); + const id = enqueueSpawn(d); + await d.tick(); + host.continueSucceeds = false; + expireLease(id); + await d.tick(); // reclaim → re-claim → worker-2 + expect(d.tasksForSession("worker-1")).toEqual([]); + expect(d.tasksForSession("worker-2")).toEqual([id]); + d.stop(); + }); +}); + +describe("dispatcher — held tasks", () => { + test("hold survives lease expiry and a restart; unhold resumes and the worker's result applies", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueSpawn(d); + await d.tick(); + const gen = store.dispatchGet(id)!.generation; + expect(d.hold(id, gen + 1, "stale")).toBe(false); + expect(d.hold(id, gen, "approval outside the profile")).toBe(true); + expect(d.tasksForSession("worker-1")).toEqual([]); + + expireLease(id); + const restarted = makeDispatcher(host); + await restarted.tick(); + expect(store.dispatchGet(id)!.status).toBe("held"); + expect(host.spawned).toHaveLength(1); + + expect(await restarted.unhold(id, gen)).toBe(true); + expect(restarted.tasksForSession("worker-1")).toEqual([id]); + restarted.onSessionStatus("worker-1", "idle"); + await Bun.sleep(10); + expect(store.dispatchGet(id)!.status).toBe("done"); + d.stop(); + restarted.stop(); + }); + + test("unhold of a worker that settled while held finishes the task immediately", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueSpawn(d); + await d.tick(); + const gen = store.dispatchGet(id)!.generation; + d.hold(id, gen, "approval"); + // The approval was granted out of band and the turn ran to the end while + // nobody was watching — no transition will ever arrive for it. + host.statuses.set("worker-1", "idle"); + d.onSessionStatus("worker-1", "idle"); + expect(store.dispatchGet(id)!.status).toBe("held"); + + await d.unhold(id, gen); + await Bun.sleep(10); + expect(store.dispatchGet(id)!.status).toBe("done"); + d.stop(); + }); +}); + +describe("dispatcher — attempt tasks", () => { + test("enqueue insists on loop ids for attempts and refuses them elsewhere", () => { + const d = makeDispatcher(new LoopHost()); + expect(() => + d.enqueue({ ...TENANT, kind: "attempt", shape: "ship", prompt: "x", createdBy: "c" }), + ).toThrow(/runId and attemptId/); + expect(() => + d.enqueue({ + ...TENANT, + kind: "spawn", + shape: "ship", + prompt: "x", + createdBy: "c", + runId: "r", + attemptId: "a", + }), + ).toThrow(/only valid on an attempt/); + }); + + test("routes to the host's attempt hooks: no fleet events, no teardown", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + expect(store.dispatchGet(id)!.failureLimit).toBe(1); + await d.tick(); + expect(host.spawned).toEqual([]); + expect(host.attemptSpawned.map((t) => t.attemptId)).toEqual(["attempt-1"]); + + d.onSessionStatus("attempt-worker-1", "idle"); + await Bun.sleep(10); + await d.tick(); + expect(store.dispatchGet(id)!.status).toBe("done"); + expect(host.settled.map((s) => s.status)).toEqual(["done"]); + expect(host.settled[0]!.task.status).toBe("done"); + expect(host.destroyed).toEqual([]); + expect(allEvents(host)).toEqual([]); + d.stop(); + }); + + test("an approval wedge reports to the loop, not the conductor, and the loop can hold it", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + await d.tick(); + d.onSessionStatus("attempt-worker-1", "waiting_approval"); + await d.tick(); + expect(host.waiting.map((t) => t.id)).toEqual([id]); + expect(allEvents(host)).toEqual([]); + expect(store.dispatchGet(id)!.status).toBe("running"); // not failed + + const waiting = host.waiting[0]!; + expect(d.hold(waiting.id, waiting.generation, "approval outside the run's profile")).toBe(true); + expireLease(id); + await makeDispatcher(host).tick(); + expect(store.dispatchGet(id)!.status).toBe("held"); + expect(host.settled).toEqual([]); + d.stop(); + }); + + test("a reclaim is terminal and reported as `reclaimed`; the queue never re-runs it", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + await d.tick(); + expireLease(id); + await d.tick(); + await d.tick(); + expect(store.dispatchGet(id)!.status).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["reclaimed"]); + expect(host.attemptSpawned).toHaveLength(1); + expect(host.destroyed).toEqual([]); + expect(allEvents(host)).toEqual([]); + d.stop(); + }); + + test("a worker error settles the attempt as `failed` without a retry", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + await d.tick(); + d.onSessionStatus("attempt-worker-1", "error"); + await Bun.sleep(10); + await d.tick(); + expect(store.dispatchGet(id)!.status).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["failed"]); + expect(host.attemptSpawned).toHaveLength(1); + expect(allEvents(host)).toEqual([]); + d.stop(); + }); + + test("fails closed on a host without loop support — never runs as an ordinary spawn", async () => { + const host = new FakeHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + await d.tick(); + const row = store.dispatchGet(id)!; + expect(row.status).toBe("failed"); + expect(row.error).toMatch(/no loop host/); + expect(host.spawned).toEqual([]); + expect(allEvents(host)).toEqual([]); + d.stop(); + }); + + test("attempts count against the per-tenant worker cap", async () => { + const host = new LoopHost(); + const d = new Dispatcher(store, host, { + tickMs: 999_999, + leaseMs: LEASE_MS, + failureLimit: 3, + maxConcurrentWorkers: 1, + workerToolBudget: 10, + retryBaseMs: 0, + }); + enqueueAttempt(d); + const spawn = enqueueSpawn(d); + await d.tick(); + expect(host.attemptSpawned).toHaveLength(1); + expect(host.spawned).toEqual([]); + expect(store.dispatchGet(spawn)!.status).toBe("queued"); + d.stop(); + }); +}); diff --git a/src/tests/dispatch-host.test.ts b/src/tests/dispatch-host.test.ts index bcc28800..c03c996b 100644 --- a/src/tests/dispatch-host.test.ts +++ b/src/tests/dispatch-host.test.ts @@ -79,6 +79,21 @@ function sayTurn(text: string): ProviderEvent[] { let tmp: string; let workdir: string; let store: Store; + +/** + * Force a dispatch task terminal, bypassing the claim lifecycle. These tests + * exercise what is drawn or delivered once a task has settled, not how it got + * there — and `dispatchComplete` / `dispatchFail` now refuse a row that was + * never claimed (the generation fence), which is exactly right for a worker + * result and exactly wrong for staging one. + */ +function settleTask(id: string, status: "done" | "failed", digest: string): void { + store.database + .prepare( + "UPDATE dispatch_tasks SET status = ?, result_digest = ?, error = ?, updated_at = ? WHERE id = ?", + ) + .run(status, status === "done" ? digest : null, status === "done" ? null : digest, Date.now(), id); +} let transcript: TranscriptStore; let providers: MockSessionProvider[]; let manager: SessionManager; @@ -391,7 +406,7 @@ describe("dispatch host — event routing", () => { shape: "scout", }); for (const m of store.dispatchGroupMembers(AUTH.accountId, AUTH.projectId, groupId)) { - store.dispatchComplete(m.id, `digest ${m.id.slice(0, 4)}`, Date.now()); + settleTask(m.id, "done", `digest ${m.id.slice(0, 4)}`); } store.dispatchEventAdd({ accountId: AUTH.accountId, @@ -437,7 +452,7 @@ describe("dispatch host — event routing", () => { targetSession: targetId, prompt: "go", }); - store.dispatchComplete(taskId, "delivered", Date.now()); + settleTask(taskId, "done", "delivered"); store.dispatchEventAdd({ accountId: AUTH.accountId, projectId: AUTH.projectId, diff --git a/src/tests/dispatch-store.test.ts b/src/tests/dispatch-store.test.ts index 13cafe0b..e736ba6d 100644 --- a/src/tests/dispatch-store.test.ts +++ b/src/tests/dispatch-store.test.ts @@ -129,11 +129,13 @@ describe("dispatch queue — claims", () => { describe("dispatch queue — failure semantics", () => { test("retryable failures requeue until failure_limit, then auto-block", () => { const id = enqueue(); // failureLimit 2 - store.dispatchClaimNext("boot-1", 100); - expect(store.dispatchFail(id, "worker died", 101, { retryable: true })).toBe("queued"); + const first = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchFail(id, first.generation, "worker died", 101, { retryable: true })).toBe( + "queued", + ); - store.dispatchClaimNext("boot-1", 102); - expect(store.dispatchFail(id, "worker died again", 103, { retryable: true })).toBe( + const second = store.dispatchClaimNext("boot-1", 102)!; + expect(store.dispatchFail(id, second.generation, "worker died again", 103, { retryable: true })).toBe( "blocked", ); const row = store.dispatchGet(id)!; @@ -143,16 +145,18 @@ describe("dispatch queue — failure semantics", () => { test("non-retryable failures go terminal immediately", () => { const id = enqueue(); - store.dispatchClaimNext("boot-1", 100); - expect(store.dispatchFail(id, "target gone", 101, { retryable: false })).toBe("failed"); + const claim = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchFail(id, claim.generation, "target gone", 101, { retryable: false })).toBe( + "failed", + ); }); test("complete stores the digest and clears the error", () => { const id = enqueue(); - store.dispatchClaimNext("boot-1", 100); - store.dispatchFail(id, "first try failed", 101, { retryable: true }); - store.dispatchClaimNext("boot-1", 102); - store.dispatchComplete(id, "all done", 103); + const first = store.dispatchClaimNext("boot-1", 100)!; + store.dispatchFail(id, first.generation, "first try failed", 101, { retryable: true }); + const second = store.dispatchClaimNext("boot-1", 102)!; + expect(store.dispatchComplete(id, second.generation, "all done", 103)).toBe(true); const row = store.dispatchGet(id)!; expect(row.status).toBe("done"); @@ -164,8 +168,8 @@ describe("dispatch queue — failure semantics", () => { describe("dispatch queue — stale-claim reclaim (crash recovery)", () => { test("claims held by another boot are reclaimed with attempts++", () => { const id = enqueue(); - store.dispatchClaimNext("boot-DEAD", 100); - store.dispatchMarkRunning(id, "worker-1", 100); + const claim = store.dispatchClaimNext("boot-DEAD", 100)!; + store.dispatchMarkRunning(id, claim.generation, "worker-1", 100); const reclaimed = store.dispatchReclaimStale("boot-NEW", 60_000, 200); expect(reclaimed.map((t) => t.id)).toEqual([id]); @@ -191,8 +195,8 @@ describe("dispatch queue — stale-claim reclaim (crash recovery)", () => { test("touch renews the lease and prevents reclaim", () => { const id = enqueue(); - store.dispatchClaimNext("boot-1", 1_000); - store.dispatchTouch([id], 50_000); + const claim = store.dispatchClaimNext("boot-1", 1_000)!; + store.dispatchTouch([{ id, generation: claim.generation }], 50_000); const reclaimed = store.dispatchReclaimStale("boot-1", 60_000, 100_000); expect(reclaimed).toHaveLength(0); }); @@ -221,8 +225,8 @@ describe("dispatch queue — tenancy + counters", () => { enqueue({ kind: "send", targetSession: "s", workdir: undefined }); enqueue({ accountId: "acc-b", projectId: "proj-b", id: "task-b" }); - store.dispatchClaimNext("boot-1", 100); // claims `a` (oldest) - store.dispatchMarkRunning(a, "w-1", 100); + const claim = store.dispatchClaimNext("boot-1", 100)!; // claims `a` (oldest) + store.dispatchMarkRunning(a, claim.generation, "w-1", 100); expect(store.dispatchActiveSpawnCount(TENANT.accountId, TENANT.projectId)).toBe(1); expect(store.dispatchActiveSpawnCount("acc-b", "proj-b")).toBe(0); }); From 5b96a8be9427c62349fd73b08af9d96967790171 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 25 Oct 2026 08:22:44 +0800 Subject: [PATCH 2/3] fix: bound held attempts, scope hold/unhold/cancel to the owning daemon, keep attempts off the fleet board MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-1 audit of the dispatch-generations slice: - web: classify the new `held` dispatch status (fleet lanes → awaiting, with an exhaustive default; attention queue → approval, deduped against the session's own; panel badge), fixing the TS2366 build break. - held_until column + DispatchConfig.holdMaxMs (default 24h). Reclaim settles an expired hold, and a hold placed by a previous boot of this daemon, as `reclaimed`; held attempts still count toward maxConcurrentWorkers. - Dispatcher.cancel(taskId, reason) for attempts in any non-terminal state, bumping the generation and reporting onAttemptSettled(task, 'failed'). - dispatchHold is attempt-only; hold/unhold/cancel are scoped by owner_daemon. - unhold: a hold from a previous boot settles `reclaimed` (never reads the worker's idle as success); a gone worker fails at once; a worker still waiting on an approval re-fires onAttemptWaiting. - late-start: only a session created in that call is destroyed; a late attempt worker is reported via the new onAttemptOrphanWorker hook. - #execute's catch unwatches before the stale-generation return. - attempts are excluded from fleet_tasks, the board snapshot, its deltas and its counters. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/daemon/dispatch.ts | 169 ++++++++-- src/daemon/store.ts | 173 ++++++++-- src/tests/dispatch-generations.test.ts | 450 ++++++++++++++++++++----- web/src/components/SessionListPane.tsx | 4 +- web/src/lib/attention.test.ts | 20 ++ web/src/lib/attention.ts | 19 +- web/src/lib/fleet-lanes.test.ts | 6 + web/src/lib/fleet-lanes.ts | 11 + 8 files changed, 697 insertions(+), 155 deletions(-) diff --git a/src/daemon/dispatch.ts b/src/daemon/dispatch.ts index 8b74510a..1ea78c71 100644 --- a/src/daemon/dispatch.ts +++ b/src/daemon/dispatch.ts @@ -60,6 +60,12 @@ export interface DispatchConfig { * the SAME tick and burn its whole failure budget in milliseconds. */ retryBaseMs: number; + /** + * Longest a loop attempt may stay `held`. A held attempt keeps its worker + * session and counts against `maxConcurrentWorkers`; past this the hold is + * treated as dead and the attempt settles `reclaimed`, freeing the slot. + */ + holdMaxMs: number; /** * Stable identity (host:port) of the daemon that owns the tasks it enqueues, * scoping claim and reclaim to its own work. @@ -85,6 +91,7 @@ export const DEFAULT_DISPATCH_CONFIG: DispatchConfig = { maxConcurrentWorkers: 2, workerToolBudget: 50, retryBaseMs: 15_000, + holdMaxMs: 24 * 60 * 60_000, }; /** Thrown by host methods when retrying can never succeed (target gone, bad input). */ @@ -146,6 +153,13 @@ export interface DispatcherHost { * dispatcher neither fails it nor tells a conductor. */ onAttemptWaiting?(task: DispatchTaskRow): void; + /** + * An attempt worker came up for a claim that was replaced while it started + * (the start outlived the lease). The dispatcher does not record or watch + * it, and never destroys a loop's sessions, so without this report the + * session would run on with nobody owning it. + */ + onAttemptOrphanWorker?(task: DispatchTaskRow, sessionId: string): void; /** * Inject pending events into whichever session each one belongs to, batched * per target — the tenant's conductor for its own dispatches, and the @@ -416,18 +430,28 @@ export class Dispatcher { } /** - * Park a live task as `held` — waiting on something outside its worker (an - * approval outside a loop's permission profile, say). Reclaim never touches - * a held task, so the wait costs no attempt however long the person takes. + * Park a live loop attempt as `held` — waiting on something outside its + * worker (an approval outside the run's permission profile). Reclaim leaves a + * held task alone until its hold lapses (`holdMaxMs`) or the boot that placed + * it is gone, so the wait costs no attempt while a person decides. * - * `generation` is the one the caller was handed with the task (e.g. via + * `generation` is the one the caller was handed with the task (via * `onAttemptWaiting`); a stale generation is refused, so a replaced worker's - * request cannot park its successor. The watch is dropped while held — a - * status change is not this dispatcher's to act on until `unhold`, which - * re-reads the worker's state rather than relying on a transition it missed. + * request cannot park its successor. Attempts only, and only this daemon's. + * The watch is dropped while held — a status change is not this + * dispatcher's to act on until `unhold`, which re-reads the worker's state + * rather than relying on a transition it missed. */ hold(taskId: string, generation: number, reason: string): boolean { - const held = this.#store.dispatchHold(taskId, generation, reason, Date.now()); + const now = Date.now(); + const held = this.#store.dispatchHold( + taskId, + generation, + reason, + now, + now + this.#config.holdMaxMs, + this.#config.daemonId, + ); if (!held) { this.#staleResult(taskId, generation, "hold"); return false; @@ -443,34 +467,99 @@ export class Dispatcher { } /** - * Resume a held task as `running` under this boot, with a fresh lease, at the - * same generation — the same worker carries on, so its result still applies. + * Resume a held attempt as `running` with a fresh lease, at the same + * generation — the same worker carries on, so its result still applies. + * Returns true only when the attempt is running again. + * + * A hold placed by a previous boot is NOT resumed: the approval it waited on + * died with that process, and the worker's idle would read as a success it + * never had. It settles `reclaimed` instead (reclaim does the same on the + * first tick; this covers an unhold that races it). * - * The worker may have moved while held (or the daemon restarted), so its - * CURRENT status decides what happens next: still working → watch it; - * already settled → finish the task now; gone → leave it unwatched, and the - * lease reclaims it like any other dead worker. + * Otherwise the worker's CURRENT status decides: gone → fail now; settled → + * finish now; still waiting on an approval → tell the loop again; working → + * watch it. */ async unhold(taskId: string, generation: number): Promise { - const resumed = this.#store.dispatchUnhold(taskId, generation, this.bootId, Date.now()); - if (!resumed) { + const task = this.#store.dispatchGet(taskId); + if (!task || task.status !== "held" || task.generation !== generation) { + this.#staleResult(taskId, generation, "unhold"); + return false; + } + const now = Date.now(); + if (task.claimOwner !== this.bootId) { + const settled = this.#store.dispatchSettleHeld( + taskId, + generation, + "reclaimed: hold outlived the boot that placed it", + now, + this.#config.daemonId, + ); + if (!settled) { + this.#staleResult(taskId, generation, "unhold"); + return false; + } + this.#settleAttempt(settled, "reclaimed"); + this.#signalBoardChange(); + return false; + } + if (!this.#store.dispatchUnhold(taskId, generation, this.bootId, now, this.#config.daemonId)) { this.#staleResult(taskId, generation, "unhold"); return false; } this.#host.audit("dispatch.unheld", `task=${taskId} generation=${generation}`); this.#signalBoardChange(); - const task = this.#store.dispatchGet(taskId); - const sessionId = task?.workerSessionId; - if (!sessionId) return true; - const status = this.#host.workerStatus(sessionId); + + const sessionId = task.workerSessionId; + const status = sessionId ? this.#host.workerStatus(sessionId) : null; + if (!sessionId || status === null) { + // Nothing to resume. Waiting for the lease to notice would hold a worker + // slot for the whole lease on a session that is already known gone. + if ( + this.#store.dispatchFail(taskId, generation, "worker session gone while held", Date.now(), { + retryable: false, + }) + ) { + this.#settleAttempt(this.#store.dispatchGet(taskId) ?? task, "failed"); + } + return false; + } if (status === "idle" || status === "error") { await this.#finishWorkerTask(taskId, generation, sessionId, status); - } else if (status) { - this.#watch(sessionId, taskId, generation); + return true; + } + this.#watch(sessionId, taskId, generation); + if (status === "waiting_approval") { + // Still (or again) parked on an approval: the transition that would + // report it happened while unwatched, so report it now. + const fresh = this.#store.dispatchGet(taskId) ?? task; + this.#callAttemptHook("onAttemptWaiting", () => this.#host.onAttemptWaiting?.(fresh)); } return true; } + /** + * Cancel a loop attempt in any non-terminal state (queued, claimed, running, + * held): terminal `failed` with `reason`, reported as `failed`. The loop + * controller's call — so it is not generation-fenced, but it bumps the + * generation, which fences the worker's late result. The worker session is + * the loop's to stop; the dispatcher only stops watching it. Returns false + * when there was nothing to cancel (terminal already, not an attempt, or + * another daemon's). + */ + cancel(taskId: string, reason: string): boolean { + const row = this.#store.dispatchCancelAttempt(taskId, reason, Date.now(), this.#config.daemonId); + if (!row) { + this.#host.audit("dispatch.cancel_refused", `task=${taskId}`); + return false; + } + this.#unwatchTask(taskId); + this.#host.audit("dispatch.cancelled", `task=${taskId} reason=${reason.slice(0, 200)}`); + this.#settleAttempt(row, "failed"); + this.#signalBoardChange(); + return true; + } + /** * One dispatcher pass: reclaim stale claims, renew live leases, claim + * execute ready tasks, deliver pending conductor events. Public so tests @@ -682,7 +771,7 @@ export class Dispatcher { const continued = await this.#host.continueWorker(task); if (continued) { if (!this.#store.dispatchMarkRunning(task.id, task.generation, task.workerSessionId, now)) { - return await this.#staleStart(task, task.workerSessionId); + return await this.#staleStart(task, task.workerSessionId, false); } this.#watch(task.workerSessionId, task.id, task.generation); this.#host.audit( @@ -695,7 +784,7 @@ export class Dispatcher { } const { sessionId } = await this.#host.spawnWorker(task); if (!this.#store.dispatchMarkRunning(task.id, task.generation, sessionId, Date.now())) { - return await this.#staleStart(task, sessionId); + return await this.#staleStart(task, sessionId, true); } this.#watch(sessionId, task.id, task.generation); this.#host.audit("dispatch.spawned", `task=${task.id} worker=${sessionId}`); @@ -707,12 +796,14 @@ export class Dispatcher { retryable, notBefore: retryable ? this.#retryAt(task.attempts, now) : undefined, }); - if (status === null) return this.#staleResult(task.id, task.generation, "execute"); // Whatever this execution registered is now dead: a grouped send watches // its target BEFORE delivering, so a delivery that threw left a watcher // behind, and the target's next idle arrived as a result for a task that - // had already failed. A re-queued task re-watches when it runs again. + // had already failed. A re-queued task re-watches when it runs again — + // and a stale one must not leave its watcher behind either, so this + // comes before the fence. this.#unwatchTask(task.id); + if (status === null) return this.#staleResult(task.id, task.generation, "execute"); this.#host.audit( "dispatch.failed", `task=${task.id} retryable=${retryable} status=${status} error=${message.slice(0, 200)}`, @@ -763,7 +854,7 @@ export class Dispatcher { } if (task.workerSessionId && host.continueAttempt && (await host.continueAttempt(task))) { if (!this.#store.dispatchMarkRunning(task.id, task.generation, task.workerSessionId, Date.now())) { - return await this.#staleStart(task, task.workerSessionId); + return await this.#staleStart(task, task.workerSessionId, false); } this.#watch(task.workerSessionId, task.id, task.generation); this.#host.audit( @@ -774,7 +865,7 @@ export class Dispatcher { } const { sessionId } = await host.spawnAttemptWorker(task); if (!this.#store.dispatchMarkRunning(task.id, task.generation, sessionId, Date.now())) { - return await this.#staleStart(task, sessionId); + return await this.#staleStart(task, sessionId, true); } this.#watch(sessionId, task.id, task.generation); this.#host.audit( @@ -786,13 +877,23 @@ export class Dispatcher { /** * A worker came up for a claim that was reclaimed while it was starting — * the start outlived the lease. The successor owns the task now, so this - * worker must not be recorded or watched. A spawned worker is ours and is - * torn down; an attempt's belongs to the loop host, which the reclaim - * already told (`reclaimed`). + * worker must not be recorded or watched. + * + * `fresh` = this call created the session. Only then is a spawn's worker + * ours to tear down: a CONTINUED session predates the call and may be the + * very one the successor has since re-attached to. An attempt's worker is + * never destroyed here — it is reported to the loop host instead, which + * otherwise could not know a session is running for a claim nobody holds. */ - async #staleStart(task: DispatchTaskRow, sessionId: string): Promise { - this.#staleResult(task.id, task.generation, `start worker=${sessionId}`); - if (task.kind === "spawn") { + async #staleStart(task: DispatchTaskRow, sessionId: string, fresh: boolean): Promise { + this.#staleResult(task.id, task.generation, `start worker=${sessionId} fresh=${fresh}`); + if (task.kind === "attempt") { + this.#callAttemptHook("onAttemptOrphanWorker", () => + this.#host.onAttemptOrphanWorker?.(task, sessionId), + ); + return; + } + if (task.kind === "spawn" && fresh) { await this.#host.destroyWorker(sessionId, `task ${task.id} claim was replaced while starting`); } } diff --git a/src/daemon/store.ts b/src/daemon/store.ts index f1bcd1cf..d1d9037f 100644 --- a/src/daemon/store.ts +++ b/src/daemon/store.ts @@ -86,6 +86,12 @@ export interface DispatchTaskRow { * what every row written before generations existed reads as. */ generation: number; + /** + * Epoch ms a `held` task's hold lapses. A hold occupies a worker session and + * counts against the worker cap, so it must not be able to do so forever: + * past this, reclaim treats the hold as dead. NULL unless held. + */ + heldUntil: number | null; createdAt: number; updatedAt: number; } @@ -132,6 +138,7 @@ interface RawDispatchRow { run_id: string | null; attempt_id: string | null; generation: number; + held_until: number | null; created_at: number; updated_at: number; } @@ -165,6 +172,7 @@ function rowToDispatchTask(r: RawDispatchRow): DispatchTaskRow { runId: r.run_id ?? null, attemptId: r.attempt_id ?? null, generation: r.generation ?? 0, + heldUntil: r.held_until ?? null, createdAt: r.created_at, updatedAt: r.updated_at, }; @@ -507,6 +515,7 @@ export class Store { run_id TEXT, -- attempt: loop run id; NULL otherwise attempt_id TEXT, -- attempt: loop attempt id; NULL otherwise generation INTEGER NOT NULL DEFAULT 0, -- claim generation; fences stale worker results + held_until INTEGER, -- held: epoch ms the hold lapses; NULL otherwise created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); @@ -619,6 +628,7 @@ export class Store { this.#addColumnIfMissing("dispatch_tasks", "run_id", "TEXT"); this.#addColumnIfMissing("dispatch_tasks", "attempt_id", "TEXT"); this.#addColumnIfMissing("dispatch_tasks", "generation", "INTEGER NOT NULL DEFAULT 0"); + this.#addColumnIfMissing("dispatch_tasks", "held_until", "INTEGER"); // AFTER the ALTERs, never inside the CREATE block above. On an existing // database `CREATE TABLE IF NOT EXISTS` is a no-op, so an index declared // there would run against a table that does not have the column yet and @@ -1278,42 +1288,119 @@ export class Store { } /** - * Park a live task on something outside its worker (`held`). Fenced on the - * generation like a result, since it is one: a stale worker's approval - * request must not park the claim its successor holds. Returns whether it - * applied. The claim owner and worker are kept — the task is still this - * daemon's, just not progressing — and reclaim never selects a held row. + * Park a live loop attempt on something outside its worker (`held`) until + * `heldUntil`. Fenced on the generation like a result, since it is one: a + * stale worker's approval request must not park the claim its successor + * holds. Returns whether it applied. + * + * Attempts only. A hold exists so a loop can wait on a person without its + * attempt being reclaimed; a send or spawn has no loop to resolve the wait, + * so a held one would sit on a worker session with nobody to release it. + * + * Scoped to this daemon's tasks (or unowned pre-upgrade rows): another + * daemon's attempt is not this daemon's to park. The claim owner and worker + * are kept — the task is still this boot's, just not progressing. */ - dispatchHold(id: string, generation: number, reason: string, now: number): boolean { + dispatchHold( + id: string, + generation: number, + reason: string, + now: number, + heldUntil: number, + ownerDaemon?: string, + ): boolean { const res = this.#db .prepare( `UPDATE dispatch_tasks - SET status = 'held', error = ?, updated_at = ? - WHERE id = ? AND generation = ? AND status IN ('claimed', 'running')`, + SET status = 'held', error = ?, held_until = ?, updated_at = ? + WHERE id = ? AND generation = ? AND status IN ('claimed', 'running') + AND kind = 'attempt' + AND (owner_daemon IS NULL OR owner_daemon IS ?)`, ) - .run(reason, now, id, generation); + .run(reason, heldUntil, now, id, generation, ownerDaemon ?? null); return res.changes > 0; } /** - * Resume a held task as `running` under THIS boot with a fresh lease. + * Resume a held task as `running` with a fresh lease, at the same generation + * — the same worker carries on, so its eventual result still applies. * - * Re-owned by the caller's boot id because the hold may have outlived the - * daemon that placed it; keeping the old boot id would have the very next - * tick reclaim the task as a crashed claim. The generation is unchanged — - * the same worker carries on — so its eventual result still applies. + * Only for a hold THIS boot placed (`claim_owner = bootId`). A hold that + * outlived its boot waited on an approval that died with that process; the + * caller settles it as reclaimed (`dispatchSettleHeld`) instead, because + * resuming would read the worker's idle as success it never had. */ - dispatchUnhold(id: string, generation: number, bootId: string, now: number): boolean { + dispatchUnhold( + id: string, + generation: number, + bootId: string, + now: number, + ownerDaemon?: string, + ): boolean { const res = this.#db .prepare( `UPDATE dispatch_tasks - SET status = 'running', claim_owner = ?, claimed_at = ?, error = NULL, updated_at = ? - WHERE id = ? AND generation = ? AND status = 'held'`, + SET status = 'running', claimed_at = ?, held_until = NULL, error = NULL, updated_at = ? + WHERE id = ? AND generation = ? AND status = 'held' AND claim_owner IS ? + AND (owner_daemon IS NULL OR owner_daemon IS ?)`, ) - .run(bootId, now, now, id, generation); + .run(now, now, id, generation, bootId, ownerDaemon ?? null); return res.changes > 0; } + /** + * Settle a held task terminal `failed` with `error`, bumping the generation so + * its worker can no longer report. Same fence and scope as `dispatchUnhold`, + * minus the boot check — it is how a hold that cannot be resumed ends. + */ + dispatchSettleHeld( + id: string, + generation: number, + error: string, + now: number, + ownerDaemon?: string, + ): DispatchTaskRow | null { + const row = this.#db + .prepare( + `UPDATE dispatch_tasks + SET status = 'failed', attempts = attempts + 1, error = ?, claim_owner = NULL, + claimed_at = NULL, held_until = NULL, generation = generation + 1, updated_at = ? + WHERE id = ? AND generation = ? AND status = 'held' + AND (owner_daemon IS NULL OR owner_daemon IS ?) + RETURNING *`, + ) + .get(error, now, id, generation, ownerDaemon ?? null) as RawDispatchRow | null; + return row ? rowToDispatchTask(row) : null; + } + + /** + * Cancel a loop attempt in any non-terminal state — terminal `failed` with + * `reason`. Not generation-fenced: cancellation is the loop controller's + * decision, not a worker's result, and must land whichever worker currently + * holds the claim. It bumps the generation instead, so that worker's late + * result is fenced off. Returns the settled row, or null when there was + * nothing to cancel (already terminal, not an attempt, another daemon's). + */ + dispatchCancelAttempt( + id: string, + reason: string, + now: number, + ownerDaemon?: string, + ): DispatchTaskRow | null { + const row = this.#db + .prepare( + `UPDATE dispatch_tasks + SET status = 'failed', error = ?, claim_owner = NULL, claimed_at = NULL, + held_until = NULL, generation = generation + 1, updated_at = ? + WHERE id = ? AND kind = 'attempt' + AND status IN ('queued', 'claimed', 'running', 'held') + AND (owner_daemon IS NULL OR owner_daemon IS ?) + RETURNING *`, + ) + .get(reason, now, id, ownerDaemon ?? null) as RawDispatchRow | null; + return row ? rowToDispatchTask(row) : null; + } + /** * Reclaim tasks whose claim is dead: held by another boot (daemon crashed) * or past the lease (worker hung and the dispatcher stopped renewing). @@ -1323,10 +1410,14 @@ export class Store { * a re-claimed spawn can continue its (resumed) worker session. * * A reclaim bumps `generation`, fencing the replaced worker the moment its - * claim is declared dead rather than only at the next claim. `held` rows are - * never selected — they are parked, not dead. A loop attempt is never - * re-queued: a reclaimed attempt goes terminal `failed`, and the loop decides - * whether to start another. + * claim is declared dead rather than only at the next claim. A loop attempt + * is never re-queued: a reclaimed attempt goes terminal `failed`, and the + * loop decides whether to start another. + * + * A `held` row is parked, not dead, and is left alone — with two exceptions + * that bound it. Its hold lapsed (`held_until`), so it cannot sit on a worker + * slot forever; or a previous boot of this daemon placed it, so the approval + * it waited on died with that process and nothing will ever release it. */ /** * Reclaim claims this daemon can prove are dead. @@ -1351,9 +1442,12 @@ export class Store { WHEN kind = 'attempt' THEN 'failed' WHEN attempts + 1 >= failure_limit THEN 'blocked' ELSE 'queued' END, - claim_owner = NULL, claimed_at = NULL, generation = generation + 1, - error = COALESCE(error, 'reclaimed: stale claim'), updated_at = ? - WHERE status IN ('claimed', 'running') + claim_owner = NULL, claimed_at = NULL, held_until = NULL, generation = generation + 1, + error = CASE WHEN status = 'held' THEN 'reclaimed: hold lapsed or outlived its boot' + ELSE COALESCE(error, 'reclaimed: stale claim') END, + updated_at = ? + WHERE ( + status IN ('claimed', 'running') AND ( -- Mine: a previous boot of THIS daemon crashed, or the lease ran out. (owner_daemon IS ? AND (claim_owner IS NOT ? OR claimed_at IS NULL OR claimed_at + ? < ?)) @@ -1361,6 +1455,15 @@ export class Store { -- is what let another daemon steal a healthy claim. OR (owner_daemon IS NULL AND (claimed_at IS NULL OR claimed_at + ? < ?)) ) + ) OR ( + status = 'held' + AND ( + -- Mine, held by a previous boot: the wait died with that process. + (owner_daemon IS ? AND claim_owner IS NOT ?) + -- Mine or unowned, and the hold lapsed. + OR ((owner_daemon IS NULL OR owner_daemon IS ?) AND held_until IS NOT NULL AND held_until < ?) + ) + ) RETURNING *`, ) .all( @@ -1371,6 +1474,10 @@ export class Store { now, leaseMs, now, + ownerDaemon ?? null, + bootId, + ownerDaemon ?? null, + now, ) as RawDispatchRow[]; return rows.map(rowToDispatchTask); } @@ -1490,6 +1597,10 @@ export class Store { * showing every other session's dispatches — target names, prompts' result * digests — would be the one place that scoping leaked. Filtered in SQL, not * after the fact, so `limit` still returns `limit` of the caller's own rows. + * + * Loop attempts are excluded (as on every fleet-board read): they are a + * loop's internals, not conductor dispatches, and get board rows of their + * own — one per run — in a later slice. */ dispatchListForTenant( accountId: string, @@ -1502,6 +1613,7 @@ export class Store { `SELECT * FROM dispatch_tasks WHERE account_id = ? AND project_id = ? AND (? IS NULL OR created_by = ?) + AND kind != 'attempt' ORDER BY created_at DESC LIMIT ?`, ) .all( @@ -1545,6 +1657,10 @@ export class Store { * Computed in SQL over the WHOLE table rather than by counting the capped * task list the snapshot ships: a tenant with more than `limit` tasks would * otherwise report "3 active" purely because the older ones fell off the page. + * + * Loop attempts are excluded (as on every fleet-board read): they are a + * loop's internals, not conductor dispatches, and get board rows of their + * own — one per run — in a later slice. */ dispatchStatusCounts( accountId: string, @@ -1556,7 +1672,7 @@ export class Store { SUM(CASE WHEN status IN ('queued','claimed','running','held') THEN 1 ELSE 0 END) AS active, SUM(CASE WHEN status = 'blocked' THEN 1 ELSE 0 END) AS blocked FROM dispatch_tasks - WHERE account_id = ? AND project_id = ?`, + WHERE account_id = ? AND project_id = ? AND kind != 'attempt'`, ) .get(accountId, projectId) as { active: number | null; blocked: number | null }; // SUM over zero rows is NULL, not 0. @@ -1573,6 +1689,10 @@ export class Store { * watermark. Re-sending a task the client already has is harmless — a delta * carries the whole row and is idempotent by construction — whereas dropping * one strands a node in a stale state until the next unrelated change. + * + * Loop attempts are excluded (as on every fleet-board read): they are a + * loop's internals, not conductor dispatches, and get board rows of their + * own — one per run — in a later slice. */ dispatchTasksUpdatedSince( accountId: string, @@ -1583,6 +1703,7 @@ export class Store { .prepare( `SELECT * FROM dispatch_tasks WHERE account_id = ? AND project_id = ? AND updated_at >= ? + AND kind != 'attempt' ORDER BY updated_at ASC`, ) .all(accountId, projectId, since) as RawDispatchRow[]; diff --git a/src/tests/dispatch-generations.test.ts b/src/tests/dispatch-generations.test.ts index 19a0e565..12f23627 100644 --- a/src/tests/dispatch-generations.test.ts +++ b/src/tests/dispatch-generations.test.ts @@ -86,7 +86,7 @@ describe("dispatch store — claim generations", () => { expect(store.dispatchComplete(id, old.generation, "stale digest", 400)).toBe(false); expect(store.dispatchFail(id, old.generation, "stale error", 400, { retryable: true })).toBeNull(); expect(store.dispatchMarkRunning(id, old.generation, "worker-1", 400)).toBe(false); - expect(store.dispatchHold(id, old.generation, "stale approval", 400)).toBe(false); + expect(store.dispatchHold(id, old.generation, "stale approval", 400, 1_000_000)).toBe(false); let row = store.dispatchGet(id)!; expect(row.status).toBe("running"); expect(row.workerSessionId).toBe("worker-2"); @@ -118,41 +118,97 @@ describe("dispatch store — claim generations", () => { }); describe("dispatch store — held tasks", () => { - test("a held task is never reclaimed, by boot mismatch or by lease expiry", () => { - const id = enqueueRow(); + const HOLD_END = 1_000_000; + const attempt = (o: Partial[0]> = {}) => + enqueueRow({ kind: "attempt", runId: "run", attemptId: `att-${seq + 1}`, failureLimit: 1, ...o }); + + test("only a loop attempt can be held — a spawn with nobody to release it is refused", () => { + const spawn = enqueueRow(); const claim = store.dispatchClaimNext("boot-1", 100)!; - expect(store.dispatchHold(id, claim.generation, "waiting on approval", 150)).toBe(true); - expect(store.dispatchReclaimStale("boot-2", LEASE_MS, 10_000_000)).toEqual([]); - const row = store.dispatchGet(id)!; - expect(row.status).toBe("held"); - expect(row.attempts).toBe(0); - expect(row.generation).toBe(claim.generation); + expect(claim.id).toBe(spawn); + expect(store.dispatchHold(spawn, claim.generation, "approval", 101, HOLD_END)).toBe(false); + expect(store.dispatchGet(spawn)!.status).toBe("claimed"); }); - test("unhold resumes under the caller's boot with a fresh lease and the same generation", () => { - const id = enqueueRow(); - const claim = store.dispatchClaimNext("boot-OLD", 100)!; - store.dispatchHold(id, claim.generation, "approval", 150); - expect(store.dispatchUnhold(id, claim.generation + 1, "boot-NEW", 5_000)).toBe(false); - expect(store.dispatchUnhold(id, claim.generation, "boot-NEW", 5_000)).toBe(true); + test("this boot's hold survives lease expiry until it lapses, then is reclaimed terminal", () => { + const id = attempt(); + const claim = store.dispatchClaimNext("boot-1", 100)!; + expect(store.dispatchHold(id, claim.generation, "approval", 150, HOLD_END)).toBe(true); + expect(store.dispatchGet(id)!.heldUntil).toBe(HOLD_END); + // Long past the lease, inside the hold: untouched. + expect(store.dispatchReclaimStale("boot-1", LEASE_MS, HOLD_END - 1)).toEqual([]); + expect(store.dispatchGet(id)!.status).toBe("held"); + // Past the hold: dead, and an attempt is never re-queued. + const [reclaimed] = store.dispatchReclaimStale("boot-1", LEASE_MS, HOLD_END + 1); + expect(reclaimed!.status).toBe("failed"); + expect(reclaimed!.heldUntil).toBeNull(); + expect(reclaimed!.generation).toBe(claim.generation + 1); + expect(reclaimed!.error).toMatch(/hold lapsed/); + }); + + test("a hold placed by a previous boot of this daemon is reclaimed on the next boot", () => { + const id = attempt({ ownerDaemon: "host:1" }); + const claim = store.dispatchClaimNext("boot-OLD", 100, [], "host:1")!; + store.dispatchHold(id, claim.generation, "approval", 150, HOLD_END, "host:1"); + expect(store.dispatchReclaimStale("boot-NEW", LEASE_MS, 200, "host:1").map((t) => t.id)).toEqual([id]); + expect(store.dispatchGet(id)!.status).toBe("failed"); + }); + + test("another daemon's hold is neither reclaimed on boot mismatch nor resumable from here", () => { + const id = attempt({ ownerDaemon: "host:A" }); + const claim = store.dispatchClaimNext("boot-A", 100, [], "host:A")!; + expect(store.dispatchHold(id, claim.generation, "x", 101, HOLD_END, "host:B")).toBe(false); + expect(store.dispatchHold(id, claim.generation, "x", 101, HOLD_END, "host:A")).toBe(true); + expect(store.dispatchReclaimStale("boot-B", LEASE_MS, 200, "host:B")).toEqual([]); + expect(store.dispatchUnhold(id, claim.generation, "boot-A", 300, "host:B")).toBe(false); + expect(store.dispatchSettleHeld(id, claim.generation, "x", 300, "host:B")).toBeNull(); + expect(store.dispatchCancelAttempt(id, "x", 300, "host:B")).toBeNull(); + expect(store.dispatchGet(id)!.status).toBe("held"); + }); + + test("unhold resumes only under the boot that held it, with a fresh lease and the same generation", () => { + const id = attempt(); + const claim = store.dispatchClaimNext("boot-1", 100)!; + store.dispatchHold(id, claim.generation, "approval", 150, HOLD_END); + expect(store.dispatchUnhold(id, claim.generation + 1, "boot-1", 5_000)).toBe(false); + expect(store.dispatchUnhold(id, claim.generation, "boot-OTHER", 5_000)).toBe(false); + expect(store.dispatchUnhold(id, claim.generation, "boot-1", 5_000)).toBe(true); const row = store.dispatchGet(id)!; expect(row.status).toBe("running"); - expect(row.claimOwner).toBe("boot-NEW"); expect(row.claimedAt).toBe(5_000); + expect(row.heldUntil).toBeNull(); expect(row.generation).toBe(claim.generation); - // Re-owned, so the new boot's fast path does not take it for a crash. - expect(store.dispatchReclaimStale("boot-NEW", LEASE_MS, 6_000)).toEqual([]); - // And the worker's result, at the unchanged generation, still applies. + // The worker's result, at the unchanged generation, still applies. expect(store.dispatchComplete(id, claim.generation, "ok", 7_000)).toBe(true); }); - test("only a live claim can be held, and a held task is still active and still a worker", () => { - const queued = enqueueRow(); - expect(store.dispatchHold(queued, 0, "nope", 100)).toBe(false); + test("a held attempt holds a worker slot but stays off the fleet board", () => { + const queued = attempt(); + expect(store.dispatchHold(queued, 0, "nope", 100, HOLD_END)).toBe(false); const claim = store.dispatchClaimNext("boot-1", 100)!; - store.dispatchHold(claim.id, claim.generation, "approval", 101); - expect(store.dispatchStatusCounts(TENANT.accountId, TENANT.projectId).active).toBe(1); + store.dispatchHold(claim.id, claim.generation, "approval", 101, HOLD_END); expect(store.dispatchActiveSpawnCount(TENANT.accountId, TENANT.projectId)).toBe(1); + expect(store.dispatchStatusCounts(TENANT.accountId, TENANT.projectId).active).toBe(0); + expect(store.dispatchListForTenant(TENANT.accountId, TENANT.projectId)).toEqual([]); + expect(store.dispatchTasksUpdatedSince(TENANT.accountId, TENANT.projectId, 0)).toEqual([]); + }); + + test("cancel ends an attempt from any live state and fences its worker", () => { + for (const stage of ["queued", "claimed", "held"] as const) { + const id = attempt(); + let gen = 0; + if (stage !== "queued") { + gen = store.dispatchClaimNext("boot-1", 100, [])!.generation; + if (stage === "held") store.dispatchHold(id, gen, "approval", 101, HOLD_END); + } + const row = store.dispatchCancelAttempt(id, "run cancelled", 200); + expect(row?.status).toBe("failed"); + expect(row?.error).toBe("run cancelled"); + expect(store.dispatchComplete(id, gen, "late", 300)).toBe(false); + expect(store.dispatchCancelAttempt(id, "again", 400)).toBeNull(); + } + const spawn = enqueueRow(); + expect(store.dispatchCancelAttempt(spawn, "not an attempt", 500)).toBeNull(); }); }); @@ -188,6 +244,7 @@ describe("migration — opening a database written before generations", () => { const db = new Database(dbPath); // The run index references run_id; SQLite refuses to drop an indexed column. db.exec("DROP INDEX IF EXISTS idx_dispatch_run"); + db.exec("ALTER TABLE dispatch_tasks DROP COLUMN held_until"); db.exec("ALTER TABLE dispatch_tasks DROP COLUMN generation"); db.exec("ALTER TABLE dispatch_tasks DROP COLUMN attempt_id"); db.exec("ALTER TABLE dispatch_tasks DROP COLUMN run_id"); @@ -226,12 +283,13 @@ describe("migration — opening a database written before generations", () => { const upgraded = new Store(dbPath); try { expect(columns(dbPath)).toEqual( - expect.arrayContaining(["run_id", "attempt_id", "generation"]), + expect.arrayContaining(["run_id", "attempt_id", "generation", "held_until"]), ); const row = upgraded.dispatchGet("pre-upgrade")!; expect(row.generation).toBe(0); expect(row.runId).toBeNull(); expect(row.attemptId).toBeNull(); + expect(row.heldUntil).toBeNull(); expect(row.status).toBe("claimed"); // The old binary's worker presents generation 0; once the new boot @@ -249,6 +307,26 @@ describe("migration — opening a database written before generations", () => { // ── Dispatcher ──────────────────────────────────────────────────────────── +/** + * Wait for a condition the dispatcher reaches asynchronously (a worker's turn + * end is handled on a detached promise). Polls on the event loop rather than + * sleeping a guessed interval, and fails loudly instead of passing on timing. + */ +async function until(pred: () => boolean, label = "condition"): Promise { + for (let i = 0; i < 500; i++) { + if (pred()) return; + await Bun.sleep(0); + } + throw new Error(`timed out waiting for ${label}`); +} + +/** Drain pending microtasks/immediates so "nothing happened" is actually settled. */ +async function settle(): Promise { + for (let i = 0; i < 20; i++) await Bun.sleep(0); +} + +const statusOf = (id: string) => store.dispatchGet(id)!.status; + /** Minimal scriptable host — no attempt hooks (a daemon without a loop service). */ class FakeHost implements DispatcherHost { spawned: DispatchTaskRow[] = []; @@ -257,6 +335,8 @@ class FakeHost implements DispatcherHost { audits: string[] = []; statuses = new Map(); continueSucceeds = true; + /** Runs inside continueWorker/continueAttempt before it resolves — a slow start. */ + duringContinue: (() => void) | null = null; #next = 0; async sendToSession(): Promise {} @@ -269,6 +349,7 @@ class FakeHost implements DispatcherHost { } async continueWorker(task: DispatchTaskRow): Promise { + this.duringContinue?.(); if (this.continueSucceeds && task.workerSessionId) { this.statuses.set(task.workerSessionId, "thinking"); return true; @@ -308,10 +389,14 @@ class LoopHost extends FakeHost { attemptSpawned: DispatchTaskRow[] = []; settled: Array<{ task: DispatchTaskRow; status: AttemptSettledStatus }> = []; waiting: DispatchTaskRow[] = []; + orphans: string[] = []; + /** Runs inside spawnAttemptWorker before it resolves — a slow start. */ + duringAttemptSpawn: (() => void) | null = null; #nextAttempt = 0; async spawnAttemptWorker(task: DispatchTaskRow): Promise<{ sessionId: string }> { this.attemptSpawned.push(task); + this.duringAttemptSpawn?.(); const sessionId = `attempt-worker-${++this.#nextAttempt}`; this.statuses.set(sessionId, "thinking"); return { sessionId }; @@ -324,9 +409,16 @@ class LoopHost extends FakeHost { onAttemptWaiting(task: DispatchTaskRow): void { this.waiting.push(task); } + + onAttemptOrphanWorker(_task: DispatchTaskRow, sessionId: string): void { + this.orphans.push(sessionId); + } } -function makeDispatcher(host: DispatcherHost): Dispatcher { +function makeDispatcher( + host: DispatcherHost, + over: Partial[2]> = {}, +): Dispatcher { return new Dispatcher(store, host, { tickMs: 999_999, leaseMs: LEASE_MS, @@ -334,6 +426,7 @@ function makeDispatcher(host: DispatcherHost): Dispatcher { maxConcurrentWorkers: 4, workerToolBudget: 10, retryBaseMs: 0, + ...over, }); } @@ -348,7 +441,7 @@ function enqueueSpawn(d: Dispatcher): string { }); } -function enqueueAttempt(d: Dispatcher): string { +function enqueueAttempt(d: Dispatcher, attemptId = "attempt-1"): string { return d.enqueue({ ...TENANT, kind: "attempt", @@ -356,11 +449,22 @@ function enqueueAttempt(d: Dispatcher): string { workdir: "/tmp/w", prompt: "make the failing test pass", runId: "run-1", - attemptId: "attempt-1", + attemptId, createdBy: "wimse://test/loop", }); } +/** Start an attempt and park it on an approval, as a loop would. */ +async function heldAttempt(d: Dispatcher, host: LoopHost): Promise<{ id: string; gen: number }> { + const id = enqueueAttempt(d); + await d.tick(); + d.onSessionStatus("attempt-worker-1", "waiting_approval"); + host.statuses.set("attempt-worker-1", "waiting_approval"); + const waiting = host.waiting.at(-1)!; + expect(d.hold(waiting.id, waiting.generation, "approval outside the run's profile")).toBe(true); + return { id, gen: waiting.generation }; +} + const allEvents = (host: FakeHost) => host.delivered.flat(); describe("dispatcher — stale workers are fenced", () => { @@ -374,8 +478,8 @@ describe("dispatcher — stale workers are fenced", () => { await bootA.tick(); expect(store.dispatchGet(id)!.workerSessionId).toBe("worker-1"); - // Boot B reclaims and re-claims; worker-1 did not survive - // as far as B can tell, so it spawns worker-2. + // Boot B reclaims and re-claims; worker-1 did not survive as far as B can + // tell, so it spawns worker-2. host.continueSucceeds = false; expireLease(id); // unowned rows (no daemonId) are reclaimed on lease expiry only const bootB = makeDispatcher(host); @@ -386,7 +490,7 @@ describe("dispatcher — stale workers are fenced", () => { // worker-1 finishes: boot A still watches it, at generation 1. bootA.onSessionStatus("worker-1", "idle"); - await Bun.sleep(10); + await until(() => host.audits.some((a) => a.startsWith("dispatch.stale_result")), "stale audit"); row = store.dispatchGet(id)!; expect(row.status).toBe("running"); expect(row.resultDigest).toBeNull(); @@ -396,13 +500,53 @@ describe("dispatcher — stale workers are fenced", () => { expect(host.destroyed).toEqual([]); bootB.onSessionStatus("worker-2", "idle"); - await Bun.sleep(10); - row = store.dispatchGet(id)!; - expect(row.status).toBe("done"); + await until(() => statusOf(id) === "done", "done"); + await until(() => allEvents(host).some((e) => e.type === "task_done"), "task_done"); expect(allEvents(host).filter((e) => e.type === "task_done")).toHaveLength(1); expect(host.destroyed).toEqual(["worker-2"]); - bootA.stop(); - bootB.stop(); + }); + + test("two generations on one continued worker: only the current one's result applies", async () => { + // A reclaim that CONTINUES the same session is the case a session-keyed + // fence cannot handle: both generations watch the same worker, and the + // same idle transition reaches both. Only the generation fence tells them + // apart. + const host = new FakeHost(); + const bootA = makeDispatcher(host); + const id = enqueueSpawn(bootA); + await bootA.tick(); + expireLease(id); + const bootB = makeDispatcher(host); + await bootB.tick(); // reclaim (gen 2) → re-claim (gen 3) → continue worker-1 + expect(host.spawned).toHaveLength(1); + expect(store.dispatchGet(id)!.generation).toBe(3); + expect(bootA.tasksForSession("worker-1")).toEqual([id]); + expect(bootB.tasksForSession("worker-1")).toEqual([id]); + + bootA.onSessionStatus("worker-1", "idle"); // gen 1 — fenced + await until(() => host.audits.some((a) => a.startsWith("dispatch.stale_result")), "stale audit"); + expect(statusOf(id)).toBe("running"); + bootB.onSessionStatus("worker-1", "idle"); // gen 3 — applies + await until(() => statusOf(id) === "done", "done"); + expect(host.destroyed).toEqual(["worker-1"]); + }); + + test("a continued worker that comes up after its claim was replaced is not destroyed", async () => { + // The continued session predates the call — the successor may own it now. + const host = new FakeHost(); + const d = makeDispatcher(host); + const id = enqueueSpawn(d); + await d.tick(); + expireLease(id); + host.duringContinue = () => { + host.duringContinue = null; + // The continue is slow: another reclaim lands before it resolves. + expireLease(id); + store.dispatchReclaimStale("boot-other", LEASE_MS, Date.now()); + }; + await d.tick(); + expect(host.audits.some((a) => a.startsWith("dispatch.stale_result") && a.includes("fresh=false"))).toBe(true); + expect(host.destroyed).toEqual([]); }); test("a lease-expired worker is unwatched at reclaim, so its late turn is never routed", async () => { @@ -415,53 +559,164 @@ describe("dispatcher — stale workers are fenced", () => { await d.tick(); // reclaim → re-claim → worker-2 expect(d.tasksForSession("worker-1")).toEqual([]); expect(d.tasksForSession("worker-2")).toEqual([id]); - d.stop(); }); }); -describe("dispatcher — held tasks", () => { - test("hold survives lease expiry and a restart; unhold resumes and the worker's result applies", async () => { +describe("dispatcher — held attempts", () => { + test("hold is refused for a spawn", async () => { const host = new LoopHost(); const d = makeDispatcher(host); const id = enqueueSpawn(d); await d.tick(); - const gen = store.dispatchGet(id)!.generation; + expect(d.hold(id, store.dispatchGet(id)!.generation, "approval")).toBe(false); + expect(statusOf(id)).toBe("running"); + }); + + test("hold survives lease expiry; unhold resumes and the worker's result applies", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id, gen } = await heldAttempt(d, host); expect(d.hold(id, gen + 1, "stale")).toBe(false); - expect(d.hold(id, gen, "approval outside the profile")).toBe(true); - expect(d.tasksForSession("worker-1")).toEqual([]); + expect(d.tasksForSession("attempt-worker-1")).toEqual([]); expireLease(id); - const restarted = makeDispatcher(host); - await restarted.tick(); - expect(store.dispatchGet(id)!.status).toBe("held"); - expect(host.spawned).toHaveLength(1); + await d.tick(); + expect(statusOf(id)).toBe("held"); + + host.statuses.set("attempt-worker-1", "thinking"); + expect(await d.unhold(id, gen)).toBe(true); + expect(d.tasksForSession("attempt-worker-1")).toEqual([id]); + d.onSessionStatus("attempt-worker-1", "idle"); + await until(() => host.settled.length > 0, "settled"); + expect(host.settled.map((s) => s.status)).toEqual(["done"]); + expect(statusOf(id)).toBe("done"); + }); - expect(await restarted.unhold(id, gen)).toBe(true); - expect(restarted.tasksForSession("worker-1")).toEqual([id]); - restarted.onSessionStatus("worker-1", "idle"); - await Bun.sleep(10); - expect(store.dispatchGet(id)!.status).toBe("done"); - d.stop(); - restarted.stop(); + test("a held row across a restart settles `reclaimed` on the new boot's first tick", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id } = await heldAttempt(d, host); + await makeDispatcher(host).tick(); + expect(statusOf(id)).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["reclaimed"]); + expect(host.destroyed).toEqual([]); + }); + + test("an unhold that races the restart reconcile never reads the worker's idle as success", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id, gen } = await heldAttempt(d, host); + host.statuses.set("attempt-worker-1", "idle"); // the turn ended while the daemon was down + const restarted = makeDispatcher(host); + expect(await restarted.unhold(id, gen)).toBe(false); + expect(statusOf(id)).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["reclaimed"]); }); test("unhold of a worker that settled while held finishes the task immediately", async () => { const host = new LoopHost(); const d = makeDispatcher(host); - const id = enqueueSpawn(d); - await d.tick(); - const gen = store.dispatchGet(id)!.generation; - d.hold(id, gen, "approval"); + const { id, gen } = await heldAttempt(d, host); // The approval was granted out of band and the turn ran to the end while // nobody was watching — no transition will ever arrive for it. - host.statuses.set("worker-1", "idle"); - d.onSessionStatus("worker-1", "idle"); - expect(store.dispatchGet(id)!.status).toBe("held"); + host.statuses.set("attempt-worker-1", "idle"); + d.onSessionStatus("attempt-worker-1", "idle"); + expect(statusOf(id)).toBe("held"); - await d.unhold(id, gen); - await Bun.sleep(10); - expect(store.dispatchGet(id)!.status).toBe("done"); - d.stop(); + expect(await d.unhold(id, gen)).toBe(true); + expect(statusOf(id)).toBe("done"); + expect(host.settled.map((s) => s.status)).toEqual(["done"]); + }); + + test("unhold when the worker session is gone fails the attempt at once", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id, gen } = await heldAttempt(d, host); + host.statuses.delete("attempt-worker-1"); + expect(await d.unhold(id, gen)).toBe(false); + expect(statusOf(id)).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["failed"]); + }); + + test("unhold while the worker is still waiting on an approval reports it to the loop again", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id, gen } = await heldAttempt(d, host); + expect(host.waiting).toHaveLength(1); + expect(await d.unhold(id, gen)).toBe(true); + expect(host.waiting).toHaveLength(2); + expect(host.waiting[1]!.status).toBe("running"); + expect(d.tasksForSession("attempt-worker-1")).toEqual([id]); + }); + + test("an expired hold frees its worker slot for queued work", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host, { maxConcurrentWorkers: 1 }); + const { id } = await heldAttempt(d, host); + const spawn = enqueueSpawn(d); + await d.tick(); + expect(statusOf(spawn)).toBe("queued"); // the held attempt still occupies the slot + + store.database.prepare("UPDATE dispatch_tasks SET held_until = 1 WHERE id = ?").run(id); + await d.tick(); + expect(statusOf(id)).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["reclaimed"]); + expect(statusOf(spawn)).toBe("running"); + }); + + test("the hold window comes from config", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host, { holdMaxMs: 5_000 }); + const before = Date.now(); + const { id } = await heldAttempt(d, host); + const until_ = store.dispatchGet(id)!.heldUntil!; + expect(until_).toBeGreaterThanOrEqual(before + 5_000); + expect(until_).toBeLessThanOrEqual(Date.now() + 5_000); + }); + + test("cancel of a held attempt settles it `failed` and fences its worker", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id, gen } = await heldAttempt(d, host); + expect(d.cancel(id, "run cancelled by owner")).toBe(true); + expect(statusOf(id)).toBe("failed"); + expect(store.dispatchGet(id)!.error).toBe("run cancelled by owner"); + expect(host.settled.map((s) => s.status)).toEqual(["failed"]); + expect(await d.unhold(id, gen)).toBe(false); + expect(d.cancel(id, "again")).toBe(false); + expect(host.settled).toHaveLength(1); + }); + + test("cancel of a running attempt drops the watch, so its late turn is ignored", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + await d.tick(); + expect(d.cancel(id, "stop")).toBe(true); + expect(d.tasksForSession("attempt-worker-1")).toEqual([]); + d.onSessionStatus("attempt-worker-1", "idle"); + await settle(); + expect(statusOf(id)).toBe("failed"); + expect(host.settled.map((s) => s.status)).toEqual(["failed"]); + }); + + test("hold, unhold and cancel are scoped to the owning daemon", async () => { + const host = new LoopHost(); + const mine = makeDispatcher(host, { daemonId: "host:A" }); + const other = makeDispatcher(host, { daemonId: "host:B" }); + const id = enqueueAttempt(mine); + await mine.tick(); + const gen = store.dispatchGet(id)!.generation; + + expect(other.hold(id, gen, "not yours")).toBe(false); + expect(mine.hold(id, gen, "approval")).toBe(true); + expect(await other.unhold(id, gen)).toBe(false); + expect(other.cancel(id, "not yours")).toBe(false); + expect(statusOf(id)).toBe("held"); + expect(host.settled).toEqual([]); + + expect(mine.cancel(id, "mine")).toBe(true); + expect(statusOf(id)).toBe("failed"); }); }); @@ -494,17 +749,16 @@ describe("dispatcher — attempt tasks", () => { expect(host.attemptSpawned.map((t) => t.attemptId)).toEqual(["attempt-1"]); d.onSessionStatus("attempt-worker-1", "idle"); - await Bun.sleep(10); + await until(() => host.settled.length > 0, "settled"); await d.tick(); - expect(store.dispatchGet(id)!.status).toBe("done"); + expect(statusOf(id)).toBe("done"); expect(host.settled.map((s) => s.status)).toEqual(["done"]); expect(host.settled[0]!.task.status).toBe("done"); expect(host.destroyed).toEqual([]); expect(allEvents(host)).toEqual([]); - d.stop(); }); - test("an approval wedge reports to the loop, not the conductor, and the loop can hold it", async () => { + test("an approval wedge reports to the loop, not the conductor", async () => { const host = new LoopHost(); const d = makeDispatcher(host); const id = enqueueAttempt(d); @@ -513,15 +767,7 @@ describe("dispatcher — attempt tasks", () => { await d.tick(); expect(host.waiting.map((t) => t.id)).toEqual([id]); expect(allEvents(host)).toEqual([]); - expect(store.dispatchGet(id)!.status).toBe("running"); // not failed - - const waiting = host.waiting[0]!; - expect(d.hold(waiting.id, waiting.generation, "approval outside the run's profile")).toBe(true); - expireLease(id); - await makeDispatcher(host).tick(); - expect(store.dispatchGet(id)!.status).toBe("held"); - expect(host.settled).toEqual([]); - d.stop(); + expect(statusOf(id)).toBe("running"); // not failed }); test("a reclaim is terminal and reported as `reclaimed`; the queue never re-runs it", async () => { @@ -532,12 +778,27 @@ describe("dispatcher — attempt tasks", () => { expireLease(id); await d.tick(); await d.tick(); - expect(store.dispatchGet(id)!.status).toBe("failed"); + expect(statusOf(id)).toBe("failed"); expect(host.settled.map((s) => s.status)).toEqual(["reclaimed"]); expect(host.attemptSpawned).toHaveLength(1); expect(host.destroyed).toEqual([]); expect(allEvents(host)).toEqual([]); - d.stop(); + }); + + test("a worker started after its attempt was reclaimed is reported, not orphaned", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const id = enqueueAttempt(d); + host.duringAttemptSpawn = () => { + // The start outlives the lease: the attempt is reclaimed mid-spawn. + expireLease(id); + store.dispatchReclaimStale("boot-other", LEASE_MS, Date.now()); + }; + await d.tick(); + expect(statusOf(id)).toBe("failed"); + expect(host.orphans).toEqual(["attempt-worker-1"]); + expect(host.destroyed).toEqual([]); + expect(d.tasksForSession("attempt-worker-1")).toEqual([]); }); test("a worker error settles the attempt as `failed` without a retry", async () => { @@ -546,13 +807,12 @@ describe("dispatcher — attempt tasks", () => { const id = enqueueAttempt(d); await d.tick(); d.onSessionStatus("attempt-worker-1", "error"); - await Bun.sleep(10); + await until(() => host.settled.length > 0, "settled"); await d.tick(); - expect(store.dispatchGet(id)!.status).toBe("failed"); + expect(statusOf(id)).toBe("failed"); expect(host.settled.map((s) => s.status)).toEqual(["failed"]); expect(host.attemptSpawned).toHaveLength(1); expect(allEvents(host)).toEqual([]); - d.stop(); }); test("fails closed on a host without loop support — never runs as an ordinary spawn", async () => { @@ -565,25 +825,29 @@ describe("dispatcher — attempt tasks", () => { expect(row.error).toMatch(/no loop host/); expect(host.spawned).toEqual([]); expect(allEvents(host)).toEqual([]); - d.stop(); }); test("attempts count against the per-tenant worker cap", async () => { const host = new LoopHost(); - const d = new Dispatcher(store, host, { - tickMs: 999_999, - leaseMs: LEASE_MS, - failureLimit: 3, - maxConcurrentWorkers: 1, - workerToolBudget: 10, - retryBaseMs: 0, - }); + const d = makeDispatcher(host, { maxConcurrentWorkers: 1 }); enqueueAttempt(d); const spawn = enqueueSpawn(d); await d.tick(); expect(host.attemptSpawned).toHaveLength(1); expect(host.spawned).toEqual([]); - expect(store.dispatchGet(spawn)!.status).toBe("queued"); - d.stop(); + expect(statusOf(spawn)).toBe("queued"); + }); + + test("attempts stay off the conductor's board and its deltas", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + enqueueAttempt(d); + const spawn = enqueueSpawn(d); + await d.tick(); + expect(store.dispatchListForTenant(TENANT.accountId, TENANT.projectId).map((t) => t.id)).toEqual([spawn]); + expect(store.dispatchTasksUpdatedSince(TENANT.accountId, TENANT.projectId, 0).map((t) => t.id)).toEqual([ + spawn, + ]); + expect(store.dispatchStatusCounts(TENANT.accountId, TENANT.projectId).active).toBe(1); }); }); diff --git a/web/src/components/SessionListPane.tsx b/web/src/components/SessionListPane.tsx index b832b657..f2ded95f 100644 --- a/web/src/components/SessionListPane.tsx +++ b/web/src/components/SessionListPane.tsx @@ -676,10 +676,12 @@ const PanelProgress: Component<{ /** A child's position and state within the live fan-out. */ const PanelMemberBadge: Component<{ ordinal: number; - status: "queued" | "claimed" | "running" | "done" | "failed" | "blocked"; + status: "queued" | "claimed" | "running" | "held" | "done" | "failed" | "blocked"; }> = (props) => { const look = () => { switch (props.status) { + case "held": + return { cls: "border-warn/50 bg-warn/10 text-warn", note: "waiting on you" }; case "running": case "claimed": return { cls: "border-warn/50 bg-warn/10 text-warn animate-pulse", note: "working" }; diff --git a/web/src/lib/attention.test.ts b/web/src/lib/attention.test.ts index 9a950caf..8b165860 100644 --- a/web/src/lib/attention.test.ts +++ b/web/src/lib/attention.test.ts @@ -64,6 +64,26 @@ describe("collectAttention", () => { expect(items.some((i) => i.key.includes("t3"))).toBe(false); }); + it("lists a held task as an approval, once — not again beside its session's own", () => { + const wedged = session("w", { status: "waiting_approval" }); + const quiet = session("q"); + const items = collectAttention( + sources({ + sessions: [wedged, quiet], + tasks: [ + task("held-1", { status: "held", workerSessionId: "w", error: "approval outside profile" }), + task("held-2", { status: "held", workerSessionId: "q" }), + ], + taskSession: (t) => (t.workerSessionId === "w" ? wedged : t.workerSessionId === "q" ? quiet : null), + }), + NOW, + ); + expect(items.map((i) => [i.key, i.kind])).toEqual([ + ["approval:w", "approval"], + ["task:held-2", "approval"], + ]); + }); + it("does not double-count one stoppage as both a dialog and an approval", () => { // A session showing a provider dialog is usually ALSO waiting_approval. // Counting both would tell the operator two things need them when one does. diff --git a/web/src/lib/attention.ts b/web/src/lib/attention.ts index a78c3ae8..b906a267 100644 --- a/web/src/lib/attention.ts +++ b/web/src/lib/attention.ts @@ -125,8 +125,10 @@ export function collectAttention(src: AttentionSources, now: number): AttentionI // listed for that session: they are the same stoppage, and showing both // would double-count one interruption in a queue whose whole job is to say // how many things need you. + const approvals = new Set(); for (const s of src.sessions) { if (s.status !== "waiting_approval" || asked.has(s.id)) continue; + approvals.add(s.id); items.push({ key: `approval:${s.id}`, kind: "approval", @@ -139,8 +141,23 @@ export function collectAttention(src: AttentionSources, now: number): AttentionI // Stopped dispatch tasks. for (const task of src.tasks) { - if (task.status !== "blocked" && task.status !== "failed") continue; const session = src.taskSession(task); + if (task.status === "held") { + // Parked on a decision outside the worker (a loop's out-of-profile + // approval). Same stoppage as its session's approval when that is + // already listed, so it is not counted twice. + if (session && (asked.has(session.id) || approvals.has(session.id))) continue; + items.push({ + key: `task:${task.id}`, + kind: "approval", + sessionId: session?.id ?? null, + label: session?.name ?? `${task.kind} ${task.id.slice(0, 8)}`, + detail: task.error ?? "held: waiting on a decision", + since: task.createdAt, + }); + continue; + } + if (task.status !== "blocked" && task.status !== "failed") continue; items.push({ key: `task:${task.id}`, kind: task.status, diff --git a/web/src/lib/fleet-lanes.test.ts b/web/src/lib/fleet-lanes.test.ts index bf71e6a2..9a73f590 100644 --- a/web/src/lib/fleet-lanes.test.ts +++ b/web/src/lib/fleet-lanes.test.ts @@ -48,6 +48,12 @@ describe("nodeState", () => { expect(nodeState(task("t", { status: "claimed" }), session("idle"))).toBe("idle"); }); + it("files a held task under awaiting — it is parked on a human, whatever its agent shows", () => { + expect(nodeState(task("t", { status: "held" }), session("idle"))).toBe("awaiting"); + expect(nodeState(task("t", { status: "held" }), null)).toBe("awaiting"); + expect(laneFor("awaiting", task("t", { status: "held" }))).toBe("needs-you"); + }); + it("treats a session error as a failure of the task", () => { expect(nodeState(task("t"), session("error"))).toBe("failed"); }); diff --git a/web/src/lib/fleet-lanes.ts b/web/src/lib/fleet-lanes.ts index ad800bc5..eb148723 100644 --- a/web/src/lib/fleet-lanes.ts +++ b/web/src/lib/fleet-lanes.ts @@ -83,6 +83,10 @@ export function nodeState(task: FleetTaskWire, session: SessionInfo | null): Fle return "done"; case "queued": return "queued"; + case "held": + // Parked on something outside the worker — an approval the run's profile + // does not cover. That is a human's decision, whatever the session says. + return "awaiting"; case "claimed": case "running": { if (session === null) { @@ -96,9 +100,16 @@ export function nodeState(task: FleetTaskWire, session: SessionInfo | null): Fle if (session.status === "error") return "failed"; return BUSY.has(session.status) ? "working" : "idle"; } + default: + return assertNever(task.status); } } +/** Compile-time exhaustiveness: a new dispatch status must be classified here. */ +function assertNever(status: never): never { + throw new Error(`unclassified dispatch status: ${String(status)}`); +} + /** The four triage lanes of §4, in display order. */ export type FleetLane = "needs-you" | "working" | "review" | "done"; From b392d41e5c0158e817e0e14a5ddfe24f4930153c Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 25 Oct 2026 08:36:00 +0800 Subject: [PATCH 3/3] fix: unhold reports what it did; holdMaxMs is configurable and defaults to 2h - Dispatcher.unhold returns an UnholdOutcome ("resumed" | "waiting" | "finished" | "settled" | "refused") instead of a boolean that was true only when the attempt ended up running and watched again. - dispatch.holdMaxMs added to DispatchSchema, the CodeoidConfig type and the settings manifest; the default drops from 24h to 2h, since a held attempt occupies a worker slot. - Dispatcher.hold documents that unhold, cancel or expiry is mandatory: a held worker is unwatched. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/config.ts | 9 +++++ src/daemon/dispatch.ts | 50 +++++++++++++++++++------- src/daemon/settings/manifest.ts | 1 + src/tests/dispatch-generations.test.ts | 30 ++++++++++++---- 4 files changed, 70 insertions(+), 20 deletions(-) diff --git a/src/config.ts b/src/config.ts index 445ab225..8aac3cfe 100644 --- a/src/config.ts +++ b/src/config.ts @@ -608,6 +608,12 @@ const DispatchSchema = z workerToolBudget: z.number().int().min(1).default(50), /** Base retry backoff (ms) for retryable failures — doubles per attempt, capped at leaseMs. */ retryBaseMs: z.number().int().min(0).default(15_000), + /** + * Longest a loop attempt may stay `held` on an approval (ms). A held attempt + * keeps its worker session and occupies a `maxConcurrentWorkers` slot, so + * past this the hold lapses and the attempt settles as reclaimed. + */ + holdMaxMs: z.number().int().min(60_000).default(2 * 60 * 60_000), }) .default({ enabled: true, @@ -617,6 +623,7 @@ const DispatchSchema = z maxConcurrentWorkers: 2, workerToolBudget: 50, retryBaseMs: 15_000, + holdMaxMs: 2 * 60 * 60_000, }); /** @@ -1348,6 +1355,8 @@ export interface CodeoidConfig { maxConcurrentWorkers: number; workerToolBudget: number; retryBaseMs: number; + /** Longest a loop attempt may stay held (ms). Optional for hand-built configs; default 2h. */ + holdMaxMs?: number; }; /** * Collaborative-session bounds (see CollaborationSchema). Optional in the diff --git a/src/daemon/dispatch.ts b/src/daemon/dispatch.ts index 1ea78c71..9a991ac9 100644 --- a/src/daemon/dispatch.ts +++ b/src/daemon/dispatch.ts @@ -61,9 +61,10 @@ export interface DispatchConfig { */ retryBaseMs: number; /** - * Longest a loop attempt may stay `held`. A held attempt keeps its worker - * session and counts against `maxConcurrentWorkers`; past this the hold is - * treated as dead and the attempt settles `reclaimed`, freeing the slot. + * Longest a loop attempt may stay `held` (default 2h). A held attempt keeps + * its worker session and counts against `maxConcurrentWorkers`; past this the + * hold is treated as dead and the attempt settles `reclaimed`, freeing the + * slot. */ holdMaxMs: number; /** @@ -91,7 +92,7 @@ export const DEFAULT_DISPATCH_CONFIG: DispatchConfig = { maxConcurrentWorkers: 2, workerToolBudget: 50, retryBaseMs: 15_000, - holdMaxMs: 24 * 60 * 60_000, + holdMaxMs: 2 * 60 * 60_000, }; /** Thrown by host methods when retrying can never succeed (target gone, bad input). */ @@ -100,6 +101,21 @@ export class NonRetryableDispatchError extends Error {} /** The hooks a host must implement before an attempt task may run on it. */ const ATTEMPT_HOOKS = "spawnAttemptWorker / onAttemptSettled / onAttemptWaiting are not implemented"; +/** + * What `Dispatcher.unhold` did with a held attempt. + * + * - `resumed` — running again, worker working, watched. + * - `waiting` — running again and watched, but the worker is still parked on + * an approval; `onAttemptWaiting` has fired again. + * - `finished` — the worker's turn ended while held; the attempt was settled + * from it (`onAttemptSettled` with `done` or `failed`). + * - `settled` — could not be resumed and was ended instead: the hold outlived + * its boot (`reclaimed`) or the worker session is gone (`failed`). + * - `refused` — nothing changed: not held, a stale generation, or another + * daemon's attempt. + */ +export type UnholdOutcome = "resumed" | "waiting" | "finished" | "settled" | "refused"; + /** How an attempt task ended, as reported to `DispatcherHost.onAttemptSettled`. */ export type AttemptSettledStatus = "done" | "failed" | "reclaimed"; @@ -441,6 +457,12 @@ export class Dispatcher { * The watch is dropped while held — a status change is not this * dispatcher's to act on until `unhold`, which re-reads the worker's state * rather than relying on a transition it missed. + * + * The caller MUST release a hold — `unhold` once the decision is made, or + * `cancel` — or leave it to lapse. A held worker is unwatched: its turn + * ending, erroring or vanishing goes unnoticed, and the attempt keeps its + * worker slot, until one of those happens. `holdMaxMs` (default 2h) is the + * backstop, not the plan. */ hold(taskId: string, generation: number, reason: string): boolean { const now = Date.now(); @@ -469,7 +491,8 @@ export class Dispatcher { /** * Resume a held attempt as `running` with a fresh lease, at the same * generation — the same worker carries on, so its result still applies. - * Returns true only when the attempt is running again. + * Returns what happened (`UnholdOutcome`); only `resumed` and `waiting` leave + * the attempt running. * * A hold placed by a previous boot is NOT resumed: the approval it waited on * died with that process, and the worker's idle would read as a success it @@ -480,11 +503,11 @@ export class Dispatcher { * finish now; still waiting on an approval → tell the loop again; working → * watch it. */ - async unhold(taskId: string, generation: number): Promise { + async unhold(taskId: string, generation: number): Promise { const task = this.#store.dispatchGet(taskId); if (!task || task.status !== "held" || task.generation !== generation) { this.#staleResult(taskId, generation, "unhold"); - return false; + return "refused"; } const now = Date.now(); if (task.claimOwner !== this.bootId) { @@ -497,15 +520,15 @@ export class Dispatcher { ); if (!settled) { this.#staleResult(taskId, generation, "unhold"); - return false; + return "refused"; } this.#settleAttempt(settled, "reclaimed"); this.#signalBoardChange(); - return false; + return "settled"; } if (!this.#store.dispatchUnhold(taskId, generation, this.bootId, now, this.#config.daemonId)) { this.#staleResult(taskId, generation, "unhold"); - return false; + return "refused"; } this.#host.audit("dispatch.unheld", `task=${taskId} generation=${generation}`); this.#signalBoardChange(); @@ -522,11 +545,11 @@ export class Dispatcher { ) { this.#settleAttempt(this.#store.dispatchGet(taskId) ?? task, "failed"); } - return false; + return "settled"; } if (status === "idle" || status === "error") { await this.#finishWorkerTask(taskId, generation, sessionId, status); - return true; + return "finished"; } this.#watch(sessionId, taskId, generation); if (status === "waiting_approval") { @@ -534,8 +557,9 @@ export class Dispatcher { // report it happened while unwatched, so report it now. const fresh = this.#store.dispatchGet(taskId) ?? task; this.#callAttemptHook("onAttemptWaiting", () => this.#host.onAttemptWaiting?.(fresh)); + return "waiting"; } - return true; + return "resumed"; } /** diff --git a/src/daemon/settings/manifest.ts b/src/daemon/settings/manifest.ts index 91c03f58..d7f55db2 100644 --- a/src/daemon/settings/manifest.ts +++ b/src/daemon/settings/manifest.ts @@ -356,6 +356,7 @@ const fleet: SettingsTab = { cfg("dispatch.tickMs", "Tick interval (ms)", "Dispatcher claim/reclaim/deliver cadence.", { kind: "int", min: 250, default: 5000, advanced: true }), cfg("dispatch.leaseMs", "Claim lease (ms)", "An unrenewed claim past this is reclaimed.", { kind: "int", min: 10000, default: 600000, advanced: true }), cfg("dispatch.retryBaseMs", "Retry backoff base (ms)", "Base backoff for retryable failures — doubles per attempt.", { kind: "int", min: 0, default: 15000, advanced: true }), + cfg("dispatch.holdMaxMs", "Approval hold limit (ms)", "Longest a loop attempt may wait on an approval while keeping its worker slot.", { kind: "int", min: 60000, default: 7200000, advanced: true }), ], }, { diff --git a/src/tests/dispatch-generations.test.ts b/src/tests/dispatch-generations.test.ts index 12f23627..86e77ca2 100644 --- a/src/tests/dispatch-generations.test.ts +++ b/src/tests/dispatch-generations.test.ts @@ -14,8 +14,10 @@ import { Database } from "bun:sqlite"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { loadConfig } from "../config.js"; import { type AttemptSettledStatus, + DEFAULT_DISPATCH_CONFIG, Dispatcher, type DispatcherHost, } from "../daemon/dispatch.js"; @@ -584,7 +586,7 @@ describe("dispatcher — held attempts", () => { expect(statusOf(id)).toBe("held"); host.statuses.set("attempt-worker-1", "thinking"); - expect(await d.unhold(id, gen)).toBe(true); + expect(await d.unhold(id, gen)).toBe("resumed"); expect(d.tasksForSession("attempt-worker-1")).toEqual([id]); d.onSessionStatus("attempt-worker-1", "idle"); await until(() => host.settled.length > 0, "settled"); @@ -608,7 +610,7 @@ describe("dispatcher — held attempts", () => { const { id, gen } = await heldAttempt(d, host); host.statuses.set("attempt-worker-1", "idle"); // the turn ended while the daemon was down const restarted = makeDispatcher(host); - expect(await restarted.unhold(id, gen)).toBe(false); + expect(await restarted.unhold(id, gen)).toBe("settled"); expect(statusOf(id)).toBe("failed"); expect(host.settled.map((s) => s.status)).toEqual(["reclaimed"]); }); @@ -623,7 +625,7 @@ describe("dispatcher — held attempts", () => { d.onSessionStatus("attempt-worker-1", "idle"); expect(statusOf(id)).toBe("held"); - expect(await d.unhold(id, gen)).toBe(true); + expect(await d.unhold(id, gen)).toBe("finished"); expect(statusOf(id)).toBe("done"); expect(host.settled.map((s) => s.status)).toEqual(["done"]); }); @@ -633,7 +635,7 @@ describe("dispatcher — held attempts", () => { const d = makeDispatcher(host); const { id, gen } = await heldAttempt(d, host); host.statuses.delete("attempt-worker-1"); - expect(await d.unhold(id, gen)).toBe(false); + expect(await d.unhold(id, gen)).toBe("settled"); expect(statusOf(id)).toBe("failed"); expect(host.settled.map((s) => s.status)).toEqual(["failed"]); }); @@ -643,7 +645,7 @@ describe("dispatcher — held attempts", () => { const d = makeDispatcher(host); const { id, gen } = await heldAttempt(d, host); expect(host.waiting).toHaveLength(1); - expect(await d.unhold(id, gen)).toBe(true); + expect(await d.unhold(id, gen)).toBe("waiting"); expect(host.waiting).toHaveLength(2); expect(host.waiting[1]!.status).toBe("running"); expect(d.tasksForSession("attempt-worker-1")).toEqual([id]); @@ -674,6 +676,20 @@ describe("dispatcher — held attempts", () => { expect(until_).toBeLessThanOrEqual(Date.now() + 5_000); }); + test("the default hold window is 2 hours, in the dispatcher and the config schema", () => { + expect(DEFAULT_DISPATCH_CONFIG.holdMaxMs).toBe(2 * 60 * 60_000); + const cfg = loadConfig({ configPath: join(tmp, "absent.json"), env: {} }); + expect(cfg.dispatch?.holdMaxMs).toBe(2 * 60 * 60_000); + }); + + test("a stale generation is refused without touching the hold", async () => { + const host = new LoopHost(); + const d = makeDispatcher(host); + const { id, gen } = await heldAttempt(d, host); + expect(await d.unhold(id, gen + 1)).toBe("refused"); + expect(statusOf(id)).toBe("held"); + }); + test("cancel of a held attempt settles it `failed` and fences its worker", async () => { const host = new LoopHost(); const d = makeDispatcher(host); @@ -682,7 +698,7 @@ describe("dispatcher — held attempts", () => { expect(statusOf(id)).toBe("failed"); expect(store.dispatchGet(id)!.error).toBe("run cancelled by owner"); expect(host.settled.map((s) => s.status)).toEqual(["failed"]); - expect(await d.unhold(id, gen)).toBe(false); + expect(await d.unhold(id, gen)).toBe("refused"); expect(d.cancel(id, "again")).toBe(false); expect(host.settled).toHaveLength(1); }); @@ -710,7 +726,7 @@ describe("dispatcher — held attempts", () => { expect(other.hold(id, gen, "not yours")).toBe(false); expect(mine.hold(id, gen, "approval")).toBe(true); - expect(await other.unhold(id, gen)).toBe(false); + expect(await other.unhold(id, gen)).toBe("refused"); expect(other.cancel(id, "not yours")).toBe(false); expect(statusOf(id)).toBe("held"); expect(host.settled).toEqual([]);