From 1efd629ea571c0d3d720d9ecb275fdc74bd51e74 Mon Sep 17 00:00:00 2001 From: zaridan <1617679+zaridan@users.noreply.github.com> Date: Tue, 23 Jun 2026 18:34:54 -0700 Subject: [PATCH 1/2] feat(orchestration): coordinator resume-on-boot (F3, #14) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Coordinator is an in-memory instance with no boot hook, so after an app restart a leftover `coordinator_runs.status='running'` row is a zombie: it trips F1's per-target active-run guard and blocks a fresh run for that target while nothing drives the old one. F3 adds a startup reconciler that, on boot, scans every running run and converges it. MUST (the floor — zombie fix): - `reconcileCoordinatorRunsOnBoot` (boot-resume.ts) classifies each running run: finalize (work already done → completed/failed), resume, or — the floor — mark failed. A run that cannot be resumed always converges to failed, never left running with no loop. Idempotent (a second pass finds no running rows), F1-isolation aware (run-scoped + target_key), wired into desktop boot behind the experimental flag, fire-and-forget so it can never block startup. SHOULD (real resume + worktree re-adoption, per F2-DESIGN §8): - Persist the in-memory-only coordinator options (schema v9: max_concurrent, worktree_backed, worker_agent) so a restart rebuilds the SAME run instead of guessing — a legacy-mode guess of a worktree-backed run would dispatch into a shell that never completes (a fresh zombie). - `resumeCoordinatorRunOnBoot` rebuilds the coordinator, reclaims dead in-flight dispatches so the loop is guaranteed to converge, re-adopts existing track worktrees by scanning the director's lineage children (parentWorktreeId === directorWorktreeId, same run — the data Mission Control uses) via `buildAdoptedTrackWorktrees` + `Coordinator.seedAdoptedTrackWorktrees`, and relaunches the worker agent in the existing checkout rather than forking a new worktree/branch. Declines (→ failed) when the director worktree is gone. Tests: - boot-resume.test.ts: orphaned running run reconciled to failed AND a fresh run for the target then starts (guard unblocked); idempotency; finalize-when-done; resume decision (succeed/decline/throw); target isolation; reclaim; track re-adoption from lineage. - coordinator.test.ts: a seeded track is re-adopted on resume (no duplicate worktree; agent relaunched in the existing checkout). - Hardened a pre-existing ~40%-flaky test ("respects maxConcurrent limit") that completed tasks in a fixed array order while the cap picks 2 of 3 by random id — now completes tasks as actually dispatched and always drains the loop. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/main/index.ts | 18 ++ .../runtime/orchestration/boot-resume.test.ts | 290 ++++++++++++++++++ src/main/runtime/orchestration/boot-resume.ts | 147 +++++++++ .../runtime/orchestration/coordinator.test.ts | 118 ++++++- src/main/runtime/orchestration/coordinator.ts | 23 +- src/main/runtime/orchestration/db.test.ts | 35 +++ src/main/runtime/orchestration/db.ts | 104 ++++++- src/main/runtime/orchestration/types.ts | 9 + .../rpc/methods/orchestration-gates.ts | 132 +++++++- 9 files changed, 836 insertions(+), 40 deletions(-) create mode 100644 src/main/runtime/orchestration/boot-resume.test.ts create mode 100644 src/main/runtime/orchestration/boot-resume.ts diff --git a/src/main/index.ts b/src/main/index.ts index d684845923a..a91303d30cc 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -33,6 +33,7 @@ import { initOnboardingCohortClassifier } from './telemetry/onboarding-cohort-cl import { resolveConsent } from './telemetry/consent' import { triggerStartupNotificationRegistration } from './ipc/notifications' import { OrcaRuntimeService } from './runtime/orca-runtime' +import { runOrchestrationBootReconcile } from './runtime/rpc/methods/orchestration-gates' import { OrcaRuntimeRpcServer } from './runtime/runtime-rpc' import { awaitRuntimeFileWatcherUnsubscribes } from './runtime/orca-runtime-files' import { clearRuntimeMetadataIfOwned } from './runtime/runtime-metadata' @@ -1523,6 +1524,23 @@ app.whenReady().then(async () => { }) logStartupMilestone('services-initialized') + + // Why (F3 #14): the coordinator is in-memory with no boot hook, so a leftover + // `coordinator_runs.status='running'` row from a pre-restart run is a zombie — + // it blocks a fresh run for that target via F1's active-run guard while nothing + // drives it. Reconcile on boot (finalize / resume / fail) so the guard unblocks. + // Gated on the experimental flag (the only surface that starts runs) to stay + // additive, fire-and-forget so a reconcile error can never block startup, and + // run here — before any window is shown — so it lands before a user-triggered + // run could hit the guard. + if (store.getSettings().experimentalOrchestrators) { + void runOrchestrationBootReconcile(runtimeService, (msg) => + console.log('[orchestration-resume]', msg) + ).catch((error) => { + console.error('[orchestration-resume] boot reconcile failed:', error) + }) + } + await ensureMainI18n() await setMainUiLanguage(store.getSettings().uiLanguage) logStartupMilestone('i18n-ready') diff --git a/src/main/runtime/orchestration/boot-resume.test.ts b/src/main/runtime/orchestration/boot-resume.test.ts new file mode 100644 index 00000000000..4128162761f --- /dev/null +++ b/src/main/runtime/orchestration/boot-resume.test.ts @@ -0,0 +1,290 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb, CoordinatorRunConflictError } from './db' +import type { CoordinatorRun } from './types' +import { + buildAdoptedTrackWorktrees, + reconcileCoordinatorRunsOnBoot, + type ReconcileCoordinatorRunsOnBootDeps +} from './boot-resume' +import type { WorktreeLineage } from '../../../shared/types' + +// Why (F3 #14): regression suite for the zombie-director failure. After an app +// restart the in-memory coordinator is gone but its `coordinator_runs.status= +// 'running'` row survives, tripping F1's per-target active-run guard so a fresh +// run for that target is refused while nothing drives the old one. These tests +// pin: (1) an orphaned running run is reconciled to failed AND a fresh run for +// the target then starts; (2) idempotency; (3) finalize-when-done; (4) the +// resume decision; (5) track re-adoption from lineage. + +describe('reconcileCoordinatorRunsOnBoot (F3 #14)', () => { + let db: OrchestrationDb | undefined + + afterEach(() => { + db?.close() + db = undefined + }) + + function createDb(): OrchestrationDb { + db = new OrchestrationDb(':memory:') + return db + } + + it('reconciles an orphaned running run to failed and unblocks a fresh run for the target', async () => { + const d = createDb() + const targetKey = 'worktree:wt-director' + const run = d.startCoordinatorRun({ + spec: 'do work', + coordinatorHandle: 'coordinator-x', + targetKey + }) + // Outstanding work: a ready task owned by the run. + d.createTask({ spec: 'task A', coordinatorRunId: run.id, targetKey }) + + // Without the fix the leftover running row blocks a fresh run for the target. + expect(() => + d.startCoordinatorRun({ spec: 'fresh', coordinatorHandle: 'coordinator-y', targetKey }) + ).toThrow(CoordinatorRunConflictError) + + // The boot hook (no resume callback → MUST-only) reconciles the zombie. + const results = await reconcileCoordinatorRunsOnBoot({ db: d }) + expect(results).toEqual([{ runId: run.id, disposition: 'failed', reason: expect.any(String) }]) + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + + // The guard is now unblocked: a fresh run for the same target starts. + const fresh = d.startCoordinatorRun({ + spec: 'fresh', + coordinatorHandle: 'coordinator-y', + targetKey + }) + expect(fresh.status).toBe('running') + }) + + it('is idempotent: a second reconcile does not re-act', async () => { + const d = createDb() + const targetKey = 'worktree:wt-director' + const run = d.startCoordinatorRun({ spec: 'do work', coordinatorHandle: 'c', targetKey }) + d.createTask({ spec: 'task A', coordinatorRunId: run.id, targetKey }) + + const first = await reconcileCoordinatorRunsOnBoot({ db: d }) + expect(first).toHaveLength(1) + const second = await reconcileCoordinatorRunsOnBoot({ db: d }) + expect(second).toHaveLength(0) + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('finalizes a running run whose tasks are all completed as completed (not a zombie)', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'done', coordinatorHandle: 'c' }) + const task = d.createTask({ spec: 'task A', coordinatorRunId: run.id }) + d.updateTaskStatus(task.id, 'completed') + + const results = await reconcileCoordinatorRunsOnBoot({ db: d }) + expect(results[0].disposition).toBe('finalized-completed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('completed') + }) + + it('finalizes a running run with no tasks as failed', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'empty', coordinatorHandle: 'c' }) + + const results = await reconcileCoordinatorRunsOnBoot({ db: d }) + expect(results[0].disposition).toBe('finalized-failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('finalizes a running run with a failed task as failed', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'mixed', coordinatorHandle: 'c' }) + const a = d.createTask({ spec: 'A', coordinatorRunId: run.id }) + const b = d.createTask({ spec: 'B', coordinatorRunId: run.id }) + d.updateTaskStatus(a.id, 'completed') + d.updateTaskStatus(b.id, 'failed') + + const results = await reconcileCoordinatorRunsOnBoot({ db: d }) + expect(results[0].disposition).toBe('finalized-failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('resumes a run with outstanding work when the resume callback succeeds', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) + d.createTask({ spec: 'A', coordinatorRunId: run.id }) + + const resumed: string[] = [] + const deps: ReconcileCoordinatorRunsOnBootDeps = { + db: d, + resume: async (r: CoordinatorRun) => { + resumed.push(r.id) + return true + } + } + const results = await reconcileCoordinatorRunsOnBoot(deps) + expect(results[0].disposition).toBe('resumed') + expect(resumed).toEqual([run.id]) + // A resumed run legitimately stays running (a live loop drives it). + expect(d.getCoordinatorRun(run.id)?.status).toBe('running') + }) + + it('falls back to failed when the resume callback declines', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) + d.createTask({ spec: 'A', coordinatorRunId: run.id }) + + const results = await reconcileCoordinatorRunsOnBoot({ db: d, resume: async () => false }) + expect(results[0].disposition).toBe('failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('falls back to failed when the resume callback throws (never left running with no loop)', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) + d.createTask({ spec: 'A', coordinatorRunId: run.id }) + + const results = await reconcileCoordinatorRunsOnBoot({ + db: d, + resume: async () => { + throw new Error('resume boom') + } + }) + expect(results[0].disposition).toBe('failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('is target-isolated: only outstanding work on the run is considered (F1)', async () => { + const d = createDb() + // Two concurrent running rows on different targets (F1 allows this). + const runA = d.startCoordinatorRun({ + spec: 'A', + coordinatorHandle: 'ca', + targetKey: 'worktree:wtA' + }) + const runB = d.startCoordinatorRun({ + spec: 'B', + coordinatorHandle: 'cb', + targetKey: 'worktree:wtB' + }) + d.createTask({ spec: 'A-1', coordinatorRunId: runA.id, targetKey: 'worktree:wtA' }) + // runB has no outstanding work → finalized-failed (no tasks), runA → failed. + const results = await reconcileCoordinatorRunsOnBoot({ db: d }) + const byId = Object.fromEntries(results.map((r) => [r.runId, r.disposition])) + expect(byId[runA.id]).toBe('failed') + expect(byId[runB.id]).toBe('finalized-failed') + }) +}) + +describe('reclaimInFlightDispatchesForResume (F3 #14)', () => { + let db: OrchestrationDb | undefined + afterEach(() => { + db?.close() + db = undefined + }) + + it('returns dead in-flight dispatches to ready so the resumed loop converges', () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) + const task = d.createTask({ spec: 'A', coordinatorRunId: run.id }) + d.createDispatchContext(task.id, 'worker_x') + expect(d.getTask(task.id)?.status).toBe('dispatched') + + const reclaimed = d.reclaimInFlightDispatchesForResume(run.id, 'restarted') + expect(reclaimed).toBe(1) + expect(d.getTask(task.id)?.status).toBe('ready') + }) +}) + +describe('buildAdoptedTrackWorktrees (F3 #14, design §8)', () => { + let db: OrchestrationDb | undefined + afterEach(() => { + db?.close() + db = undefined + }) + + function lineage(partial: Partial & { worktreeId: string }): WorktreeLineage { + return { + worktreeInstanceId: `${partial.worktreeId}-inst`, + parentWorktreeId: 'director', + parentWorktreeInstanceId: 'director-inst', + origin: 'orchestration', + capture: { source: 'orchestration-context', confidence: 'explicit' }, + createdAt: 0, + ...partial + } + } + + it('re-adopts only this run lineage children of the director, keyed by track, deduped', () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'c', + targetKey: 'worktree:director', + worktreeBacked: true, + workerAgent: 'claude' + }) + // Lead task of track 'alpha' (declared in spec). + const lead = d.createTask({ spec: 'track: alpha\nimplement X', coordinatorRunId: run.id }) + // A second same-track task (review) — should NOT add a second map entry. + const review = d.createTask({ spec: 'track: alpha\nreview X', coordinatorRunId: run.id }) + + const map = buildAdoptedTrackWorktrees( + d, + d.getCoordinatorRun(run.id)!, + { + leadChild: lineage({ + worktreeId: 'wt-alpha', + orchestrationRunId: run.id, + taskId: lead.id + }), + reviewChild: lineage({ + worktreeId: 'wt-alpha-2', + orchestrationRunId: run.id, + taskId: review.id + }), + otherRun: lineage({ + worktreeId: 'wt-other', + orchestrationRunId: 'run_other', + taskId: lead.id + }), + otherParent: lineage({ + worktreeId: 'wt-elsewhere', + parentWorktreeId: 'someone-else', + orchestrationRunId: run.id, + taskId: lead.id + }) + }, + 'director', + true + ) + + // One track ('alpha'), first child (the lead) wins; cross-run and + // cross-parent children are ignored (F1 isolation + lineage scoping). + expect([...map.keys()]).toEqual(['alpha']) + const entry = map.get('alpha')! + expect(entry.worktreeId).toBe('wt-alpha') + expect(entry.isAgent).toBe(true) + // Sentinel handle (the pre-crash terminal is dead) → relaunch on dispatch. + expect(entry.terminalHandle).toBe('orch-readopt:wt-alpha') + }) + + it('defaults the track key to the task id when the spec declares no track', () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'c', + targetKey: 'worktree:director', + worktreeBacked: true + }) + const task = d.createTask({ spec: 'implement Y', coordinatorRunId: run.id }) + + const map = buildAdoptedTrackWorktrees( + d, + d.getCoordinatorRun(run.id)!, + { + child: lineage({ worktreeId: 'wt-y', orchestrationRunId: run.id, taskId: task.id }) + }, + 'director', + false + ) + expect([...map.keys()]).toEqual([task.id]) + expect(map.get(task.id)?.isAgent).toBe(false) + }) +}) diff --git a/src/main/runtime/orchestration/boot-resume.ts b/src/main/runtime/orchestration/boot-resume.ts new file mode 100644 index 00000000000..c95c215b420 --- /dev/null +++ b/src/main/runtime/orchestration/boot-resume.ts @@ -0,0 +1,147 @@ +// Why (F3 #14): the Coordinator is an in-memory instance with no boot hook, so +// after an app restart a leftover `coordinator_runs.status='running'` row has no +// loop driving it. That zombie trips F1's per-target active-run guard +// (startCoordinatorRun's BEGIN IMMEDIATE check) and blocks a fresh run for that +// target forever. This module reconciles every running run on boot so the guard +// is unblocked: each run is finalized (its work was already done), resumed (a +// converging loop is restarted), or — the floor — marked failed. Hard rule: a run +// that cannot be resumed MUST converge to failed, never be left running with no +// loop. The reconcile is pure orchestration over the DB plus an injected resume +// callback, so it is testable without Electron or a live runtime. +import type { OrchestrationDb } from './db' +import type { CoordinatorRun, TaskRow } from './types' +import { parseTrackFromSpec } from './coordinator' +import type { WorktreeLineage } from '../../../shared/types' + +export type BootRunDisposition = 'finalized-completed' | 'finalized-failed' | 'resumed' | 'failed' + +export type ReconcileRunOnBootResult = { + runId: string + disposition: BootRunDisposition + reason?: string +} + +export type ReconcileCoordinatorRunsOnBootDeps = { + db: OrchestrationDb + // Attempt to resume a run that still has outstanding work. Returns true when a + // converging coordinator loop was started (the run legitimately stays + // 'running' under a live loop). Returns false — or throws — when resume is + // declined or fails, and the reconciler marks the run failed instead. Omitted + // entirely → MUST-only mode: every outstanding run is reconciled to failed. + resume?: (run: CoordinatorRun) => Promise + onLog?: (msg: string) => void +} + +function isTerminal(task: TaskRow): boolean { + return task.status === 'completed' || task.status === 'failed' +} + +// Why: a running row whose tasks are all terminal (or which has no tasks) is not +// a resumable in-flight run — its loop already finished the work (or never had +// any) and only the row outlived the process. Finalize it to its real outcome so +// it stops blocking the guard: failed if any task failed OR there were no tasks +// (a run that never produced a converged DAG is a failure, matching the +// coordinator's own "No tasks found" → failed path); completed otherwise. +export function classifyOutstandingWork( + db: OrchestrationDb, + run: CoordinatorRun +): { outstanding: TaskRow[]; finalizeStatus: 'completed' | 'failed' | null } { + const tasks = db.listTasks({ coordinatorRunId: run.id }) + const outstanding = tasks.filter((task) => !isTerminal(task)) + if (outstanding.length > 0) { + return { outstanding, finalizeStatus: null } + } + const anyFailed = tasks.length === 0 || tasks.some((task) => task.status === 'failed') + return { outstanding, finalizeStatus: anyFailed ? 'failed' : 'completed' } +} + +export async function reconcileCoordinatorRunsOnBoot( + deps: ReconcileCoordinatorRunsOnBootDeps +): Promise { + const { db, resume } = deps + const onLog = deps.onLog ?? (() => {}) + // Snapshot the running rows up front. Idempotent by construction: a second pass + // finds finalized/failed rows gone from this list, and the only rows still + // 'running' belong to a live resumed loop (the production caller runs once). + const runs = db.listCoordinatorRuns({ status: 'running' }) + const results: ReconcileRunOnBootResult[] = [] + + for (const run of runs) { + const { finalizeStatus } = classifyOutstandingWork(db, run) + if (finalizeStatus) { + db.updateCoordinatorRun(run.id, finalizeStatus) + onLog(`Boot reconcile: run ${run.id} had no outstanding work; finalized as ${finalizeStatus}`) + results.push({ + runId: run.id, + disposition: finalizeStatus === 'failed' ? 'finalized-failed' : 'finalized-completed' + }) + continue + } + + if (resume) { + let resumed = false + try { + resumed = await resume(run) + } catch (err) { + onLog(`Boot reconcile: resume of run ${run.id} threw; marking failed (${String(err)})`) + } + if (resumed) { + onLog(`Boot reconcile: resumed run ${run.id}`) + results.push({ runId: run.id, disposition: 'resumed' }) + continue + } + } + + // Floor: cannot/should not resume → converge to failed so the guard unblocks. + const reason = 'coordinator restarted with no resumable loop' + db.updateCoordinatorRun(run.id, 'failed') + onLog(`Boot reconcile: run ${run.id} not resumed; marked failed (${reason})`) + results.push({ runId: run.id, disposition: 'failed', reason }) + } + + return results +} + +// Why (F3 #14, design §8): rebuild the coordinator's in-memory track→worktree map +// from lineage — the SAME `parentWorktreeId === directorWorktreeId` data Mission +// Control keys on — scoped to THIS run (orchestrationRunId) so a different run's +// children on the same director are never adopted. Each child's recorded taskId +// resolves to a task whose track key (spec `track:` hint → default = task id, +// mirroring Coordinator.trackKeyForTask) is the map key; the first child seen for +// a track (its lead task) wins. The seeded terminalHandle is a post-restart +// sentinel — the pre-crash handle is dead — so resolveTrackTerminal relaunches +// the worker agent in the existing checkout instead of trusting a stale handle. +export function buildAdoptedTrackWorktrees( + db: OrchestrationDb, + run: CoordinatorRun, + lineageById: Record, + directorWorktreeId: string, + wantsAgent: boolean +): Map { + const map = new Map() + for (const lineage of Object.values(lineageById)) { + if (lineage.parentWorktreeId !== directorWorktreeId) { + continue + } + if (lineage.orchestrationRunId !== run.id) { + continue + } + if (!lineage.taskId) { + continue + } + const task = db.getTask(lineage.taskId) + if (!task) { + continue + } + const trackKey = parseTrackFromSpec(task.spec).trackKey ?? task.id + if (map.has(trackKey)) { + continue + } + map.set(trackKey, { + worktreeId: lineage.worktreeId, + terminalHandle: `orch-readopt:${lineage.worktreeId}`, + isAgent: wantsAgent + }) + } + return map +} diff --git a/src/main/runtime/orchestration/coordinator.test.ts b/src/main/runtime/orchestration/coordinator.test.ts index 58d332dc6ae..c9e68f0da88 100644 --- a/src/main/runtime/orchestration/coordinator.test.ts +++ b/src/main/runtime/orchestration/coordinator.test.ts @@ -197,6 +197,24 @@ function lineageMapFromCreated(created: CreatedWorktree[]): Record boolean, + { timeoutMs = 3000, stepMs = 5 }: { timeoutMs?: number; stepMs?: number } = {} +): Promise { + const start = Date.now() + while (!condition()) { + if (Date.now() - start > timeoutMs) { + throw new Error('waitFor: condition not met within timeout') + } + await new Promise((r) => { + setTimeout(r, stepMs) + }) + } +} + function insertWorkerDone( db: OrchestrationDb, params: { @@ -486,25 +504,40 @@ describe('Coordinator', () => { }) const runPromise = coordinator.run() - - await new Promise((r) => { - setTimeout(r, 100) - }) - - // Only 2 should be dispatched - const dispatched = db.listTasks({ status: 'dispatched' }) - expect(dispatched.length).toBe(2) - - // Complete all tasks - for (const task of [t1, t2, t3]) { - insertWorkerDone(db, { taskId: task.id }) + try { + // Wait for the cap to bind on a CONDITION rather than a fixed delay: under + // parallel-test load the first tick can land after a fixed sleep, leaving 0 + // dispatched and throwing before `await runPromise` — which would strand the + // loop past db teardown (an unhandled "database is not open"). maxConcurrent + // holds it at exactly 2; assert it does not creep above 2 across a few ticks. + await waitFor(() => db.listTasks({ status: 'dispatched' }).length === 2) await new Promise((r) => { - setTimeout(r, 100) + setTimeout(r, 60) }) - } + expect(db.listTasks({ status: 'dispatched' }).length).toBe(2) + + // Complete tasks as they are ACTUALLY dispatched, not in a fixed array order: + // the cap picks 2 of the 3 by (created_at, id) — which two is not the array + // order — so completing [t1,t2,t3] blindly can target an undispatched task. + const remaining = new Set([t1.id, t2.id, t3.id]) + while (remaining.size > 0) { + await waitFor(() => + db.listTasks({ status: 'dispatched' }).some((task) => remaining.has(task.id)) + ) + for (const task of db.listTasks({ status: 'dispatched' })) { + if (remaining.has(task.id)) { + insertWorkerDone(db, { taskId: task.id }) + remaining.delete(task.id) + } + } + } - const result = await runPromise - expect(result.status).toBe('completed') + const result = await runPromise + expect(result.status).toBe('completed') + } finally { + coordinator.stop() + await runPromise.catch(() => {}) + } }) it('logs a stale warning for dispatched rows past the threshold and does not auto-fail', async () => { @@ -1683,6 +1716,59 @@ allow-stale-base: true` const sent = runtime.sentMessages.find((m) => m.handle === 'term_a') expect(sent!.text).not.toContain('--- BASE DRIFT ---') }) + + // F3 #14 (design §8): on resume-on-boot a worktree-backed run re-adopts its + // existing track worktrees instead of recreating them. seedAdoptedTrackWorktrees + // pre-seeds the track map; the next same-track dispatch must be a HIT (reuse the + // existing checkout) — NO createWorktree, and because the cached handle is a + // post-restart sentinel, the worker agent is relaunched IN the existing worktree. + // Without the seeding this would create a brand-new worktree/branch (a duplicate). + it('re-adopts a seeded track worktree on resume: no duplicate worktree, relaunches the agent', async () => { + db = new OrchestrationDb(':memory:') + const runtime = createMockRuntime() + + const task = db.createTask({ spec: 'track: alpha\nimplement the thing' }) + + const coordinator = new Coordinator(db, runtime, { + spec: 'go', + coordinatorHandle: 'coord', + pollIntervalMs: 20, + worktree: 'director-wt', + worktreeBacked: true, + workerAgent: 'claude' + }) + // Simulate the boot reconciler re-seeding the track map from lineage. + coordinator.seedAdoptedTrackWorktrees([ + [ + 'alpha', + { worktreeId: 'wt-existing', terminalHandle: 'orch-readopt:wt-existing', isAgent: true } + ] + ]) + + const runPromise = coordinator.run() + try { + // Wait for the relaunch+dispatch on a CONDITION (not a fixed delay), then + // converge before asserting — so a slow tick under load can't throw an + // assertion before we await the loop, which would leave it ticking past + // db teardown (an unhandled "database is not open" rejection). + await waitFor(() => runtime.launchAgentTerminalCalls.length === 1) + insertWorkerDone(db, { taskId: task.id, from: 'term_agent_0' }) + const result = await runPromise + expect(result.status).toBe('completed') + + // Re-adopted, not recreated: no new worktree was forked for the track. + expect(runtime.createdWorktrees).toHaveLength(0) + // The dead pre-crash terminal handle forced a relaunch in the SAME checkout. + expect(runtime.launchAgentTerminalCalls[0].worktree).toBe('id:wt-existing') + // The dispatch preamble landed in the relaunched agent terminal. + expect(runtime.sentMessages.some((m) => m.handle === 'term_agent_0')).toBe(true) + } finally { + // Always drain the loop so a failed assertion can never strand a ticking + // coordinator past this test's db teardown. + coordinator.stop() + await runPromise.catch(() => {}) + } + }) }) }) diff --git a/src/main/runtime/orchestration/coordinator.ts b/src/main/runtime/orchestration/coordinator.ts index 866c28208f2..680ea5ab15c 100644 --- a/src/main/runtime/orchestration/coordinator.ts +++ b/src/main/runtime/orchestration/coordinator.ts @@ -237,9 +237,9 @@ export class Coordinator { // First dispatch of a track lazily creates a worktree (the miss path) and caches // it here; later same-track tasks reuse it (the hit path) so review continues // implement's branch (one PR). Keyed by trackKey (spec hint → default = task id). - // TODO(F3 #14): on resume, seed this map by re-discovering existing children of - // the director worktree (parentWorktreeId === directorWorktreeId) instead of - // recreating them — out of scope here, see design §8. + // On resume-on-boot (#14) this map is pre-seeded via seedAdoptedTrackWorktrees so + // a restarted run re-adopts its existing track worktrees (design §8) instead of + // recreating them. private trackWorktrees = new Map< string, { worktreeId: string; terminalHandle: string; isAgent: boolean } @@ -300,6 +300,23 @@ export class Coordinator { return this.executeLoop(runId) } + // Why (F3 #14, design §8): resume-on-boot re-adopts a crashed run's existing + // track worktrees instead of recreating them. The boot reconciler discovers the + // director's lineage children (parentWorktreeId === directorWorktreeId, same + // run) — the SAME data Mission Control uses — and seeds them here BEFORE the + // loop starts. A re-adopted track is then a hit (dispatchIntoExistingTrack): + // its cached terminal handle is a post-restart sentinel that won't match a live + // terminal, so resolveTrackTerminal relaunches the worker agent IN the existing + // checkout (preserving the predecessor's commits) rather than forking a new + // worktree/branch. Must be called before run/runFromExistingRun. + seedAdoptedTrackWorktrees( + entries: Iterable<[string, { worktreeId: string; terminalHandle: string; isAgent: boolean }]> + ): void { + for (const [trackKey, track] of entries) { + this.trackWorktrees.set(trackKey, track) + } + } + private async executeLoop(runId: string): Promise<{ runId: string status: CoordinatorStatus diff --git a/src/main/runtime/orchestration/db.test.ts b/src/main/runtime/orchestration/db.test.ts index dc252bc6d45..a6e0bdfe546 100644 --- a/src/main/runtime/orchestration/db.test.ts +++ b/src/main/runtime/orchestration/db.test.ts @@ -545,6 +545,41 @@ describe('OrchestrationDb', () => { expect(d.listCoordinatorRuns()).toHaveLength(3) }) + // Why (F3 #14): resume-on-boot reconstructs a crashed run's coordinator from + // the row, so the v9 options must round-trip. A persisted worktreeBacked=false + // must read back as 0 (an explicit legacy run), distinct from NULL (unknown). + it('persists and round-trips the v9 resume options (worktree-backed run)', () => { + const d = createDb() + const run = d.startCoordinatorRun({ + spec: 'build', + coordinatorHandle: 'coord', + targetKey: 'worktree:wt-1', + maxConcurrent: 7, + worktreeBacked: true, + workerAgent: 'claude' + }) + const read = d.getCoordinatorRun(run.id)! + expect(read.max_concurrent).toBe(7) + expect(read.worktree_backed).toBe(1) + expect(read.worker_agent).toBe('claude') + }) + + it('records worktreeBacked=false as 0 (explicit legacy), unset options as NULL', () => { + const d = createDb() + const legacy = d.startCoordinatorRun({ + spec: 'legacy', + coordinatorHandle: 'coord', + worktreeBacked: false + }) + expect(d.getCoordinatorRun(legacy.id)?.worktree_backed).toBe(0) + + const bare = d.createCoordinatorRun({ spec: 'bare', coordinatorHandle: 'coord2' }) + const read = d.getCoordinatorRun(bare.id)! + expect(read.worktree_backed).toBeNull() + expect(read.max_concurrent).toBeNull() + expect(read.worker_agent).toBeNull() + }) + it('counts outstanding tasks and active dispatches', () => { const d = createDb() expect(d.countOutstandingTasks()).toBe(0) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 8197dd97406..cf650ca09a5 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -40,6 +40,20 @@ function generateId(prefix: string): string { // sharing one DB file) — the prior in-memory `getActiveCoordinatorRun()` check // did not span processes, which is how two coordinators clashed. Runs on // different targets start in parallel. Callers map this to a friendly RPC error. +// Why (#14): the options persisted with a run so resume-on-boot can rebuild its +// coordinator. maxConcurrent/worktreeBacked/workerAgent are the in-memory-only +// coordinator settings that the run row did not record before — losing them on +// restart is what made a faithful resume impossible. +export type CoordinatorRunInsert = { + spec: string + coordinatorHandle: string + pollIntervalMs?: number + targetKey?: string | null + maxConcurrent?: number + worktreeBacked?: boolean + workerAgent?: string +} + export class CoordinatorRunConflictError extends Error { constructor(message = 'A coordinator run is already in progress for this target') { super(message) @@ -64,7 +78,11 @@ export class CoordinatorRunConflictError extends Error { // closing the cross-target poaching gap that per-target run-start alone left // open. The guard is enforced at write time by startCoordinatorRun // (BEGIN IMMEDIATE), not by a schema index. -const SCHEMA_VERSION = 8 +// v8 → v9 adds `coordinator_runs.{max_concurrent,worktree_backed,worker_agent}` +// so resume-on-boot (#14) can rebuild a crashed run's coordinator faithfully — +// the in-memory coordinator (its worktree-backed flag, worker agent, concurrency) +// is otherwise lost on restart, and a wrong guess re-zombies the run. +const SCHEMA_VERSION = 9 export class OrchestrationDb { private db: Database.Database @@ -170,6 +188,9 @@ export class OrchestrationDb { coordinator_handle TEXT NOT NULL, poll_interval_ms INTEGER NOT NULL DEFAULT 2000, target_key TEXT, + max_concurrent INTEGER, + worktree_backed INTEGER, + worker_agent TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), completed_at TEXT ); @@ -311,6 +332,20 @@ export class OrchestrationDb { this.db.exec(`ALTER TABLE tasks ADD COLUMN target_key TEXT`) } } + // v8 → v9: persist the coordinator options resume-on-boot (#14) needs to + // rebuild a crashed run's coordinator without guessing. Nullable/additive: + // pre-existing rows read NULL and resume falls back to defaults. + if (current < 9) { + if (!this.hasColumn('coordinator_runs', 'max_concurrent')) { + this.db.exec(`ALTER TABLE coordinator_runs ADD COLUMN max_concurrent INTEGER`) + } + if (!this.hasColumn('coordinator_runs', 'worktree_backed')) { + this.db.exec(`ALTER TABLE coordinator_runs ADD COLUMN worktree_backed INTEGER`) + } + if (!this.hasColumn('coordinator_runs', 'worker_agent')) { + this.db.exec(`ALTER TABLE coordinator_runs ADD COLUMN worker_agent TEXT`) + } + } this.createUndeliveredInboxIndexIfPossible() // Why: attach run-scoped lookup indexes now that the v6 columns exist // (createTables ran before the ALTERs above on an upgraded DB). @@ -1007,18 +1042,26 @@ export class OrchestrationDb { // ── Coordinator Runs ── - createCoordinatorRun(run: { - spec: string - coordinatorHandle: string - pollIntervalMs?: number - targetKey?: string | null - }): CoordinatorRun { + createCoordinatorRun(run: CoordinatorRunInsert): CoordinatorRun { const id = generateId('run') this.db .prepare( - "INSERT INTO coordinator_runs (id, spec, status, coordinator_handle, poll_interval_ms, target_key) VALUES (?, ?, 'running', ?, ?, ?)" + `INSERT INTO coordinator_runs + (id, spec, status, coordinator_handle, poll_interval_ms, target_key, max_concurrent, worktree_backed, worker_agent) + VALUES (?, ?, 'running', ?, ?, ?, ?, ?, ?)` + ) + .run( + id, + run.spec, + run.coordinatorHandle, + run.pollIntervalMs ?? 2000, + run.targetKey ?? null, + run.maxConcurrent ?? null, + // Why (#14): store the boolean as 0/1 so a NULL strictly means "pre-v9 / + // unknown" and resume can tell an explicit legacy run from a missing flag. + run.worktreeBacked === undefined ? null : run.worktreeBacked ? 1 : 0, + run.workerAgent ?? null ) - .run(id, run.spec, run.coordinatorHandle, run.pollIntervalMs ?? 2000, run.targetKey ?? null) return this.db.prepare('SELECT * FROM coordinator_runs WHERE id = ?').get(id) as CoordinatorRun } @@ -1033,12 +1076,7 @@ export class OrchestrationDb { // parallel. A null target_key (no worktree given at run-start) shares one // slot — unidentified targets fall back to single-run. Used by the RPC path; // tests/coordinator.run() use the plain createCoordinatorRun insert. - startCoordinatorRun(run: { - spec: string - coordinatorHandle: string - pollIntervalMs?: number - targetKey?: string | null - }): CoordinatorRun { + startCoordinatorRun(run: CoordinatorRunInsert): CoordinatorRun { const targetKey = run.targetKey ?? null this.db.exec('BEGIN IMMEDIATE') let committed = false @@ -1057,9 +1095,20 @@ export class OrchestrationDb { const id = generateId('run') this.db .prepare( - "INSERT INTO coordinator_runs (id, spec, status, coordinator_handle, poll_interval_ms, target_key) VALUES (?, ?, 'running', ?, ?, ?)" + `INSERT INTO coordinator_runs + (id, spec, status, coordinator_handle, poll_interval_ms, target_key, max_concurrent, worktree_backed, worker_agent) + VALUES (?, ?, 'running', ?, ?, ?, ?, ?, ?)` + ) + .run( + id, + run.spec, + run.coordinatorHandle, + run.pollIntervalMs ?? 2000, + targetKey, + run.maxConcurrent ?? null, + run.worktreeBacked === undefined ? null : run.worktreeBacked ? 1 : 0, + run.workerAgent ?? null ) - .run(id, run.spec, run.coordinatorHandle, run.pollIntervalMs ?? 2000, targetKey) this.db.exec('COMMIT') committed = true return this.db @@ -1093,6 +1142,27 @@ export class OrchestrationDb { return this.getCoordinatorRun(id) } + // Why (#14): on resume-on-boot, a task left 'dispatched' by the crashed + // coordinator has a worker that is no longer being monitored (the in-memory + // coordinator that would receive its worker_done is gone). Leaving it + // 'dispatched' would never converge — the resumed loop neither re-dispatches it + // (only 'ready' tasks dispatch) nor times it out (the stale detector only + // warns). So reclaim each in-flight dispatch through the SAME breaker a live + // failure uses (failActiveDispatchForTask): the task returns to 'ready' to be + // re-dispatched into its re-adopted worktree, and the failure strike bounds an + // infinite restart→resume loop on a poison run (after 3 it converges to + // 'failed'). Run-scoped so a concurrent run on another target is untouched. + reclaimInFlightDispatchesForResume(coordinatorRunId: string, reason: string): number { + const dispatched = this.listTasks({ status: 'dispatched', coordinatorRunId }) + let reclaimed = 0 + for (const task of dispatched) { + if (this.failActiveDispatchForTask(task.id, reason)) { + reclaimed++ + } + } + return reclaimed + } + getActiveCoordinatorRun(): CoordinatorRun | undefined { return this.db .prepare( diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index bfd6fe1abcc..80cf599a93e 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -98,6 +98,15 @@ export type CoordinatorRun = { // only a duplicate run on the same target, not all concurrency. NULL when no // worktree was given at run-start (those runs share a single-run slot). target_key: string | null + // Why (F3 #14): the coordinator is an in-memory instance with no boot hook, so + // after an app restart these persisted options let resume-on-boot reconstruct + // the SAME run (worktree-backed vs legacy, which agent, concurrency) instead of + // guessing — a legacy bare-terminal resume of a worktree-backed run would + // dispatch into a shell that never reports done (a fresh zombie). NULL on + // pre-v9 rows / non-worktree runs; resume falls back to coordinator defaults. + max_concurrent: number | null + worktree_backed: number | null + worker_agent: string | null created_at: string completed_at: string | null } diff --git a/src/main/runtime/rpc/methods/orchestration-gates.ts b/src/main/runtime/rpc/methods/orchestration-gates.ts index 8fbc1d65722..200327a1933 100644 --- a/src/main/runtime/rpc/methods/orchestration-gates.ts +++ b/src/main/runtime/rpc/methods/orchestration-gates.ts @@ -1,11 +1,21 @@ import { randomBytes } from 'crypto' import { z } from 'zod' -import type { TuiAgent } from '../../../../shared/types' +import type { TuiAgent, WorktreeLineage } from '../../../../shared/types' import { TUI_AGENT_CONFIG } from '../../../../shared/tui-agent-config' import { defineMethod, type RpcMethod } from '../core' import { OptionalBoolean, OptionalFiniteNumber, OptionalString, requiredString } from '../schemas' -import { CoordinatorRunConflictError, type GateStatus } from '../../orchestration/db' -import { Coordinator } from '../../orchestration/coordinator' +import { + CoordinatorRunConflictError, + type GateStatus, + type OrchestrationDb, + type CoordinatorRun +} from '../../orchestration/db' +import { Coordinator, type CoordinatorRuntime } from '../../orchestration/coordinator' +import { + buildAdoptedTrackWorktrees, + reconcileCoordinatorRunsOnBoot, + type ReconcileRunOnBootResult +} from '../../orchestration/boot-resume' // Why: the most-recently-started coordinator is stored at module scope so // orchestration.runStop can signal it to halt. Concurrent runs on different @@ -117,7 +127,13 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [ spec: params.spec, coordinatorHandle, pollIntervalMs: params.pollIntervalMs, - targetKey + targetKey, + // Why (F3 #14): persist the in-memory-only coordinator options so a + // restart can rebuild THIS run faithfully (worktree-backed vs legacy, + // which agent, concurrency) instead of guessing and re-zombieing it. + ...(params.maxConcurrent !== undefined ? { maxConcurrent: params.maxConcurrent } : {}), + ...(params.worktreeBacked !== undefined ? { worktreeBacked: params.worktreeBacked } : {}), + ...(params.workerAgent ? { workerAgent: params.workerAgent } : {}) }) } catch (err) { if (err instanceof CoordinatorRunConflictError) { @@ -222,3 +238,111 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [ } }) ] + +// Why (F3 #14): the runtime surface resume-on-boot needs. CoordinatorRuntime +// supplies the dispatch/worktree methods the rebuilt Coordinator uses; the three +// extras resolve the run's DB, verify the director worktree still exists, and +// read the lineage map for track re-adoption (the same map Mission Control uses). +type OrchestrationBootRuntime = CoordinatorRuntime & { + getOrchestrationDb(): OrchestrationDb + resolveOrchestrationTargetKey(selector?: string): Promise + listWorktreeLineage(): Promise> +} + +const WORKTREE_TARGET_PREFIX = 'worktree:' + +// Why (F3 #14): rebuild and restart a crashed run's coordinator. Returns false +// (→ the reconciler marks the run failed, killing the zombie) when the run cannot +// be safely resumed: no worktree target to anchor lineage/drift on, or the +// director worktree no longer resolves (nothing to branch children from). On the +// happy path it reclaims dead in-flight dispatches (so the loop converges), +// re-adopts existing track worktrees, registers the coordinator as active (so +// orchestration.runStop can halt it), and fires the loop. Any throw propagates to +// the reconciler, which also falls back to failed — never leaves the run running +// without a loop. +async function resumeCoordinatorRunOnBoot( + runtime: OrchestrationBootRuntime, + db: OrchestrationDb, + run: CoordinatorRun, + onLog: (msg: string) => void +): Promise { + if (!run.target_key || !run.target_key.startsWith(WORKTREE_TARGET_PREFIX)) { + onLog(`Resume: run ${run.id} has no worktree target; cannot resume`) + return false + } + const directorWorktreeId = run.target_key.slice(WORKTREE_TARGET_PREFIX.length) + const directorSelector = `id:${directorWorktreeId}` + try { + // Throws when the director worktree was deleted while Orca was closed. + await runtime.resolveOrchestrationTargetKey(directorSelector) + } catch { + onLog(`Resume: run ${run.id} director worktree ${directorWorktreeId} is gone; cannot resume`) + return false + } + + const worktreeBacked = run.worktree_backed === 1 + const workerAgent = run.worker_agent ?? undefined + + // Converge-safety: a task left 'dispatched' by the crashed loop has a dead, + // unmonitored worker — reclaim it to 'ready' so the resumed loop re-dispatches + // (or breaker-fails) it rather than spinning forever on a hung dispatch. + const reclaimed = db.reclaimInFlightDispatchesForResume( + run.id, + 'reclaimed on boot: coordinator restarted' + ) + if (reclaimed > 0) { + onLog(`Resume: run ${run.id} reclaimed ${reclaimed} in-flight dispatch(es)`) + } + + const coordinator = new Coordinator(db, runtime, { + spec: run.spec, + coordinatorHandle: run.coordinator_handle, + pollIntervalMs: run.poll_interval_ms, + ...(run.max_concurrent != null ? { maxConcurrent: run.max_concurrent } : {}), + worktree: directorSelector, + ...(worktreeBacked ? { worktreeBacked: true } : {}), + ...(workerAgent ? { workerAgent: workerAgent as TuiAgent } : {}), + onLog + }) + + if (worktreeBacked) { + const lineage = await runtime.listWorktreeLineage() + const entries = buildAdoptedTrackWorktrees( + db, + run, + lineage, + directorWorktreeId, + workerAgent !== undefined + ) + coordinator.seedAdoptedTrackWorktrees(entries) + if (entries.size > 0) { + onLog(`Resume: run ${run.id} re-adopted ${entries.size} track worktree(s)`) + } + } + + activeCoordinator = coordinator + coordinator.runFromExistingRun(run.id).finally(() => { + if (activeCoordinator === coordinator) { + activeCoordinator = null + } + }) + return true +} + +// Why (F3 #14): the boot entry point. Scans running coordinator runs and +// reconciles each (finalize / resume / fail) so a restart never leaves a zombie +// that blocks a fresh run for the target. Call once on app boot, before any +// launch path that checks the per-target active-run guard. Errors are the +// caller's to isolate — a reconcile failure must not block app startup. +export async function runOrchestrationBootReconcile( + runtime: OrchestrationBootRuntime, + onLog?: (msg: string) => void +): Promise { + const db = runtime.getOrchestrationDb() + const log = onLog ?? (() => {}) + return reconcileCoordinatorRunsOnBoot({ + db, + resume: (run) => resumeCoordinatorRunOnBoot(runtime, db, run, log), + onLog: log + }) +} From 31f7542f21bd0b4c821bd2123887147e98e01e6a Mon Sep 17 00:00:00 2001 From: zaridan <1617679+zaridan@users.noreply.github.com> Date: Tue, 23 Jun 2026 19:02:48 -0700 Subject: [PATCH 2/2] fix(orchestration): close F3 resume-on-boot hard-rule escape hatches (#14, round 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 review found a real HARD-RULE violation (a run left running with no loop) and that the production resume path was untested. This closes both plus the should-fixes. Must-fix: 1. Catch-less detached loop → fresh zombie. `executeLoop` ran `adoptUnownedTasks` (+ the pre-read) OUTSIDE its `try`, and the resume's `runFromExistingRun(...)` had no `.catch`. A throw there rejected the loop un-finalized and the reconciler had already skipped its failed-fallback → run stuck `running` with no loop. Fix (belt + suspenders): moved the pre-loop work INSIDE `executeLoop`'s try, and added a `.catch` on the resumed loop that force-marks the run `failed`. 2. Real resume path now tested. New `orchestration-resume-on-boot.test.ts` drives the production `resumeCoordinatorRunOnBoot` via `runOrchestrationBootReconcile` with a fake runtime + real DB: resumable run → loop starts + track re-adopted (no duplicate worktree); the catch-less-loop case (throwing `adoptUnownedTasks` → run ends `failed`, never `running`); and every declines→`failed` branch. The hard-rule + NULL tests FAIL against round-1 code (verified). Should-fix (same pass): 3. Pre-v9 `worktree_backed = NULL` with a worktree target is now treated as not-safely-resumable → `failed` (guessing legacy would dispatch into a bare shell that never converges → unbounded re-zombie). Only an explicit `worktree_backed = 0` resumes in legacy mode. 4. Cross-process double-drive closed. Schema v10 adds `coordinator_runs.resumed_at`, a boot-time-fenced atomic claim (`tryClaimRunForResume`, BEGIN IMMEDIATE). The resume callback claims FIRST; a loser returns `contended` → the run is left running (a live owner drives it), never double-driven and never failed out from under its owner. A strictly-greater later fence reclaims a crashed resumer's stale claim, so a crash mid-resume can't strand the run. Makes the (intentionally redundant) serve-mode boot reconcile safe without a serve guard. 5. Per-run try/catch in the reconcile loop: one row's transient error (e.g. SQLITE_BUSY) no longer strands every later running row (`reconcile-error` disposition, logged, continues). Nits: claim-based idempotency replaces "idempotent by construction" (a second pass is `contended`→`skipped`); director-resolve failure documented as fail-closed (favoring the hard rule) with transient-vs-not-found distinction noted as follow-up. Verification: vitest orchestration + MC + RPC suites green (297 passed, 5x stable); typecheck node/cli/web clean (only 4 pre-existing TuiAgent errors on main); oxlint clean; electron-vite build green. New resume tests verified red against round-1. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/main/index.ts | 4 +- .../runtime/orchestration/boot-resume.test.ts | 50 ++- src/main/runtime/orchestration/boot-resume.ts | 109 ++++--- src/main/runtime/orchestration/coordinator.ts | 19 +- src/main/runtime/orchestration/db.test.ts | 29 ++ src/main/runtime/orchestration/db.ts | 47 ++- src/main/runtime/orchestration/types.ts | 7 + .../rpc/methods/orchestration-gates.ts | 83 +++-- .../orchestration-resume-on-boot.test.ts | 302 ++++++++++++++++++ 9 files changed, 583 insertions(+), 67 deletions(-) create mode 100644 src/main/runtime/rpc/methods/orchestration-resume-on-boot.test.ts diff --git a/src/main/index.ts b/src/main/index.ts index a91303d30cc..3fc7066bdff 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -1532,7 +1532,9 @@ app.whenReady().then(async () => { // Gated on the experimental flag (the only surface that starts runs) to stay // additive, fire-and-forget so a reconcile error can never block startup, and // run here — before any window is shown — so it lands before a user-triggered - // run could hit the guard. + // run could hit the guard. This runs in serve mode too; the resume path's atomic + // boot-time-fenced claim (tryClaimRunForResume) makes that safe — a desktop and a + // serve runtime sharing one DB can't both drive the same resumed run. if (store.getSettings().experimentalOrchestrators) { void runOrchestrationBootReconcile(runtimeService, (msg) => console.log('[orchestration-resume]', msg) diff --git a/src/main/runtime/orchestration/boot-resume.test.ts b/src/main/runtime/orchestration/boot-resume.test.ts index 4128162761f..0dd768926db 100644 --- a/src/main/runtime/orchestration/boot-resume.test.ts +++ b/src/main/runtime/orchestration/boot-resume.test.ts @@ -115,7 +115,7 @@ describe('reconcileCoordinatorRunsOnBoot (F3 #14)', () => { db: d, resume: async (r: CoordinatorRun) => { resumed.push(r.id) - return true + return 'resumed' } } const results = await reconcileCoordinatorRunsOnBoot(deps) @@ -125,16 +125,62 @@ describe('reconcileCoordinatorRunsOnBoot (F3 #14)', () => { expect(d.getCoordinatorRun(run.id)?.status).toBe('running') }) + it('leaves a run running (skipped) when the resume callback reports contention', async () => { + const d = createDb() + const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) + d.createTask({ spec: 'A', coordinatorRunId: run.id }) + + // 'contended' = another runtime owns it; the reconciler must NOT fail it. + const results = await reconcileCoordinatorRunsOnBoot({ db: d, resume: async () => 'contended' }) + expect(results[0].disposition).toBe('skipped') + expect(d.getCoordinatorRun(run.id)?.status).toBe('running') + }) + it('falls back to failed when the resume callback declines', async () => { const d = createDb() const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) d.createTask({ spec: 'A', coordinatorRunId: run.id }) - const results = await reconcileCoordinatorRunsOnBoot({ db: d, resume: async () => false }) + const results = await reconcileCoordinatorRunsOnBoot({ db: d, resume: async () => 'declined' }) expect(results[0].disposition).toBe('failed') expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') }) + it('isolates a per-run reconcile error so later running rows are not stranded', async () => { + const d = createDb() + const bad = d.startCoordinatorRun({ + spec: 'bad', + coordinatorHandle: 'cb', + targetKey: 'worktree:wtA' + }) + d.createTask({ spec: 'A', coordinatorRunId: bad.id, targetKey: 'worktree:wtA' }) + const good = d.startCoordinatorRun({ + spec: 'good', + coordinatorHandle: 'cg', + targetKey: 'worktree:wtB' + }) + d.createTask({ spec: 'B', coordinatorRunId: good.id, targetKey: 'worktree:wtB' }) + + // The first run's resume throws inside a way the per-run try/catch must contain; + // the second run must still be reconciled (here: declined → failed). + const results = await reconcileCoordinatorRunsOnBoot({ + db: d, + resume: async (r) => { + if (r.id === bad.id) { + throw new Error('transient boom') + } + return 'declined' + } + }) + const byId = Object.fromEntries(results.map((r) => [r.runId, r.disposition])) + // bad: a throw is treated as declined → failed (never left running). + expect(byId[bad.id]).toBe('failed') + expect(d.getCoordinatorRun(bad.id)?.status).toBe('failed') + // good: still processed despite bad's failure. + expect(byId[good.id]).toBe('failed') + expect(d.getCoordinatorRun(good.id)?.status).toBe('failed') + }) + it('falls back to failed when the resume callback throws (never left running with no loop)', async () => { const d = createDb() const run = d.startCoordinatorRun({ spec: 'go', coordinatorHandle: 'c' }) diff --git a/src/main/runtime/orchestration/boot-resume.ts b/src/main/runtime/orchestration/boot-resume.ts index c95c215b420..d0c6b7a6787 100644 --- a/src/main/runtime/orchestration/boot-resume.ts +++ b/src/main/runtime/orchestration/boot-resume.ts @@ -13,7 +13,22 @@ import type { CoordinatorRun, TaskRow } from './types' import { parseTrackFromSpec } from './coordinator' import type { WorktreeLineage } from '../../../shared/types' -export type BootRunDisposition = 'finalized-completed' | 'finalized-failed' | 'resumed' | 'failed' +export type BootRunDisposition = + | 'finalized-completed' + | 'finalized-failed' + | 'resumed' + | 'failed' + | 'reconcile-error' + // Why (F3 #14): another runtime won the atomic resume claim and is driving this + // run, so we leave it 'running' (NOT a zombie — a live loop owns it) and do not + // touch it. Distinct from 'resumed' (we drive it) and 'failed' (no driver). + | 'skipped' + +// Why (F3 #14): the resume callback's three outcomes the reconciler must tell +// apart. 'resumed' → a converging loop now drives it (stays running). 'declined' +// → unresumable, the reconciler marks it failed (kills the zombie). 'contended' +// → another runtime claimed it; leave it running (it has a live owner). +export type ResumeOutcome = 'resumed' | 'declined' | 'contended' export type ReconcileRunOnBootResult = { runId: string @@ -23,12 +38,11 @@ export type ReconcileRunOnBootResult = { export type ReconcileCoordinatorRunsOnBootDeps = { db: OrchestrationDb - // Attempt to resume a run that still has outstanding work. Returns true when a - // converging coordinator loop was started (the run legitimately stays - // 'running' under a live loop). Returns false — or throws — when resume is - // declined or fails, and the reconciler marks the run failed instead. Omitted + // Attempt to resume a run that still has outstanding work. 'resumed' keeps the + // run 'running' under a live loop; 'declined' (or a throw) → the reconciler marks + // it failed; 'contended' → another runtime owns it, leave it running. Omitted // entirely → MUST-only mode: every outstanding run is reconciled to failed. - resume?: (run: CoordinatorRun) => Promise + resume?: (run: CoordinatorRun) => Promise onLog?: (msg: string) => void } @@ -60,46 +74,69 @@ export async function reconcileCoordinatorRunsOnBoot( ): Promise { const { db, resume } = deps const onLog = deps.onLog ?? (() => {}) - // Snapshot the running rows up front. Idempotent by construction: a second pass - // finds finalized/failed rows gone from this list, and the only rows still - // 'running' belong to a live resumed loop (the production caller runs once). + // Snapshot the running rows up front. A resumed run stays 'running' but is now + // claimed (resumed_at fenced), so a second pass's resume claim loses → 'contended' + // → left alone: the pass is safe to repeat. const runs = db.listCoordinatorRuns({ status: 'running' }) const results: ReconcileRunOnBootResult[] = [] for (const run of runs) { - const { finalizeStatus } = classifyOutstandingWork(db, run) - if (finalizeStatus) { - db.updateCoordinatorRun(run.id, finalizeStatus) - onLog(`Boot reconcile: run ${run.id} had no outstanding work; finalized as ${finalizeStatus}`) - results.push({ - runId: run.id, - disposition: finalizeStatus === 'failed' ? 'finalized-failed' : 'finalized-completed' - }) - continue + // Why (F3 #14, should-fix #5): isolate each run. A transient DB error (e.g. + // SQLITE_BUSY) on one row must not abort the loop and strand every LATER + // running row as a zombie. Log and continue; the stranded row is retried next + // boot (it is still 'running'). + try { + results.push(await reconcileOneRunOnBoot(db, run, resume, onLog)) + } catch (err) { + onLog(`Boot reconcile: run ${run.id} errored; left for next boot (${String(err)})`) + results.push({ runId: run.id, disposition: 'reconcile-error', reason: String(err) }) } + } - if (resume) { - let resumed = false - try { - resumed = await resume(run) - } catch (err) { - onLog(`Boot reconcile: resume of run ${run.id} threw; marking failed (${String(err)})`) - } - if (resumed) { - onLog(`Boot reconcile: resumed run ${run.id}`) - results.push({ runId: run.id, disposition: 'resumed' }) - continue - } + return results +} + +async function reconcileOneRunOnBoot( + db: OrchestrationDb, + run: CoordinatorRun, + resume: ((run: CoordinatorRun) => Promise) | undefined, + onLog: (msg: string) => void +): Promise { + const { finalizeStatus } = classifyOutstandingWork(db, run) + if (finalizeStatus) { + db.updateCoordinatorRun(run.id, finalizeStatus) + onLog(`Boot reconcile: run ${run.id} had no outstanding work; finalized as ${finalizeStatus}`) + return { + runId: run.id, + disposition: finalizeStatus === 'failed' ? 'finalized-failed' : 'finalized-completed' } + } - // Floor: cannot/should not resume → converge to failed so the guard unblocks. - const reason = 'coordinator restarted with no resumable loop' - db.updateCoordinatorRun(run.id, 'failed') - onLog(`Boot reconcile: run ${run.id} not resumed; marked failed (${reason})`) - results.push({ runId: run.id, disposition: 'failed', reason }) + if (resume) { + let outcome: ResumeOutcome = 'declined' + try { + outcome = await resume(run) + } catch (err) { + // A throw is treated as declined → marked failed below (never left running). + onLog(`Boot reconcile: resume of run ${run.id} threw; marking failed (${String(err)})`) + outcome = 'declined' + } + if (outcome === 'resumed') { + onLog(`Boot reconcile: resumed run ${run.id}`) + return { runId: run.id, disposition: 'resumed' } + } + if (outcome === 'contended') { + // Another runtime owns it (a live loop drives it) — leave it running. + onLog(`Boot reconcile: run ${run.id} is owned by another runtime; left running`) + return { runId: run.id, disposition: 'skipped' } + } } - return results + // Floor: cannot/should not resume → converge to failed so the guard unblocks. + const reason = 'coordinator restarted with no resumable loop' + db.updateCoordinatorRun(run.id, 'failed') + onLog(`Boot reconcile: run ${run.id} not resumed; marked failed (${reason})`) + return { runId: run.id, disposition: 'failed', reason } } // Why (F3 #14, design §8): rebuild the coordinator's in-memory track→worktree map diff --git a/src/main/runtime/orchestration/coordinator.ts b/src/main/runtime/orchestration/coordinator.ts index 680ea5ab15c..f2115916493 100644 --- a/src/main/runtime/orchestration/coordinator.ts +++ b/src/main/runtime/orchestration/coordinator.ts @@ -327,14 +327,19 @@ export class Coordinator { this.state.runId = runId this.opts.onLog(`Coordinator run ${runId} started`) - // Why (#12): tasks are created via orchestration.taskCreate before the run - // exists, so they start unowned. Claim them for this run before decompose - // reads the (now run-scoped) DAG — but only tasks on THIS run's target, so - // a concurrent run on another target can't have its tasks poached. - const targetKey = this.db.getCoordinatorRun(runId)?.target_key ?? null - this.db.adoptUnownedTasks(runId, targetKey) - try { + // Why (#12): tasks are created via orchestration.taskCreate before the run + // exists, so they start unowned. Claim them for this run before decompose + // reads the (now run-scoped) DAG — but only tasks on THIS run's target, so + // a concurrent run on another target can't have its tasks poached. + // Why (F3 #14): this MUST stay inside the try. On a resumed run a throw here + // (e.g. a transient DB error) would otherwise reject the loop promise WITHOUT + // finalizing the run — leaving it status='running' with no live loop, the + // exact zombie F3 exists to prevent. Routing it through the catch marks the + // run failed so the guard unblocks. + const targetKey = this.db.getCoordinatorRun(runId)?.target_key ?? null + this.db.adoptUnownedTasks(runId, targetKey) + await this.decompose() while (!this.stopped) { diff --git a/src/main/runtime/orchestration/db.test.ts b/src/main/runtime/orchestration/db.test.ts index a6e0bdfe546..cc4d06239c0 100644 --- a/src/main/runtime/orchestration/db.test.ts +++ b/src/main/runtime/orchestration/db.test.ts @@ -580,6 +580,35 @@ describe('OrchestrationDb', () => { expect(read.worker_agent).toBeNull() }) + // Why (F3 #14): the boot-time-fenced resume claim. One winner per fence; a + // strictly-greater later fence reclaims a stale (crashed-resumer) claim; a + // non-running run can't be claimed. + it('tryClaimRunForResume: one winner per fence, later fence reclaims, terminal not claimable', () => { + const d = createDb() + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'c', + targetKey: 'worktree:wt' + }) + const fence1 = '2026-06-23T10:00:00.000Z' + // Two runtimes at the SAME fence: exactly one wins. + expect(d.tryClaimRunForResume(run.id, fence1)).toBe(true) + expect(d.tryClaimRunForResume(run.id, fence1)).toBe(false) + expect(d.getCoordinatorRun(run.id)?.resumed_at).toBe(fence1) + + // A later boot (strictly-greater fence) reclaims a stale claim. + const fence2 = '2026-06-23T11:00:00.000Z' + expect(d.tryClaimRunForResume(run.id, fence2)).toBe(true) + expect(d.getCoordinatorRun(run.id)?.resumed_at).toBe(fence2) + + // An earlier fence cannot steal a newer claim. + expect(d.tryClaimRunForResume(run.id, fence1)).toBe(false) + + // Once terminal, the run is never claimable. + d.updateCoordinatorRun(run.id, 'failed') + expect(d.tryClaimRunForResume(run.id, '2026-06-23T12:00:00.000Z')).toBe(false) + }) + it('counts outstanding tasks and active dispatches', () => { const d = createDb() expect(d.countOutstandingTasks()).toBe(0) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index cf650ca09a5..405a218e3e5 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -82,7 +82,10 @@ export class CoordinatorRunConflictError extends Error { // so resume-on-boot (#14) can rebuild a crashed run's coordinator faithfully — // the in-memory coordinator (its worktree-backed flag, worker agent, concurrency) // is otherwise lost on restart, and a wrong guess re-zombies the run. -const SCHEMA_VERSION = 9 +// v9 → v10 adds `coordinator_runs.resumed_at`, a boot-time-fenced resume claim so +// two runtimes sharing one DB (desktop + `orca serve`) can't both drive a resumed +// run, while a strictly-greater later fence reclaims a crashed resumer's stale claim. +const SCHEMA_VERSION = 10 export class OrchestrationDb { private db: Database.Database @@ -191,6 +194,7 @@ export class OrchestrationDb { max_concurrent INTEGER, worktree_backed INTEGER, worker_agent TEXT, + resumed_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), completed_at TEXT ); @@ -346,6 +350,12 @@ export class OrchestrationDb { this.db.exec(`ALTER TABLE coordinator_runs ADD COLUMN worker_agent TEXT`) } } + // v9 → v10: the boot-time-fenced resume claim (#14). Nullable/additive. + if (current < 10) { + if (!this.hasColumn('coordinator_runs', 'resumed_at')) { + this.db.exec(`ALTER TABLE coordinator_runs ADD COLUMN resumed_at TEXT`) + } + } this.createUndeliveredInboxIndexIfPossible() // Why: attach run-scoped lookup indexes now that the v6 columns exist // (createTables ran before the ALTERs above on an upgraded DB). @@ -1163,6 +1173,41 @@ export class OrchestrationDb { return reclaimed } + // Why (F3 #14): an atomic, boot-time-fenced claim so only ONE runtime drives a + // resumed run when two share a DB file (desktop + `orca serve`). `fenceIso` is the + // claiming runtime's boot time. The claim wins iff the run is still 'running' and + // is unclaimed OR holds a STRICTLY-OLDER fence (a prior, now-dead resumer). Two + // runtimes booting together share a fence, so the strict `<` lets exactly one win + // (the other sees the just-written equal fence and loses) — no double-drive. A + // later boot's larger fence reclaims a crashed resumer's stale claim, so a crash + // mid-resume can't strand the run. BEGIN IMMEDIATE serializes the check+set across + // processes. Store ISO (not datetime('now')) so it lexically compares with fenceIso. + tryClaimRunForResume(runId: string, fenceIso: string): boolean { + this.db.exec('BEGIN IMMEDIATE') + let committed = false + try { + const info = this.db + .prepare( + `UPDATE coordinator_runs SET resumed_at = ? + WHERE id = ? + AND status = 'running' + AND (resumed_at IS NULL OR resumed_at < ?)` + ) + .run(fenceIso, runId, fenceIso) + this.db.exec('COMMIT') + committed = true + return Number((info as { changes?: number | bigint }).changes ?? 0) === 1 + } finally { + if (!committed) { + try { + this.db.exec('ROLLBACK') + } catch { + // No active transaction to roll back — ignore. + } + } + } + } + getActiveCoordinatorRun(): CoordinatorRun | undefined { return this.db .prepare( diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index 80cf599a93e..32f1473bf31 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -107,6 +107,13 @@ export type CoordinatorRun = { max_concurrent: number | null worktree_backed: number | null worker_agent: string | null + // Why (F3 #14): a boot-time-fenced resume claim. When a boot reconciler resumes + // a running run it stamps this with the resuming runtime's boot time, so a second + // runtime booting against the same shared DB (desktop + `orca serve`) loses the + // atomic claim and never double-drives the same run. A later boot (whose fence is + // strictly greater) reclaims a stale claim left by a crashed resumer, so a crash + // mid-resume cannot strand the run forever. NULL until first claimed. + resumed_at: string | null created_at: string completed_at: string | null } diff --git a/src/main/runtime/rpc/methods/orchestration-gates.ts b/src/main/runtime/rpc/methods/orchestration-gates.ts index 200327a1933..57c802f2487 100644 --- a/src/main/runtime/rpc/methods/orchestration-gates.ts +++ b/src/main/runtime/rpc/methods/orchestration-gates.ts @@ -14,7 +14,8 @@ import { Coordinator, type CoordinatorRuntime } from '../../orchestration/coordi import { buildAdoptedTrackWorktrees, reconcileCoordinatorRunsOnBoot, - type ReconcileRunOnBootResult + type ReconcileRunOnBootResult, + type ResumeOutcome } from '../../orchestration/boot-resume' // Why: the most-recently-started coordinator is stored at module scope so @@ -247,37 +248,65 @@ type OrchestrationBootRuntime = CoordinatorRuntime & { getOrchestrationDb(): OrchestrationDb resolveOrchestrationTargetKey(selector?: string): Promise listWorktreeLineage(): Promise> + // Why (F3 #14, should-fix #4): the runtime boot time fences the atomic resume + // claim so two runtimes sharing a DB can't both drive one run, while a later + // boot reclaims a crashed resumer's stale claim. + getStartedAt(): number } const WORKTREE_TARGET_PREFIX = 'worktree:' -// Why (F3 #14): rebuild and restart a crashed run's coordinator. Returns false -// (→ the reconciler marks the run failed, killing the zombie) when the run cannot -// be safely resumed: no worktree target to anchor lineage/drift on, or the -// director worktree no longer resolves (nothing to branch children from). On the -// happy path it reclaims dead in-flight dispatches (so the loop converges), -// re-adopts existing track worktrees, registers the coordinator as active (so -// orchestration.runStop can halt it), and fires the loop. Any throw propagates to -// the reconciler, which also falls back to failed — never leaves the run running -// without a loop. +// Why (F3 #14): rebuild and restart a crashed run's coordinator. Returns: +// - 'contended' — another runtime won the atomic claim; leave the run running +// (it has a live owner), do not touch it. +// - 'declined' — WE own the claim but the run is not safely resumable (no +// worktree target, pre-v9 unknown mode, or the director worktree is gone), so +// the reconciler marks it failed (kills the zombie, unblocks the guard). +// - 'resumed' — WE own the claim and started a converging loop. +// The claim is taken FIRST so a loser never declines→fails a run another runtime +// is driving. On the happy path it reclaims dead in-flight dispatches (so the loop +// converges), re-adopts existing track worktrees, registers the coordinator as +// active (so orchestration.runStop can halt it), and fires the loop — whose .catch +// force-fails the run if the loop ever rejects un-finalized (hard-rule belt). async function resumeCoordinatorRunOnBoot( runtime: OrchestrationBootRuntime, db: OrchestrationDb, run: CoordinatorRun, onLog: (msg: string) => void -): Promise { +): Promise { + // Atomic, boot-time-fenced claim FIRST: only the winner decides drive-vs-fail, + // so a second runtime sharing the DB neither double-drives nor fails a run the + // owner is driving. + const fenceIso = new Date(runtime.getStartedAt()).toISOString() + if (!db.tryClaimRunForResume(run.id, fenceIso)) { + onLog(`Resume: run ${run.id} is claimed by another runtime; skipping`) + return 'contended' + } + if (!run.target_key || !run.target_key.startsWith(WORKTREE_TARGET_PREFIX)) { onLog(`Resume: run ${run.id} has no worktree target; cannot resume`) - return false + return 'declined' + } + // Why (should-fix #3): a NULL worktree_backed predates v9 — we cannot tell if it + // was worktree-backed. Guessing legacy would dispatch into a bare shell that never + // reports done → never converges → re-resumes every restart. Fail it instead. + // Only a genuine, explicitly-recorded legacy run (worktree_backed = 0) resumes in + // legacy mode. + if (run.worktree_backed === null) { + onLog(`Resume: run ${run.id} predates v9 (worktree_backed unknown); cannot safely resume`) + return 'declined' } const directorWorktreeId = run.target_key.slice(WORKTREE_TARGET_PREFIX.length) const directorSelector = `id:${directorWorktreeId}` try { - // Throws when the director worktree was deleted while Orca was closed. + // Throws when the director worktree was deleted while Orca was closed. We fail + // closed (favor the hard rule): a run we cannot anchor converges to failed + // rather than risk a zombie. (Distinguishing a transient resolve error from a + // genuine not-found is a follow-up; fail-closed is the safe default here.) await runtime.resolveOrchestrationTargetKey(directorSelector) } catch { onLog(`Resume: run ${run.id} director worktree ${directorWorktreeId} is gone; cannot resume`) - return false + return 'declined' } const worktreeBacked = run.worktree_backed === 1 @@ -321,12 +350,26 @@ async function resumeCoordinatorRunOnBoot( } activeCoordinator = coordinator - coordinator.runFromExistingRun(run.id).finally(() => { - if (activeCoordinator === coordinator) { - activeCoordinator = null - } - }) - return true + coordinator + .runFromExistingRun(run.id) + .catch((err) => { + // Hard-rule belt (must-fix #1): if the loop ever rejects WITHOUT finalizing + // (e.g. a future pre-try line in executeLoop throws), force the run to failed + // so it can never be left 'running' with no live loop. Best-effort: a DB + // closed/locked at shutdown must not crash the process. + onLog(`Resume: run ${run.id} loop rejected; marking failed (${String(err)})`) + try { + db.updateCoordinatorRun(run.id, 'failed') + } catch { + // ignore — nothing more we can safely do here + } + }) + .finally(() => { + if (activeCoordinator === coordinator) { + activeCoordinator = null + } + }) + return 'resumed' } // Why (F3 #14): the boot entry point. Scans running coordinator runs and diff --git a/src/main/runtime/rpc/methods/orchestration-resume-on-boot.test.ts b/src/main/runtime/rpc/methods/orchestration-resume-on-boot.test.ts new file mode 100644 index 00000000000..faa5eec8c9d --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-resume-on-boot.test.ts @@ -0,0 +1,302 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from '../../orchestration/db' +import type { CoordinatorRuntime } from '../../orchestration/coordinator' +import { runOrchestrationBootReconcile } from './orchestration-gates' +import type { WorktreeLineage } from '../../../../shared/types' + +// Why (F3 #14, round 2 #2): the round-1 tests only ran `reconcileCoordinatorRunsOnBoot` +// in MUST-only mode (no resume fn), so the PRODUCTION callback `resumeCoordinatorRunOnBoot` +// (claim → guards → reclaim → v9 rebuild → lineage→adopt→seed → loop fire → .catch) was +// never exercised — which is why the catch-less-loop hard-rule bug shipped green. These +// drive the real callback via `runOrchestrationBootReconcile`. They FAIL against round-1 +// code (catch-less loop leaves the run 'running'; NULL worktree_backed resumes legacy). + +type Terminal = { handle: string; worktreeId: string; connected: boolean; writable: boolean } + +type FakeRuntime = CoordinatorRuntime & { + getOrchestrationDb(): OrchestrationDb + resolveOrchestrationTargetKey(selector?: string): Promise + listWorktreeLineage(): Promise> + getStartedAt(): number + // probes + createdWorktrees: { name: string }[] + launchAgentCalls: { worktree: string }[] + sent: { handle: string }[] +} + +function makeRuntime(opts: { + db: OrchestrationDb + lineage?: Record + resolveTarget?: (selector?: string) => Promise + startedAt?: number +}): FakeRuntime { + const terminals: Terminal[] = [] + const rt: FakeRuntime = { + createdWorktrees: [], + launchAgentCalls: [], + sent: [], + getOrchestrationDb: () => opts.db, + getStartedAt: () => opts.startedAt ?? 1_000_000, + resolveOrchestrationTargetKey: + opts.resolveTarget ?? (async (selector?: string) => `key:${selector ?? ''}`), + listWorktreeLineage: async () => opts.lineage ?? {}, + async sendTerminal(handle: string) { + rt.sent.push({ handle }) + return { handle, accepted: true } + }, + async listTerminals() { + return { terminals } + }, + async createTerminal(worktree?: string) { + const handle = `term_${terminals.length}` + const worktreeId = worktree?.replace(/^id:/, '') ?? 'wt' + terminals.push({ handle, worktreeId, connected: true, writable: true }) + return { handle, worktreeId } + }, + async launchAgentTerminal(worktree: string) { + rt.launchAgentCalls.push({ worktree }) + const handle = `term_agent_${rt.launchAgentCalls.length - 1}` + terminals.push({ + handle, + worktreeId: worktree.replace(/^id:/, ''), + connected: true, + writable: true + }) + return { handle, worktreeId: worktree.replace(/^id:/, '') } + }, + async waitForTerminal(handle: string) { + return { handle, condition: 'tui-idle' } + }, + async probeWorktreeDrift() { + return null + }, + async createWorktree(o: { name: string }) { + rt.createdWorktrees.push({ name: o.name }) + const worktreeId = `wt_new_${rt.createdWorktrees.length}` + const handle = `term_new_${rt.createdWorktrees.length}` + terminals.push({ handle, worktreeId, connected: true, writable: true }) + return { worktreeId, branch: o.name, terminalHandle: handle } + }, + async removeWorktree() {} + } + return rt +} + +function lineageChild(p: { + worktreeId: string + parentWorktreeId: string + orchestrationRunId: string + taskId: string +}): WorktreeLineage { + return { + worktreeId: p.worktreeId, + worktreeInstanceId: `${p.worktreeId}-inst`, + parentWorktreeId: p.parentWorktreeId, + parentWorktreeInstanceId: `${p.parentWorktreeId}-inst`, + origin: 'orchestration', + capture: { source: 'orchestration-context', confidence: 'explicit' }, + orchestrationRunId: p.orchestrationRunId, + taskId: p.taskId, + createdAt: 0 + } +} + +function completeDispatchedTask(db: OrchestrationDb, taskId: string): void { + const dispatch = db.getDispatchContext(taskId) + if (!dispatch) { + throw new Error(`no dispatch for ${taskId}`) + } + db.insertMessage({ + from: dispatch.assignee_handle ?? 'term', + to: 'coord', + subject: 'done', + type: 'worker_done', + payload: JSON.stringify({ taskId, dispatchId: dispatch.id }) + }) +} + +async function waitFor(cond: () => boolean, timeoutMs = 3000): Promise { + const start = Date.now() + while (!cond()) { + if (Date.now() - start > timeoutMs) { + throw new Error('waitFor timeout') + } + await new Promise((r) => { + setTimeout(r, 5) + }) + } +} + +describe('runOrchestrationBootReconcile — real resume path (F3 #14 round 2)', () => { + let db: OrchestrationDb | undefined + afterEach(() => { + db?.close() + db = undefined + }) + + it('resumes a worktree-backed run and re-adopts its track worktree (no duplicate)', async () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'coord', + targetKey: 'worktree:director', + pollIntervalMs: 10, + worktreeBacked: true, + workerAgent: 'claude' + }) + const task = d.createTask({ + spec: 'track: alpha\nimplement', + coordinatorRunId: run.id, + targetKey: 'worktree:director' + }) + const runtime = makeRuntime({ + db: d, + lineage: { + alphaChild: lineageChild({ + worktreeId: 'wt-existing', + parentWorktreeId: 'director', + orchestrationRunId: run.id, + taskId: task.id + }) + } + }) + + const results = await runOrchestrationBootReconcile(runtime) + expect(results[0].disposition).toBe('resumed') + + // The loop relaunched the agent in the EXISTING checkout — no new worktree forked. + await waitFor(() => runtime.launchAgentCalls.length === 1) + expect(runtime.createdWorktrees).toHaveLength(0) + expect(runtime.launchAgentCalls[0].worktree).toBe('id:wt-existing') + + // Drive it to convergence so the run leaves 'running' cleanly. + completeDispatchedTask(d, task.id) + await waitFor(() => d.getCoordinatorRun(run.id)?.status === 'completed') + }) + + it('HARD RULE: a throwing pre-loop DB call ends the run failed, never left running', async () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'coord', + targetKey: 'worktree:director', + pollIntervalMs: 10, + worktreeBacked: true, + workerAgent: 'claude' + }) + d.createTask({ spec: 'A', coordinatorRunId: run.id, targetKey: 'worktree:director' }) + // Simulate a transient failure on the loop's first DB call (adoptUnownedTasks + // runs at the top of executeLoop). Round-1 code ran it OUTSIDE the try and had + // no .catch on the detached loop → the run was left 'running' with no loop. + d.adoptUnownedTasks = () => { + throw new Error('adopt boom') + } + const runtime = makeRuntime({ db: d }) + + const results = await runOrchestrationBootReconcile(runtime) + // resume() returns 'resumed' (the loop was fired) — but the loop must converge + // the run to failed, never leave it running. + expect(results[0].disposition).toBe('resumed') + await waitFor(() => d.getCoordinatorRun(run.id)?.status === 'failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('declines (→ failed) a pre-v9 run whose worktree_backed is unknown (NULL)', async () => { + const d = (db = new OrchestrationDb(':memory:')) + // No worktreeBacked passed → stored NULL (simulating a pre-v9 row with a target). + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'coord', + targetKey: 'worktree:director', + pollIntervalMs: 10 + }) + d.createTask({ spec: 'A', coordinatorRunId: run.id, targetKey: 'worktree:director' }) + const runtime = makeRuntime({ db: d }) + + const results = await runOrchestrationBootReconcile(runtime) + // Round-1 code resumed NULL as legacy (→ 'resumed', stays running). Now: failed. + expect(results[0].disposition).toBe('failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + expect(runtime.launchAgentCalls).toHaveLength(0) + }) + + it('declines (→ failed) when the director worktree no longer resolves', async () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'coord', + targetKey: 'worktree:gone', + pollIntervalMs: 10, + worktreeBacked: true, + workerAgent: 'claude' + }) + d.createTask({ spec: 'A', coordinatorRunId: run.id, targetKey: 'worktree:gone' }) + const runtime = makeRuntime({ + db: d, + resolveTarget: async () => { + throw new Error('worktree_not_found') + } + }) + + const results = await runOrchestrationBootReconcile(runtime) + expect(results[0].disposition).toBe('failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('declines (→ failed) a run with no worktree target', async () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'coord', + pollIntervalMs: 10 + }) + d.createTask({ spec: 'A', coordinatorRunId: run.id }) + const runtime = makeRuntime({ db: d }) + + const results = await runOrchestrationBootReconcile(runtime) + expect(results[0].disposition).toBe('failed') + expect(d.getCoordinatorRun(run.id)?.status).toBe('failed') + }) + + it('does not double-drive: a second runtime at the same boot fence is contended → skipped', async () => { + const d = (db = new OrchestrationDb(':memory:')) + const run = d.startCoordinatorRun({ + spec: 'go', + coordinatorHandle: 'coord', + targetKey: 'worktree:director', + pollIntervalMs: 10, + worktreeBacked: true, + workerAgent: 'claude' + }) + const task = d.createTask({ + spec: 'track: alpha\nimplement', + coordinatorRunId: run.id, + targetKey: 'worktree:director' + }) + const lineage = { + alphaChild: lineageChild({ + worktreeId: 'wt-existing', + parentWorktreeId: 'director', + orchestrationRunId: run.id, + taskId: task.id + }) + } + // Same boot fence for both runtimes → the second's atomic claim must lose. + const rtA = makeRuntime({ db: d, lineage, startedAt: 5_000_000 }) + const rtB = makeRuntime({ db: d, lineage, startedAt: 5_000_000 }) + + const a = await runOrchestrationBootReconcile(rtA) + expect(a[0].disposition).toBe('resumed') + await waitFor(() => rtA.launchAgentCalls.length === 1) + + // Second runtime reconciles the SAME still-running run: must hand off, not drive. + const b = await runOrchestrationBootReconcile(rtB) + expect(b[0].disposition).toBe('skipped') + expect(rtB.launchAgentCalls).toHaveLength(0) + expect(d.getCoordinatorRun(run.id)?.status).toBe('running') + + // Clean up the live loop. + completeDispatchedTask(d, task.id) + await waitFor(() => d.getCoordinatorRun(run.id)?.status === 'completed') + }) +})