From 11e76f6655b1843c6afdc4c036880b79b2ac216c Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 31 Aug 2026 12:10:34 +0200 Subject: [PATCH] fix(server): resume agents cleanly after failed turns Track each foreground run start as pending, started, or failed so a previous turn error cannot reject a newly accepted prompt. Includes a regression for resuming after a failed turn. --- .../src/server/agent/agent-manager.test.ts | 82 +++++++++++++++++++ .../server/src/server/agent/agent-manager.ts | 29 ++++--- .../src/server/agent/agent-run-state.ts | 21 +++-- 3 files changed, 117 insertions(+), 15 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 07cd6ae72a..4293f7210b 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -5543,6 +5543,88 @@ test("waitForAgentRunStart resolves while a foreground run is still only pending expect(manager.getAgent(snapshot.id)?.lifecycle).toBe("idle"); }); +test("waitForAgentRunStart ignores a prior turn error while the next run starts", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-error-resume-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const retryStartEntered = deferred(); + const releaseRetryStart = deferred(); + + class ErrorThenResumeSession extends TestAgentSession { + override async startTurn(): Promise<{ turnId: string }> { + const turnId = `turn-${++this.turnIdCounter}`; + if (this.turnIdCounter === 1) { + void (async () => { + this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + this.pushEvent({ + type: "turn_failed", + provider: this.provider, + turnId, + error: "model at capacity", + }); + })(); + return { turnId }; + } + + retryStartEntered.resolve(); + await releaseRetryStart.promise; + void (async () => { + this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); + })(); + return { turnId }; + } + } + + class ErrorThenResumeClient extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + return new ErrorThenResumeSession(config); + } + } + + const manager = new AgentManager({ + clients: { codex: new ErrorThenResumeClient() }, + registry: storage, + logger, + }); + let agentId: string | null = null; + + try { + const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + agentId = snapshot.id; + + await manager.runAgent(agentId, "fail").catch(() => undefined); + expect(manager.getAgent(agentId)?.lifecycle).toBe("error"); + + const dispatch = await startAgentRun(manager, agentId, "resume", logger); + expect(dispatch.disposition).toBe("turn_started"); + const wait = manager.waitForAgentRunStart(agentId); + let earlyResult: "pending" | "resolved" | "rejected" = "pending"; + void wait.then( + () => { + earlyResult = "resolved"; + return earlyResult; + }, + () => { + earlyResult = "rejected"; + return earlyResult; + }, + ); + await retryStartEntered.promise; + await Promise.resolve(); + + expect(earlyResult).toBe("pending"); + releaseRetryStart.resolve(); + await expect(wait).resolves.toBeUndefined(); + await manager.waitForAgentEvent(agentId); + } finally { + releaseRetryStart.resolve(); + if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("replaceAgentRun does not emit idle or resolve waiters between interrupted and replacement runs", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-replace-run-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 3b560e5d53..de256c7954 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2165,6 +2165,7 @@ export class AgentManager { } agent.pendingReplacement = false; const errorMsg = error instanceof Error ? error.message : "Failed to start turn"; + pendingRun.start = { status: "failed", error: errorMsg }; await this.handleStreamEvent(agent, { type: "turn_failed", provider: agent.provider, @@ -2232,8 +2233,7 @@ export class AgentManager { agent.pendingReplacement = false; } const turnStartedAt = new Date(); - pendingRun.started = true; - pendingRun.turnId = turnId; + pendingRun.start = { status: "started", turnId }; agent.activeForegroundTurnId = turnId; this.openActiveTurn(agent, turnId, turnStartedAt); agent.lifecycle = "running"; @@ -2575,7 +2575,10 @@ export class AgentManager { } const pendingRun = this.runs.getPendingRun(agentId); - if ((snapshot.lifecycle === "running" || pendingRun?.started) && !snapshot.pendingReplacement) { + if ( + (snapshot.lifecycle === "running" || pendingRun?.start.status === "started") && + !snapshot.pendingReplacement + ) { return; } @@ -2640,14 +2643,19 @@ export class AgentManager { const currentPendingRun = this.runs.getPendingRun(agentId); if ( - (current.lifecycle === "running" || currentPendingRun?.started) && + (current.lifecycle === "running" || currentPendingRun?.start.status === "started") && !current.pendingReplacement ) { finishOk(); return true; } - if (current.lifecycle === "error" && !currentPendingRun?.started) { + if (currentPendingRun?.start.status === "failed") { + finishErr(new Error(currentPendingRun.start.error)); + return true; + } + + if (current.lifecycle === "error" && !currentPendingRun) { finishErr(new Error(current.lastError ?? `Agent ${agentId} failed to start`)); return true; } @@ -2734,16 +2742,17 @@ export class AgentManager { return { status: settlement === "completed" ? "settled" : "refused" }; } - if (settlement === "timed_out" && run.turnId) { + const runTurnId = this.runs.getTurnId(agentId); + if (settlement === "timed_out" && runTurnId) { this.logger.warn( - { agentId, turnId: run.turnId, kind: run.kind }, + { agentId, turnId: runTurnId, kind: run.kind }, "cancelAgentRun: acknowledged turn still active after timeout, force-canceling", ); await this.dispatchSessionEvent(agent, { type: "turn_canceled", provider: agent.provider, reason: "interrupted", - turnId: run.turnId, + turnId: runTurnId, }); await run.settledPromise; } else if (settlement === "timed_out" && run.kind === "foreground") { @@ -3023,7 +3032,7 @@ export class AgentManager { let hasStarted = isAgentBusy(initialStatus) || Boolean(snapshot.activeForegroundTurnId) || - Boolean(pendingForegroundRun?.started); + pendingForegroundRun?.start.status === "started"; let terminalStatusOverride: AgentLifecycleStatus | null = null; let finished = false; @@ -3429,7 +3438,7 @@ export class AgentManager { return; } const pendingRun = this.runs.getPendingRun(agentId); - if (pendingRun && !pendingRun.started) { + if (pendingRun?.start.status === "pending") { pendingRun.stagedEvents.push(event); return; } diff --git a/packages/server/src/server/agent/agent-run-state.ts b/packages/server/src/server/agent/agent-run-state.ts index 06176394a0..ad1eb0af34 100644 --- a/packages/server/src/server/agent/agent-run-state.ts +++ b/packages/server/src/server/agent/agent-run-state.ts @@ -14,8 +14,10 @@ export interface PendingForegroundRun { token: string; kind: "foreground"; stagedEvents: AgentStreamEvent[]; - turnId: string | null; - started: boolean; + start: + | { status: "pending" } + | { status: "started"; turnId: string } + | { status: "failed"; error: string }; settled: boolean; settledPromise: Promise; resolveSettled: () => void; @@ -64,6 +66,13 @@ export class AgentRunState { return this.runs.has(agentId); } + getTurnId(agentId: string): string | null { + const run = this.runs.get(agentId); + if (!run) return null; + if (run.kind === "autonomous") return run.turnId; + return run.start.status === "started" ? run.start.turnId : null; + } + trackAutonomousRun(agentId: string, turnId: string | null): TrackedAgentRun { const current = this.runs.get(agentId); if (current) { @@ -85,7 +94,10 @@ export class AgentRunState { if (!run) { return; } - if (run.kind === "foreground" && (run.turnId === null || run.turnId !== turnId)) { + if ( + run.kind === "foreground" && + (run.start.status !== "started" || run.start.turnId !== turnId) + ) { return; } if ( @@ -268,8 +280,7 @@ function createPendingForegroundRun(): PendingForegroundRun { return { ...createTrackedRunState(), kind: "foreground", - turnId: null, - started: false, + start: { status: "pending" }, stagedEvents: [], }; }