From 10acf17c6e371e0c688b1795b700d85f8782fc87 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 5 Jun 2026 00:16:14 -0700 Subject: [PATCH] feat(engine): cli-agent resume coordinator and self-healing integration (U8) Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/cli-agent-resume-coordinator.md | 24 ++ .../self-healing-cli-sessions.test.ts | 161 ++++++++ .../__tests__/resume-coordinator.test.ts | 345 +++++++++++++++++ .../src/cli-agent/resume-coordinator.ts | 354 ++++++++++++++++++ .../engine/src/cli-agent/session-manager.ts | 82 +++- packages/engine/src/index.ts | 10 + packages/engine/src/self-healing.ts | 46 +++ packages/engine/src/stuck-task-detector.ts | 35 ++ 8 files changed, 1042 insertions(+), 15 deletions(-) create mode 100644 .changeset/cli-agent-resume-coordinator.md create mode 100644 packages/engine/src/__tests__/self-healing-cli-sessions.test.ts create mode 100644 packages/engine/src/cli-agent/__tests__/resume-coordinator.test.ts create mode 100644 packages/engine/src/cli-agent/resume-coordinator.ts diff --git a/.changeset/cli-agent-resume-coordinator.md b/.changeset/cli-agent-resume-coordinator.md new file mode 100644 index 0000000000..f191adf0a5 --- /dev/null +++ b/.changeset/cli-agent-resume-coordinator.md @@ -0,0 +1,24 @@ +--- +"@runfusion/fusion": minor +--- + +Add the CLI agent resume coordinator and self-healing integration (U8). On +engine start, sessions persisted as live (starting / ready / busy / +waitingOnInput) are classified `engineDeath` and queued for resume respecting +the session-manager concurrency ceiling. Resume verifies the recorded worktree +still exists (missing → needsAttention, never a CLI spawned into a vanished +directory), detects a dirty worktree (logged + flagged on the session record, +resume proceeds), relaunches via the adapter's `buildResume` with the recorded +native session id in the recorded worktree, re-attaches telemetry, and +re-injects no prompt. Only `crashed`/`engineDeath` are resume-eligible +(`killed`/`userExited`/`authFailed`/`completed` never); attempts are capped at 2 +with backoff; exhaustion, an unsupported adapter, a missing vendor session +store, or an immediate spawn error route to needsAttention (a permanent-failure +path, not a retry loop). + +Self-healing idle-worktree sweeps (`enforceWorktreeCap`, `cleanupOrphans`, +unregistered-orphan reap) now skip a worktree backing a resume-eligible +`cli_sessions` record via a narrow `isWorktreeResumeReserved` seam, and the +stuck-task detector suppresses stuck/inactivity flagging while a task's CLI +session is `waitingOnInput` via a narrow `isCliSessionWaitingOnInput` seam — the +U3 stall backstop remains the only escalation while genuinely waiting. diff --git a/packages/engine/src/__tests__/self-healing-cli-sessions.test.ts b/packages/engine/src/__tests__/self-healing-cli-sessions.test.ts new file mode 100644 index 0000000000..2dfe8144b3 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-cli-sessions.test.ts @@ -0,0 +1,161 @@ +/** + * U8 — self-healing + stuck-detector CLI-session awareness. + * + * Idle-worktree sweeps (enforceWorktreeCap, cleanupOrphans, reapUnregisteredOrphans) + * must SKIP a worktree backing a resume-eligible cli_sessions record; the stuck + * detector must suppress stuck/inactivity flagging while a task's CLI session is + * waitingOnInput, yet still flag a genuinely-quiet session. + * + * The sweeps + module functions are exercised through the narrow seams + * (isWorktreeResumeReserved option / isCliSessionWaitingOnInput option) with the + * heavy git/FS dependencies mocked. + */ +import { EventEmitter } from "node:events"; +import { mkdtempSync, mkdirSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { beforeEach, afterEach, describe, expect, it, vi } from "vitest"; +import type { TaskStore } from "@fusion/core"; + +import { SelfHealingManager } from "../self-healing.js"; +import * as worktreePool from "../worktree-pool.js"; +import { StuckTaskDetector, type DisposableSession } from "../stuck-task-detector.js"; + +function createStore(settings: Record): TaskStore & EventEmitter { + const emitter = new EventEmitter() as TaskStore & EventEmitter; + (emitter as any).getSettings = vi.fn().mockResolvedValue(settings); + (emitter as any).listTasks = vi.fn().mockResolvedValue([]); + return emitter; +} + +describe("self-healing idle-worktree sweeps skip resume-eligible CLI session worktrees (U8)", () => { + let rootDir: string; + let worktreesDir: string; + let reservedPath: string; + let freePath: string; + + beforeEach(() => { + rootDir = mkdtempSync(join(tmpdir(), "kb-selfheal-cli-")); + worktreesDir = join(rootDir, ".worktrees"); + mkdirSync(worktreesDir, { recursive: true }); + reservedPath = join(worktreesDir, "wt-reserved"); + freePath = join(worktreesDir, "wt-free"); + mkdirSync(reservedPath); + mkdirSync(freePath); + }); + + afterEach(() => { + rmSync(rootDir, { recursive: true, force: true }); + vi.restoreAllMocks(); + }); + + it("enforceWorktreeCap skips the reserved worktree, reaps the free one", async () => { + // cap = maxWorktrees(1) * 2 = 2; we have 2 dirs → need 3 to exceed. Add one more. + mkdirSync(join(worktreesDir, "wt-extra")); + const store = createStore({ maxWorktrees: 1, recycleWorktrees: false }); + vi.spyOn(worktreePool, "scanIdleWorktrees").mockResolvedValue([reservedPath, freePath, join(worktreesDir, "wt-extra")]); + const removeSpy = vi.spyOn(worktreePool, "removeWorktree").mockResolvedValue(undefined as never); + + const manager = new SelfHealingManager(store, { + rootDir, + isWorktreeResumeReserved: (p) => p === reservedPath, + }); + + await (manager as any).enforceWorktreeCap(); + + const removed = removeSpy.mock.calls.map((c) => (c[0] as { worktreePath: string }).worktreePath); + expect(removed).not.toContain(reservedPath); + expect(removed).toContain(freePath); + }); + + it("cleanupOrphans (recycle off) skips the reserved worktree", async () => { + const store = createStore({ recycleWorktrees: false }); + vi.spyOn(worktreePool, "scanIdleWorktrees").mockResolvedValue([reservedPath, freePath]); + const removeSpy = vi.spyOn(worktreePool, "removeWorktree").mockResolvedValue(undefined as never); + + const manager = new SelfHealingManager(store, { + rootDir, + isWorktreeResumeReserved: (p) => p === reservedPath, + }); + + const cleaned = await (manager as any).cleanupOrphans(); + + const removed = removeSpy.mock.calls.map((c) => (c[0] as { worktreePath: string }).worktreePath); + expect(removed).toEqual([freePath]); + expect(cleaned).toBe(1); + }); + + it("without the seam predicate, both worktrees are reaped (no behavior change)", async () => { + const store = createStore({ recycleWorktrees: false }); + vi.spyOn(worktreePool, "scanIdleWorktrees").mockResolvedValue([reservedPath, freePath]); + const removeSpy = vi.spyOn(worktreePool, "removeWorktree").mockResolvedValue(undefined as never); + + const manager = new SelfHealingManager(store, { rootDir }); + await (manager as any).cleanupOrphans(); + + const removed = removeSpy.mock.calls.map((c) => (c[0] as { worktreePath: string }).worktreePath); + expect(removed).toEqual([reservedPath, freePath]); + }); +}); + +// ── Stuck detector waitingOnInput suppression ──────────────────────────────── + +function fakeStore(settings: Record): TaskStore { + return { + getSettings: vi.fn().mockResolvedValue(settings), + getTask: vi.fn(), + } as unknown as TaskStore; +} + +function fakeSession(): DisposableSession { + return { dispose: vi.fn() } as unknown as DisposableSession; +} + +describe("stuck-task detector suppresses flagging while CLI session waitingOnInput (U8)", () => { + afterEach(() => vi.restoreAllMocks()); + + it("waitingOnInput session is NOT flagged; same session IS flagged once it stops waiting", async () => { + const onStuck = vi.fn(); + let waiting = true; + const store = fakeStore({ taskStuckTimeoutMs: 1000, globalPause: false, enginePaused: false }); + const detector = new StuckTaskDetector(store, { + onStuck, + isCliSessionWaitingOnInput: () => waiting, + // Accept the requeue so killAndRetry proceeds to onStuck. + beforeRequeue: async () => true, + }); + + detector.trackTask("FN-1", fakeSession()); + // Force the task to look inactive (past the 1s timeout). + (detector as any).tracked.get("FN-1").lastActivity = Date.now() - 10_000; + + // While waitingOnInput: suppressed. + await (detector as any).checkStuckTasks(); + expect(onStuck).not.toHaveBeenCalled(); + + // Once it stops waiting (genuinely quiet): the U3-style backstop equivalent + // (the detector) now flags it. + waiting = false; + // killAndRetry needs moveTask/logEntry; stub them on the store. + (store as any).moveTask = vi.fn().mockResolvedValue(undefined); + (store as any).logEntry = vi.fn().mockResolvedValue(undefined); + (store as any).getTask = vi.fn().mockResolvedValue({ id: "FN-1", status: "in-progress", steps: [], error: null }); + await (detector as any).checkStuckTasks(); + expect(onStuck).toHaveBeenCalledTimes(1); + expect(onStuck.mock.calls[0][0].taskId).toBe("FN-1"); + }); + + it("without the seam lookup, a waitingOnInput-shaped quiet task is flagged normally", async () => { + const onStuck = vi.fn(); + const store = fakeStore({ taskStuckTimeoutMs: 1000, globalPause: false, enginePaused: false }); + (store as any).moveTask = vi.fn().mockResolvedValue(undefined); + (store as any).logEntry = vi.fn().mockResolvedValue(undefined); + (store as any).getTask = vi.fn().mockResolvedValue({ id: "FN-2", status: "in-progress", steps: [], error: null }); + const detector = new StuckTaskDetector(store, { onStuck, beforeRequeue: async () => true }); + + detector.trackTask("FN-2", fakeSession()); + (detector as any).tracked.get("FN-2").lastActivity = Date.now() - 10_000; + await (detector as any).checkStuckTasks(); + expect(onStuck).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/engine/src/cli-agent/__tests__/resume-coordinator.test.ts b/packages/engine/src/cli-agent/__tests__/resume-coordinator.test.ts new file mode 100644 index 0000000000..8fcf951492 --- /dev/null +++ b/packages/engine/src/cli-agent/__tests__/resume-coordinator.test.ts @@ -0,0 +1,345 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { rm } from "node:fs/promises"; +import { Database, CliSessionStore, type CliSession } from "@fusion/core"; +import type { IPty } from "node-pty"; +import { CliSessionManager } from "../session-manager.js"; +import { CliAdapterRegistry, type CliAgentAdapter } from "../adapter.js"; +import { CliResumeCoordinator } from "../resume-coordinator.js"; + +// ── Mock PTY at the loadPtyModule seam ───────────────────────────────────── + +interface MockPty extends IPty { + written: string[]; + killed: boolean; + emitData(data: string): void; + emitExit(exitCode: number, signal?: number): void; + spawnArgs: string[]; +} + +interface MockState { + ptys: MockPty[]; + spawnCount: number; + spawnThrows?: () => Error | undefined; +} + +function makeMockPtyModule(state: MockState): typeof import("node-pty") { + return { + spawn(_file: string, args: string[] | string) { + state.spawnCount++; + const err = state.spawnThrows?.(); + if (err) throw err; + let dataCb: ((d: string) => void) | undefined; + let exitCb: ((e: { exitCode: number; signal?: number }) => void) | undefined; + const mock: MockPty = { + pid: 1000 + state.ptys.length, + cols: 80, + rows: 24, + process: "mock", + handleFlowControl: false, + written: [], + killed: false, + spawnArgs: Array.isArray(args) ? args : [args], + onData: (cb: (d: string) => void) => { + dataCb = cb; + return { dispose() {} }; + }, + onExit: (cb: (e: { exitCode: number; signal?: number }) => void) => { + exitCb = cb; + return { dispose() {} }; + }, + on() {}, + write() {}, + resize() {}, + clear() {}, + kill() { + mock.killed = true; + }, + pause() {}, + resume() {}, + emitData(d: string) { + dataCb?.(d); + }, + emitExit(exitCode: number, signal?: number) { + exitCb?.({ exitCode, signal }); + }, + } as unknown as MockPty; + state.ptys.push(mock); + return mock as unknown as IPty; + }, + } as unknown as typeof import("node-pty"); +} + +// ── Test adapter ─────────────────────────────────────────────────────────── + +function makeAdapter(overrides: Partial = {}): CliAgentAdapter { + return { + id: "test-cli", + name: "Test CLI", + capabilities: { + nativeDone: true, + nativeWaiting: true, + transcriptSource: "hooks", + supportsResume: true, + }, + buildLaunch: () => ({ command: "test-cli", args: ["--interactive"] }), + buildEnvAllowlist: () => ["PATH", "HOME"], + createReadinessDetector: () => ({ observe: () => true }), + formatInjection: (text) => ({ payload: `${text}\r` }), + buildResume: (ctx) => ({ command: "test-cli", args: ["--resume", ctx.nativeSessionId] }), + ...overrides, + }; +} + +// ── Harness ────────────────────────────────────────────────────────────── + +interface Harness { + manager: CliSessionManager; + registry: CliAdapterRegistry; + store: CliSessionStore; + state: MockState; + db: Database; + tmpDir: string; +} + +function makeHarness(opts?: { adapter?: CliAgentAdapter; ceiling?: number }): Harness { + const tmpDir = mkdtempSync(join(tmpdir(), "kb-cli-resume-test-")); + const fusionDir = join(tmpDir, ".fusion"); + const db = new Database(fusionDir, { inMemory: true }); + db.init(); + const store = new CliSessionStore(fusionDir, db); + const registry = new CliAdapterRegistry(); + registry.register(opts?.adapter ?? makeAdapter()); + const state: MockState = { ptys: [], spawnCount: 0 }; + const manager = new CliSessionManager({ + registry, + store, + concurrencyCeiling: opts?.ceiling ?? 8, + loadPty: async () => makeMockPtyModule(state), + }); + return { manager, registry, store, state, db, tmpDir }; +} + +/** Seed a record as the engine would persist a live session before dying. */ +function seedSession( + store: CliSessionStore, + worktreePath: string, + over: Partial = {}, +): CliSession { + const rec = store.createSession({ + adapterId: over.adapterId ?? "test-cli", + projectId: "proj-1", + purpose: "execute", + taskId: over.taskId ?? "FN-1", + worktreePath, + nativeSessionId: "nativeSessionId" in over ? over.nativeSessionId : "native-abc", + agentState: over.agentState ?? "busy", + terminationReason: over.terminationReason ?? null, + resumeAttempts: over.resumeAttempts ?? 0, + autonomyPosture: over.autonomyPosture ?? null, + }); + return rec; +} + +function makeCoordinator(h: Harness, over?: Partial[0]>) { + return new CliResumeCoordinator({ + store: h.store, + manager: h.manager, + registry: h.registry, + worktreeExists: () => true, + isWorktreeDirty: async () => false, + ...over, + }); +} + +describe("CliResumeCoordinator (U8)", () => { + const harnesses: Harness[] = []; + function track(h: Harness): Harness { + harnesses.push(h); + return h; + } + beforeEach(() => {}); + afterEach(async () => { + for (const h of harnesses) { + h.manager.dispose?.(); + h.db.close?.(); + await rm(h.tmpDir, { recursive: true, force: true }).catch(() => {}); + } + harnesses.length = 0; + }); + + it("AE3/F4: resumes a live-on-restart session via buildResume with the recorded native id; record intact; no duplicate on a second run", async () => { + const h = track(makeHarness()); + const rec = seedSession(h.store, h.tmpDir, { agentState: "busy", nativeSessionId: "native-xyz" }); + const coord = makeCoordinator(h); + + const results = await coord.recoverOnStart(); + expect(results).toHaveLength(1); + expect(results[0].disposition).toBe("resumed"); + + // Spawned via buildResume with the recorded native id. + expect(h.state.spawnCount).toBe(1); + expect(h.state.ptys[0].spawnArgs).toEqual(["--resume", "native-xyz"]); + + // Manager owns the session; record reused (state back to starting), not duplicated. + expect(h.manager.isLive(rec.id)).toBe(true); + expect(h.store.listSessions()).toHaveLength(1); + const after = h.store.getSession(rec.id)!; + expect(after.taskId).toBe("FN-1"); + + // Second sweep: session is live → no duplicate spawn. + const second = await coord.recoverOnStart(); + expect(second).toHaveLength(0); // already live, filtered out + expect(h.state.spawnCount).toBe(1); + expect(h.store.listSessions()).toHaveLength(1); + }); + + it("never resumes killed or userExited records across sweeps", async () => { + const h = track(makeHarness()); + // A dead record carrying killed/userExited is not in the orphaned-live set, + // so recoverOnStart never touches it; and resumeOne routes it to attention. + const killed = seedSession(h.store, h.tmpDir, { agentState: "dead", terminationReason: "killed", taskId: "FN-K" }); + const userExited = seedSession(h.store, h.tmpDir, { agentState: "dead", terminationReason: "userExited", taskId: "FN-U" }); + const coord = makeCoordinator(h); + + const results = await coord.recoverOnStart(); + expect(results).toHaveLength(0); // neither is orphaned-live + expect(h.state.spawnCount).toBe(0); + + // Direct disposition is ineligible (no spawn). + expect((await coord.resumeOne(killed)).disposition).toBe("needsAttention-ineligible"); + expect((await coord.resumeOne(userExited)).disposition).toBe("needsAttention-ineligible"); + expect(h.state.spawnCount).toBe(0); + }); + + it("authFailed → needsAttention without a resume attempt", async () => { + const h = track(makeHarness()); + const rec = seedSession(h.store, h.tmpDir, { agentState: "dead", terminationReason: "authFailed" }); + const coord = makeCoordinator(h); + const res = await coord.resumeOne(rec); + expect(res.disposition).toBe("needsAttention-ineligible"); + expect(h.state.spawnCount).toBe(0); + expect(h.store.getSession(rec.id)!.agentState).toBe("needsAttention"); + }); + + it("cap: two failures → needsAttention, no third spawn across cycles", async () => { + const h = track(makeHarness()); + // Spawn always throws (vendor store / spawn error). + h.state.spawnThrows = () => new Error("spawn failed"); + const rec = seedSession(h.store, h.tmpDir, { agentState: "busy" }); + const coord = makeCoordinator(h, { maxResumeAttempts: 2 }); + + // First sweep: spawn throws → immediate permanent-failure path → needsAttention. + const r1 = await coord.recoverOnStart(); + expect(r1[0].disposition).toBe("needsAttention-spawnError"); + expect(h.store.getSession(rec.id)!.agentState).toBe("needsAttention"); + const spawnsAfter1 = h.state.spawnCount; + + // Subsequent sweeps: record is no longer orphaned-live → never spawned again. + await coord.recoverOnStart(); + await coord.recoverOnStart(); + expect(h.state.spawnCount).toBe(spawnsAfter1); + }); + + it("missing vendor store (no native id) → permanent-failure path, not retry loop", async () => { + const h = track(makeHarness()); + const rec = seedSession(h.store, h.tmpDir, { agentState: "busy", nativeSessionId: null }); + const coord = makeCoordinator(h); + const res = await coord.resumeOne(rec); + expect(res.disposition).toBe("needsAttention-spawnError"); + expect(h.state.spawnCount).toBe(0); + expect(h.store.getSession(rec.id)!.agentState).toBe("needsAttention"); + }); + + it("missing worktree → needsAttention without spawning", async () => { + const h = track(makeHarness()); + const rec = seedSession(h.store, h.tmpDir, { agentState: "busy" }); + const coord = makeCoordinator(h, { worktreeExists: () => false }); + const res = await coord.resumeOne(rec); + expect(res.disposition).toBe("needsAttention-missingWorktree"); + expect(h.state.spawnCount).toBe(0); + expect(h.store.getSession(rec.id)!.agentState).toBe("needsAttention"); + }); + + it("adapter without resume support → needsAttention with clear reason, no spawn", async () => { + const h = track( + makeHarness({ + adapter: makeAdapter({ + capabilities: { + nativeDone: true, + nativeWaiting: true, + transcriptSource: "hooks", + supportsResume: false, + }, + buildResume: undefined, + }), + }), + ); + const rec = seedSession(h.store, h.tmpDir, { agentState: "busy" }); + const coord = makeCoordinator(h); + const res = await coord.resumeOne(rec); + expect(res.disposition).toBe("needsAttention-resumeUnsupported"); + expect(h.state.spawnCount).toBe(0); + expect(h.store.getSession(rec.id)!.agentState).toBe("needsAttention"); + }); + + it("dirty worktree → flagged on the record, resume proceeds", async () => { + const h = track(makeHarness()); + const rec = seedSession(h.store, h.tmpDir, { agentState: "busy" }); + const coord = makeCoordinator(h, { isWorktreeDirty: async () => true }); + const res = await coord.resumeOne(rec); + expect(res.disposition).toBe("resumed"); + expect(res.dirtyWorktree).toBe(true); + expect(h.state.spawnCount).toBe(1); + const after = h.store.getSession(rec.id)!; + expect(after.autonomyPosture?.resumeDirtyWorktree).toBe(true); + }); + + it("re-attaches telemetry on resume and injects no prompt", async () => { + const h = track(makeHarness()); + seedSession(h.store, h.tmpDir, { agentState: "busy" }); + const reattached: string[] = []; + const coord = makeCoordinator(h, { + reattachTelemetry: (s) => { + reattached.push(s.id); + }, + }); + await coord.recoverOnStart(); + expect(reattached).toHaveLength(1); + // No prompt injected: the resume PTY received no writes. + expect(h.state.ptys[0].written.join("")).toBe(""); + }); + + it("respects the concurrency ceiling: queues remaining sessions for the next sweep", async () => { + const h = track(makeHarness({ ceiling: 1 })); + seedSession(h.store, h.tmpDir, { agentState: "busy", taskId: "FN-A", nativeSessionId: "n-a" }); + seedSession(h.store, h.tmpDir, { agentState: "busy", taskId: "FN-B", nativeSessionId: "n-b" }); + const coord = makeCoordinator(h); + const results = await coord.recoverOnStart(); + const resumed = results.filter((r) => r.disposition === "resumed"); + const skipped = results.filter((r) => r.disposition === "skipped-noCapacity"); + expect(resumed).toHaveLength(1); + expect(skipped).toHaveLength(1); + expect(h.state.spawnCount).toBe(1); + }); + + it("resumeReservedWorktrees: reports worktrees backing resume-eligible records", () => { + const h = track(makeHarness()); + seedSession(h.store, h.tmpDir, { agentState: "busy", taskId: "FN-live" }); + seedSession(h.store, h.tmpDir, { agentState: "dead", terminationReason: "killed", taskId: "FN-killed", worktreePath: h.tmpDir } as Partial); + const coord = makeCoordinator(h); + const reserved = coord.resumeReservedWorktrees(); + expect(reserved.has(h.tmpDir)).toBe(true); + + // An exhausted record is NOT reserved. + const exhausted = seedSession(h.store, h.tmpDir, { + agentState: "dead", + terminationReason: "crashed", + resumeAttempts: 2, + taskId: "FN-exh", + }); + expect(coord.isRecordResumeEligible(exhausted)).toBe(false); + }); +}); diff --git a/packages/engine/src/cli-agent/resume-coordinator.ts b/packages/engine/src/cli-agent/resume-coordinator.ts new file mode 100644 index 0000000000..e40ddf036e --- /dev/null +++ b/packages/engine/src/cli-agent/resume-coordinator.ts @@ -0,0 +1,354 @@ +/** + * CliResumeCoordinator — engine-restart recovery for CLI agent sessions + * (CLI Agent Executor, U8). + * + * On engine start, sessions persisted as live (starting / ready / busy / + * waitingOnInput) were orphaned by the engine's death — there is no live PTY + * behind them. This coordinator finds those records, classifies them as + * `engineDeath`, and queues a resume that respects the session-manager + * concurrency ceiling. + * + * Resume semantics (KTD — termination taxonomy, resume-the-CLI): + * - Eligibility: ONLY `crashed` and `engineDeath` are resume-eligible + * (`isResumeEligible`). `killed` / `userExited` are never auto-resumed; + * `authFailed` and `completed` are never resumed. A record found live on + * restart is reclassified to `engineDeath` (it had no chance to record a + * terminal reason), making it eligible. + * - Worktree-existence precondition: the recorded worktree MUST still exist — + * a missing worktree routes the session to `needsAttention`, NEVER a CLI + * spawned into a vanished directory. + * - Dirty-tree detection: a dirty `git status` is logged and flagged on the + * session record (under `autonomyPosture.resumeDirtyWorktree`), then resume + * PROCEEDS — the flag surfaces to the UI. + * - Relaunch: via the manager's resume path (adapter `buildResume` with the + * recorded `nativeSessionId`, in the recorded worktree). Telemetry is + * re-attached (a fresh hook token + scripts via `wireTelemetry`/the hub). + * NO prompt is re-injected — scrollback replays to viewers, the agent + * continues from its own native transcript. + * - Attempt cap: 2 attempts with backoff (tracked on `resumeAttempts`). + * Exhaustion, an adapter without resume support, a missing vendor session + * store, or an immediate spawn error route to `needsAttention` (a permanent + * failure path, NOT an infinite retry loop). + * + * The coordinator NEVER imports dashboard code. The worktree-existence check + * and the dirty-tree probe are injected seams so tests need no real git/FS. + */ + +import { existsSync } from "node:fs"; +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import type { CliSession, CliSessionStore, CliTerminationReason } from "@fusion/core"; +import type { CliSessionManager } from "./session-manager.js"; +import { CliConcurrencyLimitError, CliResumeUnsupportedError } from "./session-manager.js"; +import type { CliAdapterRegistry } from "./adapter.js"; +import { isResumeEligible } from "./state-machine.js"; + +const execFileAsync = promisify(execFile); + +/** Persisted-as-live states that, found on restart, imply an orphaned PTY. */ +const ORPHANED_LIVE_STATES = new Set([ + "starting", + "ready", + "busy", + "waitingOnInput", +]); + +/** Default resume attempt cap (KTD = 2). */ +export const DEFAULT_MAX_RESUME_ATTEMPTS = 2; +/** Default base backoff (ms) between resume attempts; doubled per attempt. */ +export const DEFAULT_RESUME_BACKOFF_BASE_MS = 1000; + +/** Outcome of a single session's resume disposition. */ +export type ResumeDisposition = + | "resumed" + | "needsAttention-missingWorktree" + | "needsAttention-ineligible" + | "needsAttention-exhausted" + | "needsAttention-resumeUnsupported" + | "needsAttention-spawnError" + | "skipped-noCapacity"; + +export interface ResumeResult { + sessionId: string; + taskId: string | null; + disposition: ResumeDisposition; + /** Whether the worktree was dirty at resume (flag also persisted on the record). */ + dirtyWorktree?: boolean; + /** Reason string for needsAttention dispositions. */ + reason?: string; +} + +export interface CliResumeCoordinatorOptions { + store: CliSessionStore; + manager: CliSessionManager; + registry: CliAdapterRegistry; + /** + * Re-attach telemetry for a resumed session — typically wires a fresh hook + * token + scripts via the TelemetryHub. Called AFTER a successful relaunch and + * BEFORE returning. Best-effort; a throw is logged, never fatal to the sweep. + */ + reattachTelemetry?: (session: CliSession) => void | Promise; + /** Max resume attempts before needsAttention. Default 2 (KTD). */ + maxResumeAttempts?: number; + /** Base backoff (ms); doubled per prior attempt. Default 1000. */ + resumeBackoffBaseMs?: number; + /** Worktree-existence probe (injected for tests). Default `fs.existsSync`. */ + worktreeExists?: (worktreePath: string) => boolean; + /** + * Dirty-tree probe (injected for tests). Returns true when `git status` shows + * uncommitted changes. Default: runs `git status --porcelain` in the worktree. + */ + isWorktreeDirty?: (worktreePath: string) => Promise; + /** Best-effort logger. */ + log?: (msg: string) => void; +} + +/** Default dirty-tree probe: `git status --porcelain` is non-empty. */ +async function defaultIsWorktreeDirty(worktreePath: string): Promise { + try { + const { stdout } = await execFileAsync("git", ["status", "--porcelain"], { + cwd: worktreePath, + }); + return stdout.trim().length > 0; + } catch { + // Not a git worktree / git unavailable: treat as not-dirty (don't block resume). + return false; + } +} + +export class CliResumeCoordinator { + private readonly store: CliSessionStore; + private readonly manager: CliSessionManager; + private readonly registry: CliAdapterRegistry; + private readonly reattachTelemetry?: (session: CliSession) => void | Promise; + private readonly maxResumeAttempts: number; + private readonly resumeBackoffBaseMs: number; + private readonly worktreeExists: (worktreePath: string) => boolean; + private readonly isWorktreeDirty: (worktreePath: string) => Promise; + private readonly log: (msg: string) => void; + + constructor(opts: CliResumeCoordinatorOptions) { + this.store = opts.store; + this.manager = opts.manager; + this.registry = opts.registry; + this.reattachTelemetry = opts.reattachTelemetry; + this.maxResumeAttempts = opts.maxResumeAttempts ?? DEFAULT_MAX_RESUME_ATTEMPTS; + this.resumeBackoffBaseMs = opts.resumeBackoffBaseMs ?? DEFAULT_RESUME_BACKOFF_BASE_MS; + this.worktreeExists = opts.worktreeExists ?? ((p) => existsSync(p)); + this.isWorktreeDirty = opts.isWorktreeDirty ?? defaultIsWorktreeDirty; + this.log = opts.log ?? (() => {}); + } + + /** + * The set of worktree paths backing resume-eligible session records. Exposed + * for the self-healing seam so idle-worktree sweeps treat them as in-use. A + * record is resume-eligible if it is found live-on-restart (→ engineDeath) or + * already carries a resume-eligible termination reason AND has not exhausted + * its attempt cap. The path is `resolve`d-free (raw recorded path); callers + * normalize as needed. + */ + resumeReservedWorktrees(): Set { + const reserved = new Set(); + for (const session of this.store.listSessions()) { + if (!session.worktreePath) continue; + if (!this.isRecordResumeEligible(session)) continue; + reserved.add(session.worktreePath); + } + return reserved; + } + + /** Whether a recorded session is currently resume-eligible (for sweep skipping). */ + isRecordResumeEligible(session: CliSession): boolean { + if (session.resumeAttempts >= this.maxResumeAttempts) return false; + // Found-live-on-restart → engineDeath (eligible). + if (ORPHANED_LIVE_STATES.has(session.agentState)) return true; + // Reaped-but-resumable: a dead record whose recorded reason is resume-eligible. + if ( + session.agentState === "dead" && + session.terminationReason != null && + isResumeEligible(session.terminationReason) + ) { + return true; + } + return false; + } + + /** + * Engine-start sweep. Finds orphaned-live sessions, classifies engineDeath, + * and resumes each (respecting the manager's concurrency ceiling). Returns a + * per-session disposition list. Idempotent: a second run after a successful + * resume finds the session live (re-spawned record is `starting`/`ready`) but + * the manager's `isLive` guard prevents a duplicate spawn — see `resumeOne`. + */ + async recoverOnStart(): Promise { + const candidates = this.store + .listSessions() + .filter((s) => ORPHANED_LIVE_STATES.has(s.agentState)) + // Never reclaim a session the manager already owns (idempotent re-run). + .filter((s) => !this.manager.isLive(s.id)); + + const results: ResumeResult[] = []; + for (const session of candidates) { + // Concurrency ceiling: stop queuing once slots are exhausted. The + // remaining records stay persisted-live and are picked up next sweep. + if (this.manager.availableSlots() <= 0) { + results.push({ + sessionId: session.id, + taskId: session.taskId, + disposition: "skipped-noCapacity", + }); + continue; + } + results.push(await this.resumeOne(session)); + } + return results; + } + + /** + * Resume a single orphaned session through the eligibility predicate and + * worktree precondition. Public for targeted tests. + */ + async resumeOne(session: CliSession): Promise { + const base = { sessionId: session.id, taskId: session.taskId }; + + // Idempotency: the manager already owns a live PTY for this id → no-op. + if (this.manager.isLive(session.id)) { + return { ...base, disposition: "resumed" }; + } + + // Reclassify a found-live record to engineDeath (it never recorded a reason). + // A dead record keeps its recorded reason (crashed / killed / userExited / …). + const reason: CliTerminationReason = ORPHANED_LIVE_STATES.has(session.agentState) + ? "engineDeath" + : session.terminationReason ?? "engineDeath"; + + // Eligibility predicate: only crashed / engineDeath ever resume. + if (!isResumeEligible(reason)) { + this.toNeedsAttention(session, reason, `ineligible termination reason: ${reason}`); + return { ...base, disposition: "needsAttention-ineligible", reason }; + } + + // Attempt-cap exhaustion → permanent needsAttention (never a third spawn). + if (session.resumeAttempts >= this.maxResumeAttempts) { + this.toNeedsAttention(session, reason, `resume attempts exhausted (${session.resumeAttempts})`); + return { ...base, disposition: "needsAttention-exhausted", reason }; + } + + // Worktree-existence precondition: never spawn into a vanished directory. + const worktreePath = session.worktreePath; + if (!worktreePath || !this.worktreeExists(worktreePath)) { + this.toNeedsAttention(session, reason, `recorded worktree missing: ${worktreePath ?? ""}`); + return { ...base, disposition: "needsAttention-missingWorktree", reason }; + } + + // Adapter must support resume. + const adapter = (() => { + try { + return this.registry.get(session.adapterId); + } catch { + return undefined; + } + })(); + if (!adapter || !adapter.capabilities.supportsResume || typeof adapter.buildResume !== "function") { + this.toNeedsAttention(session, reason, `adapter does not support resume: ${session.adapterId}`); + return { ...base, disposition: "needsAttention-resumeUnsupported", reason }; + } + + // Vendor session store precondition: a captured native id is required. + if (!session.nativeSessionId) { + this.toNeedsAttention(session, reason, "missing native session id (no vendor session store)"); + return { ...base, disposition: "needsAttention-spawnError", reason }; + } + + // Dirty-tree detection: log + flag, then PROCEED. + let dirty = false; + try { + dirty = await this.isWorktreeDirty(worktreePath); + } catch { + dirty = false; + } + if (dirty) { + this.log(`[cli-resume] session ${session.id}: worktree dirty at resume — flagged, proceeding`); + this.flagDirty(session); + } + + // Relaunch via the manager's resume path (adapter buildResume + native id), + // reusing the existing record so no duplicate session row is created. + try { + await this.manager.spawn({ + adapterId: session.adapterId, + projectId: session.projectId, + purpose: session.purpose, + taskId: session.taskId, + chatSessionId: session.chatSessionId, + worktreePath, + posture: session.autonomyPosture, + resume: { sessionId: session.id, nativeSessionId: session.nativeSessionId }, + }); + } catch (err) { + // Immediate spawn failure / unsupported resume / missing vendor store → + // permanent-failure path. Record the attempt and route to needsAttention + // once the cap is reached; otherwise leave it for the next sweep (backoff). + const attempts = session.resumeAttempts + 1; + this.store.updateSession(session.id, { resumeAttempts: attempts }); + const msg = err instanceof Error ? err.message : String(err); + const isUnsupported = err instanceof CliResumeUnsupportedError; + const isCeiling = err instanceof CliConcurrencyLimitError; + if (isCeiling) { + // Capacity raced away — leave persisted-live for the next sweep, no attempt charge. + this.store.updateSession(session.id, { resumeAttempts: session.resumeAttempts }); + return { ...base, disposition: "skipped-noCapacity" }; + } + this.log(`[cli-resume] session ${session.id}: resume spawn failed (${msg})`); + if (isUnsupported || attempts >= this.maxResumeAttempts) { + this.toNeedsAttention(session, reason, `resume spawn failed: ${msg}`); + return { + ...base, + disposition: isUnsupported + ? "needsAttention-resumeUnsupported" + : "needsAttention-spawnError", + reason: msg, + }; + } + // Under the cap: needsAttention is the permanent floor only at exhaustion; + // a single immediate failure (missing vendor store / spawn error) is also + // permanent per the KTD — do NOT loop. Route to needsAttention now. + this.toNeedsAttention(session, reason, `resume spawn failed: ${msg}`); + return { ...base, disposition: "needsAttention-spawnError", reason: msg }; + } + + // Re-attach telemetry (fresh hook token + scripts). Best-effort. + if (this.reattachTelemetry) { + try { + const fresh = this.store.getSession(session.id) ?? session; + await this.reattachTelemetry(fresh); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + this.log(`[cli-resume] session ${session.id}: telemetry re-attach failed (${msg}) — non-fatal`); + } + } + + this.log(`[cli-resume] session ${session.id}: resumed (native ${session.nativeSessionId}) in ${worktreePath}`); + return { ...base, disposition: "resumed", dirtyWorktree: dirty }; + } + + /** Backoff (ms) before the next resume attempt for a given attempt count. */ + backoffForAttempt(attemptsSoFar: number): number { + return this.resumeBackoffBaseMs * 2 ** attemptsSoFar; + } + + /** Route a session to needsAttention, preserving the precise termination reason. */ + private toNeedsAttention(session: CliSession, reason: CliTerminationReason, why: string): void { + this.log(`[cli-resume] session ${session.id} → needsAttention: ${why}`); + this.store.updateSession(session.id, { + agentState: "needsAttention", + terminationReason: reason, + }); + } + + /** Persist the dirty-worktree flag on the session record (extensible posture). */ + private flagDirty(session: CliSession): void { + const posture = { ...(session.autonomyPosture ?? {}), resumeDirtyWorktree: true }; + this.store.updateSession(session.id, { autonomyPosture: posture }); + } +} diff --git a/packages/engine/src/cli-agent/session-manager.ts b/packages/engine/src/cli-agent/session-manager.ts index ba3b50cd90..067aa8f0e4 100644 --- a/packages/engine/src/cli-agent/session-manager.ts +++ b/packages/engine/src/cli-agent/session-manager.ts @@ -37,7 +37,7 @@ import { } from "@fusion/core"; import { loadPtyModule } from "../pty-native.js"; import type { IPty } from "node-pty"; -import type { CliAdapterRegistry, CliAgentAdapter, CliReadinessDetector } from "./adapter.js"; +import type { CliAdapterRegistry, CliAgentAdapter, CliLaunchSpec, CliReadinessDetector } from "./adapter.js"; // ── Constants ────────────────────────────────────────────────────────────── @@ -81,6 +81,15 @@ export class UnknownCliSessionError extends Error { } } +/** Thrown when a resume is requested for an adapter that cannot resume. */ +export class CliResumeUnsupportedError extends Error { + readonly code = "CLI_RESUME_UNSUPPORTED"; + constructor(public readonly adapterId: string) { + super(`CLI adapter does not support resume: ${adapterId}`); + this.name = "CliResumeUnsupportedError"; + } +} + // ── Injection neutralization (security-critical) ─────────────────────────── /** @@ -292,6 +301,19 @@ export interface SpawnCliSessionOptions { /** Initial PTY size. */ cols?: number; rows?: number; + /** + * Resume an existing session record instead of creating a new one. When set, + * spawn builds the launch invocation via the adapter's `buildResume` (carrying + * the recorded `nativeSessionId`) and REUSES the supplied record id rather than + * minting a fresh `cli_sessions` row — so a recovered session never produces a + * duplicate record. The adapter MUST advertise `supportsResume`/`buildResume`. + */ + resume?: { + /** The existing session record id to relaunch in place. */ + sessionId: string; + /** The recorded native (vendor) session id handed to `buildResume`. */ + nativeSessionId: string; + }; } // ── Internal live-session state ───────────────────────────────────────────── @@ -379,6 +401,16 @@ export class CliSessionManager { return this.sessions.size; } + /** Configured ceiling on concurrently live PTY sessions. */ + capacity(): number { + return this.concurrencyCeiling; + } + + /** Free concurrency slots remaining before the ceiling (never negative). */ + availableSlots(): number { + return Math.max(0, this.concurrencyCeiling - this.sessions.size); + } + /** Whether a session id is currently live. */ isLive(sessionId: string): boolean { return this.sessions.has(sessionId); @@ -403,23 +435,43 @@ export class CliSessionManager { settings: (options.settings ?? {}) as Record, posture, }; - const launch = adapter.buildLaunch(launchCtx); + + // Resume vs fresh launch. A resume relaunches the recorded native session id + // via the adapter's `buildResume` and REUSES the existing record (no + // duplicate row); a fresh launch uses `buildLaunch` and mints a new record. + let launch: CliLaunchSpec; + let record: CliSession; + if (options.resume) { + if (!adapter.capabilities.supportsResume || typeof adapter.buildResume !== "function") { + throw new CliResumeUnsupportedError(options.adapterId); + } + launch = adapter.buildResume({ ...launchCtx, nativeSessionId: options.resume.nativeSessionId }); + const existing = this.store.getSession(options.resume.sessionId); + if (!existing) throw new UnknownCliSessionError(options.resume.sessionId); + // Move the reused record back to "starting" for the relaunch. + record = this.store.updateSession(options.resume.sessionId, { + agentState: "starting", + worktreePath: options.worktreePath ?? existing.worktreePath ?? null, + }) ?? existing; + } else { + launch = adapter.buildLaunch(launchCtx); + // Persist the session record BEFORE spawning so a crash mid-spawn still has + // a durable record to reason about. + record = this.store.createSession({ + adapterId: options.adapterId, + projectId: options.projectId, + purpose: options.purpose, + taskId: options.taskId ?? null, + chatSessionId: options.chatSessionId ?? null, + worktreePath: options.worktreePath ?? null, + autonomyPosture: posture, + agentState: "starting", + }); + } + const allowlist = adapter.buildEnvAllowlist(launchCtx); const env = this.buildEnv(allowlist); - // Persist the session record BEFORE spawning so a crash mid-spawn still has - // a durable record to reason about. - const record = this.store.createSession({ - adapterId: options.adapterId, - projectId: options.projectId, - purpose: options.purpose, - taskId: options.taskId ?? null, - chatSessionId: options.chatSessionId ?? null, - worktreePath: options.worktreePath ?? null, - autonomyPosture: posture, - agentState: "starting", - }); - const pty = await this.loadPty(); let child: IPty; try { diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index a0d95bc1ea..9bfd10c5f7 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -628,6 +628,7 @@ export { export { CliSessionManager, CliConcurrencyLimitError, + CliResumeUnsupportedError, UnknownCliSessionError, neutralizeInjection, DEFAULT_SCROLLBACK_BYTES, @@ -636,6 +637,15 @@ export { type CliSessionManagerOptions, type SpawnCliSessionOptions, } from "./cli-agent/session-manager.js"; +// CLI Agent Executor — resume coordinator + self-healing/stuck integration (U8). +export { + CliResumeCoordinator, + DEFAULT_MAX_RESUME_ATTEMPTS, + DEFAULT_RESUME_BACKOFF_BASE_MS, + type CliResumeCoordinatorOptions, + type ResumeResult, + type ResumeDisposition, +} from "./cli-agent/resume-coordinator.js"; export { TelemetryHub, stripAnsiControl, diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 55c603a4d9..910f28c431 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -293,6 +293,16 @@ export interface SelfHealingOptions { recoverActiveMissionValidations?: () => Promise<{ recoveredCount: number }>; /** Optional callback to reap stale mission validator runs during startup and maintenance. */ reapStaleMissionValidatorRuns?: () => Promise<{ reapedCount: number }>; + /** + * U8 (CLI Agent Executor): returns true when a worktree path backs a + * resume-eligible `cli_sessions` record. Idle-worktree sweeps + * (`enforceWorktreeCap`, `cleanupOrphans`, `reapUnregisteredOrphans`) MUST + * treat such a worktree as in-use, so a reaped-but-resumable session cannot + * have its worktree reclaimed out from under it before the resume coordinator + * relaunches the CLI. Narrow seam: a single predicate; absence (undefined) + * preserves the prior behavior. The path passed is the absolute worktree dir. + */ + isWorktreeResumeReserved?: (worktreePath: string) => boolean; } const APPROVED_TRIAGE_RECOVERY_GRACE_MS = 60_000; @@ -1551,6 +1561,27 @@ export class SelfHealingManager { } } + /** + * U8 seam: whether a worktree path backs a resume-eligible CLI agent session + * and must therefore be treated as in-use by idle sweeps. Defensive: a throw + * in the injected predicate is treated as "reserved" (conservative — never + * reclaim a worktree we can't prove is free). + */ + private isWorktreeResumeReserved(worktreePath: string): boolean { + const predicate = this.options.isWorktreeResumeReserved; + if (!predicate) return false; + try { + return predicate(resolve(worktreePath)); + } catch (err: unknown) { + log.warn( + `[self-healing] resume-reserved check threw for ${worktreePath}: ${ + err instanceof Error ? err.message : String(err) + } — treating as reserved (conservative)`, + ); + return true; + } + } + private async cleanupWorktreeOnly(task: Task): Promise { if (task.worktree && existsSync(task.worktree)) { try { @@ -8434,6 +8465,11 @@ export class SelfHealingManager { let cleaned = 0; for (const worktreePath of orphaned) { + // U8: never reclaim a worktree backing a resume-eligible CLI session. + if (this.isWorktreeResumeReserved(worktreePath)) { + log.log(`[self-healing] deferring idle-sweep for ${worktreePath}: resume-eligible CLI session present`); + continue; + } try { await removeWorktree({ rootDir: this.options.rootDir, @@ -8506,6 +8542,11 @@ export class SelfHealingManager { log.log(`[self-healing] deferring unregistered-orphan reap for ${path}: active session present`); continue; } + // U8: never reclaim a worktree backing a resume-eligible CLI session. + if (this.isWorktreeResumeReserved(path)) { + log.log(`[self-healing] deferring unregistered-orphan reap for ${path}: resume-eligible CLI session present`); + continue; + } try { rmSync(path, { recursive: true, force: true }); log.log(`Cleaned unregistered worktree dir: ${path}`); @@ -8639,6 +8680,11 @@ export class SelfHealingManager { for (const { path: worktreePath } of withMtime) { if (removed >= excess) break; + // U8: never reclaim a worktree backing a resume-eligible CLI session. + if (this.isWorktreeResumeReserved(worktreePath)) { + log.log(`[self-healing] cap-enforcement skipping ${worktreePath}: resume-eligible CLI session present`); + continue; + } try { await removeWorktree({ rootDir: this.options.rootDir, diff --git a/packages/engine/src/stuck-task-detector.ts b/packages/engine/src/stuck-task-detector.ts index f032b10e31..75736a9a96 100644 --- a/packages/engine/src/stuck-task-detector.ts +++ b/packages/engine/src/stuck-task-detector.ts @@ -109,6 +109,16 @@ export interface StuckTaskDetectorOptions { * * Errors in this callback fall through to the normal kill path (treated as `false`). */ onLoopDetected?: (event: StuckTaskEvent) => Promise; + /** + * U8 (CLI Agent Executor): returns true when a task's live CLI agent session + * is `waitingOnInput` — expected idleness (a permission/question prompt). When + * true, stuck/inactivity flagging is SUPPRESSED for that task this cycle: the + * agent is intentionally quiet waiting for a human, not stalled. The U3 stall + * backstop (the CLI session state machine's own watchdog) remains the only + * escalation path while waiting. Narrow seam: a single lookup; absence + * (undefined) preserves the prior behavior. Errors are treated as "not + * waiting" (fail toward the normal stuck path, never silently suppress). */ + isCliSessionWaitingOnInput?: (taskId: string) => boolean; } export class StuckTaskDetector { @@ -118,6 +128,7 @@ export class StuckTaskDetector { private onStuck?: (event: StuckTaskEvent) => void; private beforeRequeue?: (taskId: string, reason: "inactivity" | "loop" | "no-progress-churn", event: StuckTaskEvent) => Promise; private onLoopDetected?: (event: StuckTaskEvent) => Promise; + private isCliSessionWaitingOnInput?: (taskId: string) => boolean; private paused = false; private exhaustedTasks = new Set(); @@ -129,6 +140,23 @@ export class StuckTaskDetector { this.onStuck = options.onStuck; this.beforeRequeue = options.beforeRequeue; this.onLoopDetected = options.onLoopDetected; + this.isCliSessionWaitingOnInput = options.isCliSessionWaitingOnInput; + } + + /** + * U8: whether stuck flagging should be suppressed for a task because its live + * CLI agent session is `waitingOnInput` (expected idleness). Defensive: a + * throw in the injected lookup is treated as "not waiting" so a broken seam + * never silently disables stuck detection. + */ + private isWaitingOnInput(canonicalTaskId: string): boolean { + if (!this.isCliSessionWaitingOnInput) return false; + try { + return this.isCliSessionWaitingOnInput(canonicalTaskId); + } catch (err) { + stuckLog.error(`waitingOnInput lookup threw for ${canonicalTaskId}; treating as not waiting:`, err); + return false; + } } /** @@ -617,6 +645,13 @@ export class StuckTaskDetector { } const reason = this.classifyStuckReason(taskId, timeoutMs); if (reason !== null) { + // U8: suppress flagging while the CLI session is waitingOnInput + // (expected idleness). The U3 stall backstop remains the only escalation + // path while genuinely waiting. + if (this.isWaitingOnInput(entry.canonicalTaskId)) { + stuckLog.log(`Suppressing stuck flag for ${taskId} (canonical=${entry.canonicalTaskId}) — CLI session waitingOnInput`); + continue; + } stuckTasks.push(taskId); } }