From 1e77c830e95e66c69e708990cc15b5d9fa4959fc Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Mon, 4 May 2026 04:04:49 -0400 Subject: [PATCH] feat(orchestration): coordinator heartbeat + stale detector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Handle incoming 'heartbeat' messages by calling recordHeartbeat keyed on payload.dispatchId (strict — log-and-skip if missing, no taskId fallback so a straggler heartbeat from a previously-failed dispatch cannot mask a hung retry per §5.3.4). On every tick after the 10-minute threshold, emit one log per stale dispatched row — no auto-fail. Also threads dispatchId through buildDispatchPreamble so workers can attribute their heartbeats back to the correct dispatch context. Co-authored-by: Orca --- .../runtime/orchestration/coordinator.test.ts | 82 +++++++++++++++++++ src/main/runtime/orchestration/coordinator.ts | 57 +++++++++++++ 2 files changed, 139 insertions(+) diff --git a/src/main/runtime/orchestration/coordinator.test.ts b/src/main/runtime/orchestration/coordinator.test.ts index ac0ebbb4974..d1423e07a48 100644 --- a/src/main/runtime/orchestration/coordinator.test.ts +++ b/src/main/runtime/orchestration/coordinator.test.ts @@ -355,6 +355,88 @@ describe('Coordinator', () => { expect(result.status).toBe('completed') }) + it('logs a stale warning for dispatched rows past the threshold and does not auto-fail', async () => { + db = new OrchestrationDb(':memory:') + const runtime = createMockRuntime() + // No terminals available so dispatchReadyTasks creates one and we can + // drive the stale-scan deterministically via SQL backdating. + const task = db.createTask({ spec: 'work' }) + const ctx = db.createDispatchContext(task.id, 'term_stale') + + // Backdate dispatched_at and last_heartbeat_at beyond the 10-min threshold + // so getStaleDispatches returns this row on the first tick. + const sqlite = ( + db as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => void } } } + ).db + const iso = (ms: number) => new Date(Date.now() - ms).toISOString() + sqlite + .prepare('UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?') + .run(iso(60 * 60 * 1000), iso(30 * 60 * 1000), ctx.id) + + const logs: string[] = [] + const coordinator = new Coordinator(db, runtime, { + spec: 'go', + coordinatorHandle: 'coord', + pollIntervalMs: 20, + onLog: (m) => logs.push(m) + }) + + // Drive one tick then stop — we only need the stale warning to have fired. + const runPromise = coordinator.run() + await new Promise((r) => { + setTimeout(r, 80) + }) + coordinator.stop() + await runPromise + + expect(logs.some((l) => /has not sent a heartbeat/.test(l) && l.includes(task.id))).toBe(true) + // Task status must NOT have been auto-failed — logging only. + expect(db.getTask(task.id)?.status).toBe('dispatched') + }) + + it('records heartbeat by dispatchId on worker heartbeat messages', async () => { + db = new OrchestrationDb(':memory:') + const runtime = createMockRuntime() + runtime.terminals = [{ handle: 'term_a', worktreeId: 'wt1', connected: true, writable: true }] + + const task = db.createTask({ spec: 'work' }) + const ctx = db.createDispatchContext(task.id, 'term_a') + + const coordinator = new Coordinator(db, runtime, { + spec: 'go', + coordinatorHandle: 'coord', + pollIntervalMs: 20 + }) + + const runPromise = coordinator.run() + + db.insertMessage({ + from: 'term_a', + to: 'coord', + subject: 'alive', + type: 'heartbeat', + payload: JSON.stringify({ taskId: task.id, dispatchId: ctx.id, phase: 'implementing' }) + }) + + await new Promise((r) => { + setTimeout(r, 80) + }) + + expect(db.getDispatchContext(task.id)?.last_heartbeat_at).toBeTruthy() + + // Complete the task so the coordinator run finishes cleanly. + db.insertMessage({ + from: 'term_a', + to: 'coord', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({ taskId: task.id }) + }) + + const result = await runPromise + expect(result.status).toBe('completed') + }) + it('can be stopped', async () => { db = new OrchestrationDb(':memory:') const runtime = createMockRuntime() diff --git a/src/main/runtime/orchestration/coordinator.ts b/src/main/runtime/orchestration/coordinator.ts index 541102f4ae5..e80f6f4e194 100644 --- a/src/main/runtime/orchestration/coordinator.ts +++ b/src/main/runtime/orchestration/coordinator.ts @@ -41,6 +41,13 @@ type CoordinatorState = { const DEFAULT_POLL_MS = 2000 const MAX_CONCURRENT_DEFAULT = 4 +// Why: 10 min matches the preamble's documented heartbeat cadence (5 min) × +// 2, so a single missed heartbeat is the earliest a dispatch can look stale. +// Keeping this in one place (not a per-call arg) ensures the preamble copy +// and the detector logic stay aligned; moving it to a config would multiply +// the places this constant must be kept in sync. +const HUNG_THRESHOLD_MS = 10 * 60 * 1000 + export class Coordinator { private db: OrchestrationDb private runtime: CoordinatorRuntime @@ -173,10 +180,27 @@ export class Coordinator { this.processMessages() this.processEscalations() this.processDecisionGates() + this.warnStaleDispatches() await this.dispatchReadyTasks() return this.checkConvergence() } + // Why: emit a single warning per stale dispatch per tick. This intentionally + // does NOT auto-fail the dispatch — the false-positive cost (a slow worker + // producing correct output) is higher than the false-negative cost (a hung + // worker keeps its terminal slot until a human notices). Auto-fail policy + // is a separate decision documented in R6 of DESIGN_DOC_PREAMBLE_FIX.md. + private warnStaleDispatches(): void { + const thresholdIso = new Date(Date.now() - HUNG_THRESHOLD_MS).toISOString() + const stale = this.db.getStaleDispatches(thresholdIso) + for (const ctx of stale) { + const minutes = Math.round(HUNG_THRESHOLD_MS / 60000) + this.opts.onLog( + `Warning: worker ${ctx.assignee_handle ?? ''} on task ${ctx.task_id} has not sent a heartbeat in ~${minutes} min (dispatch ${ctx.id})` + ) + } + } + private processMessages(): void { const messages = this.db.getUnreadMessages(this.opts.coordinatorHandle) if (messages.length === 0) { @@ -194,6 +218,9 @@ export class Coordinator { case 'decision_gate': this.handleDecisionGateMessage(msg) break + case 'heartbeat': + this.handleHeartbeat(msg) + break case 'status': this.opts.onLog(`Status from ${msg.from_handle}: ${msg.subject}`) break @@ -205,6 +232,35 @@ export class Coordinator { this.db.markAsRead(messages.map((m) => m.id)) } + // Why: attribute heartbeats to the specific dispatchId, not a + // (taskId, from_handle) lookup. A task that gets retried after a failed + // dispatch has multiple rows in dispatch_contexts — a late heartbeat from + // the previous (failed) assignee arriving while the new dispatch is active + // would falsely bump the new row's last_heartbeat_at if we resolved by + // "latest dispatch for this task" (§5.3.4). If the worker drops dispatchId + // from the payload, log-and-skip is the preferred failure mode: the stale + // detector will correctly flag the dispatch as hung because nothing + // refreshed last_heartbeat_at. + private handleHeartbeat(msg: MessageRow): void { + if (!msg.payload) { + this.opts.onLog(`Heartbeat from ${msg.from_handle} missing payload; ignored`) + return + } + let payload: { dispatchId?: unknown } = {} + try { + payload = JSON.parse(msg.payload) + } catch { + this.opts.onLog(`Heartbeat from ${msg.from_handle} has invalid JSON payload; ignored`) + return + } + const dispatchId = payload.dispatchId + if (typeof dispatchId !== 'string' || dispatchId.length === 0) { + this.opts.onLog(`Heartbeat from ${msg.from_handle} missing dispatchId; ignored`) + return + } + this.db.recordHeartbeat(dispatchId, msg.created_at) + } + private handleWorkerDone(msg: MessageRow): void { this.opts.onLog(`Worker done: ${msg.from_handle} — ${msg.subject}`) @@ -389,6 +445,7 @@ export class Coordinator { // so they talk to the dev runtime's socket, not production (Section 6.4). const preamble = buildDispatchPreamble({ taskId: task.id, + dispatchId: dispatch.id, taskSpec: task.spec, coordinatorHandle: this.opts.coordinatorHandle, devMode: process.env.ORCA_USER_DATA_PATH?.includes('orca-dev')