diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 7e3d0913c3..d9f9710599 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -35,7 +35,16 @@ be in flight. ### Cancellation -Cancellation changes lifecycle state only after the provider acknowledges the interrupt or emits a terminal turn event. If the interrupt is rejected or times out, the agent remains `running` with its active foreground turn intact. Follow-up actions such as replacement, reload, rewind, and Stop must report that failure instead of accepting work they cannot perform. Synthesizing a local cancellation without provider acknowledgment creates a split-brain session: Paseo accepts a new prompt while the provider still owns the previous foreground turn. +Provider interruption is idempotent at the `AgentSession` boundary. It resolves when the prior +foreground turn can no longer run, including when the provider reports that it is already idle. It +rejects only when the provider may still own the turn. Provider adapters translate native errors +into that contract; lifecycle callers do not interpret provider-specific errors. + +After an acknowledged interrupt, the manager settles the captured run even when no terminal event +arrives or the run was still waiting for its provider turn id. The captured run token prevents an +older cancellation from settling a newer turn. If interruption is rejected or times out, the agent +keeps its active foreground turn and replacement, reload, rewind, and Stop report the failure. +Accepting new work after an ambiguous interruption would create a split-brain session. ## Relationships diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 99088c0843..b61a7bcd36 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1060,6 +1060,7 @@ test("isolated rewind falls back from steering to the normal replacement path", activeTurnBehavior: "steer", runOptions: { clientMessageId: "rewind-client" }, }); + await manager.waitForAgentRunStart(agentId); expect(session.interruptCount).toBe(2); expect(session.startPrompts).toContain("/rewind submitted-message-id"); expect(manager.getTimeline(agentId)).toContainEqual( @@ -7144,6 +7145,63 @@ test("streamAgent clears pending run when startTurn fails before a turn id exist ); }); +test("acknowledged cancellation settles a pending run before it has a turn id", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cancel-pending-start-")); + const startEntered = deferred(); + let rejectStart!: (error: Error) => void; + const stalledStart = new Promise<{ turnId: string }>((_resolve, reject) => { + rejectStart = reject; + }); + + class PendingStartSession extends TestAgentSession { + override async startTurn(): Promise<{ turnId: string }> { + startEntered.resolve(); + return await stalledStart; + } + + override async interrupt(): Promise {} + } + + class PendingStartClient extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + return new PendingStartSession(config); + } + } + + const manager = new AgentManager({ + clients: { codex: new PendingStartClient() }, + registry: new AgentStorage(join(workdir, "agents"), logger), + logger, + idFactory: () => "00000000-0000-4000-8000-000000000132", + }); + + try { + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + const firstRun = manager.runAgent(agent.id, "start a turn"); + void firstRun.catch(() => undefined); + await startEntered.promise; + + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + expect(manager.hasInFlightRun(agent.id)).toBe(true); + + await expect(manager.cancelAgentRun(agent.id)).resolves.toEqual({ status: "settled" }); + expect(manager.hasInFlightRun(agent.id)).toBe(false); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + + rejectStart(new Error("released canceled start")); + await expect(firstRun).rejects.toThrow("released canceled start"); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + } finally { + rejectStart(new Error("test cleanup")); + if (manager.getAgent("00000000-0000-4000-8000-000000000132")) { + await manager.closeAgent("00000000-0000-4000-8000-000000000132"); + } + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("archiveAgent persists archivedAt and updatedAt before emitting closed state", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archive-")); 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 2634324505..d8820e83ab 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -66,7 +66,11 @@ import { AgentStreamCoalescer, } from "./agent-stream-coalescer.js"; import { limitAgentTimelineItemContent } from "./agent-timeline-content.js"; -import { AgentRunState, type ForegroundTurnWaiter } from "./agent-run-state.js"; +import { + AgentRunState, + type ForegroundTurnWaiter, + type PendingForegroundRun, +} from "./agent-run-state.js"; import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js"; import { isSystemInjectedEnvelope } from "./agent-prompt.js"; import { stripInternalPaseoMcpServer, withRuntimePaseoMcpServer } from "./runtime-mcp-config.js"; @@ -2139,6 +2143,37 @@ export class AgentManager { }); } + private async startPendingForegroundTurn(params: { + agent: ActiveManagedAgent; + agentId: string; + pendingRun: PendingForegroundRun; + prompt: AgentPromptInput; + options?: AgentRunOptions; + }): Promise { + const { agent, agentId, pendingRun, prompt, options } = params; + try { + const result = await agent.session.startTurn(prompt, options); + if (pendingRun.settled) { + throw new Error(`Agent ${agentId} run was canceled before its turn started`); + } + return result.turnId; + } catch (error) { + if (pendingRun.settled) { + throw error; + } + agent.pendingReplacement = false; + const errorMsg = error instanceof Error ? error.message : "Failed to start turn"; + await this.handleStreamEvent(agent, { + type: "turn_failed", + provider: agent.provider, + error: errorMsg, + }); + this.finalizeForegroundTurn(agent); + this.runs.settleForegroundRun(agentId, pendingRun.token); + throw error; + } + } + streamAgent( agentId: string, prompt: AgentPromptInput, @@ -2183,21 +2218,13 @@ export class AgentManager { const streamForwarder = async function* streamForwarder(this: AgentManager) { let turnId: string; let turnStream: ReturnType | null = null; - try { - const result = await agent.session.startTurn(prompt, options); - turnId = result.turnId; - } catch (error) { - agent.pendingReplacement = false; - const errorMsg = error instanceof Error ? error.message : "Failed to start turn"; - await this.handleStreamEvent(agent, { - type: "turn_failed", - provider: agent.provider, - error: errorMsg, - }); - this.finalizeForegroundTurn(agent); - this.runs.settleForegroundRun(agentId, pendingRun.token); - throw error; - } + turnId = await this.startPendingForegroundTurn({ + agent, + agentId, + pendingRun, + prompt, + options, + }); if (isReplacement) { agent.pendingReplacement = false; @@ -2717,6 +2744,17 @@ export class AgentManager { turnId: run.turnId, }); await run.settledPromise; + } else if (settlement === "timed_out" && run.kind === "foreground") { + this.logger.warn( + { agentId, kind: run.kind }, + "cancelAgentRun: acknowledged pending turn still active after timeout, clearing it", + ); + this.runs.settleForegroundRun(agentId, run.token); + if (!agent.pendingReplacement) { + agent.lifecycle = "idle"; + this.touchUpdatedAt(agent); + this.emitState(agent); + } } else if (settlement === "timed_out" && run.kind === "autonomous") { this.logger.warn( { agentId, kind: run.kind }, diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index b94043c960..3a01764209 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -653,6 +653,11 @@ export interface AgentSession { response: AgentPermissionResponse, ): Promise; describePersistence(): AgentPersistenceHandle | null; + /** + * Resolve once every foreground turn that predates this call can no longer run or become active. + * Calling while already idle is a successful no-op. Reject only when foreground ownership is + * still uncertain. + */ interrupt(): Promise; /** Release live runtime resources without archiving or deleting the durable native session. */ close(): Promise; diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts index aa81e035b5..63a45be061 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts @@ -111,6 +111,62 @@ test("unexpected Codex app-server exit fails the active run and agent", async () } }); +test("replacement self-heals when Codex reports that the tracked turn is already idle", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-already-idle-replace-")); + const appServer = createFakeCodexAppServer({ + "turn/interrupt": () => ({ + __jsonRpcError: { code: -32600, message: "no active turn to interrupt" }, + }), + }); + const manager = new AgentManager({ + clients: { codex: new ProcessExitCodexClient([appServer]) }, + logger, + }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + const firstRun = manager.runAgent(agent.id, "keep working"); + const firstTurnStart = await appServer.waitForTurnStart(); + appServer.startsTurn({ + threadId: String(firstTurnStart.threadId), + turnId: "native-turn-already-idle", + }); + await manager.waitForAgentRunStart(agent.id); + + const replacementStream = await manager.replaceAgentRun(agent.id, "replacement prompt"); + const replacementEvents = (async () => { + const events = []; + for await (const event of replacementStream) { + events.push(event); + } + return events; + })(); + + await expect + .poll(() => appServer.requests().filter((message) => message.method === "turn/start").length) + .toBe(2); + appServer.startsTurn({ threadId: "thread-1", turnId: "native-turn-replacement" }); + appServer.completeTurn({ threadId: "thread-1" }); + + await expect(firstRun).resolves.toMatchObject({ canceled: true }); + await expect(replacementEvents).resolves.toContainEqual( + expect.objectContaining({ type: "turn_completed" }), + ); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("intentional agent close stays clean while a Codex turn is active", async () => { const workdir = mkdtempSync(join(tmpdir(), "codex-process-close-")); const appServer = createFakeCodexAppServer(); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 2f8c75413d..d0a9dd3dde 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -3354,6 +3354,34 @@ describe("Codex app-server provider", () => { } }); + test("treats Codex already having no active turn as an acknowledged interrupt", async () => { + const appServer = createFakeCodexAppServer({ + "turn/interrupt": () => ({ + __jsonRpcError: { code: -32600, message: "no active turn to interrupt" }, + }), + }); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Wait for the child."); + await appServer.waitForTurnStart(); + appServer.startsTurn({ threadId: "thread-1", turnId: "turn-already-idle" }); + + await expect(session.interrupt()).resolves.toBeUndefined(); + + appServer.completeTurn(); + await resultPromise; + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + test("waits for Codex to identify an accepted turn before interrupting it", async () => { const interruptedTurns: unknown[] = []; const appServer = createFakeCodexAppServer({ @@ -3386,7 +3414,7 @@ describe("Codex app-server provider", () => { } }); - test("does not interrupt after the accepted turn terminates before identification", async () => { + test("acknowledges interruption when the accepted turn terminates before identification", async () => { const interruptedTurns: unknown[] = []; const appServer = createFakeCodexAppServer({ "turn/interrupt": async (params) => { @@ -3407,9 +3435,7 @@ describe("Codex app-server provider", () => { const interruptPromise = session.interrupt(); appServer.completeTurn(); - await expect(interruptPromise).rejects.toThrow( - "Cannot interrupt Codex before turn/started identifies the active turn", - ); + await expect(interruptPromise).resolves.toBeUndefined(); await resultPromise; expect(interruptedTurns).toEqual([]); appServer.assertNoErrors(); @@ -3418,7 +3444,7 @@ describe("Codex app-server provider", () => { } }); - test("rejects an interrupt before Codex initializes the thread", async () => { + test("acknowledges interruption before Codex initializes a thread", async () => { const appServer = createFakeCodexAppServer(); const session = new CodexAppServerAgentSession( createConfig({ cwd: "/workspace/project" }), @@ -3427,13 +3453,53 @@ describe("Codex app-server provider", () => { async () => appServer.child, ); - await expect(session.interrupt()).rejects.toThrow( - "Cannot interrupt Codex before the active thread is initialized", - ); + await expect(session.interrupt()).resolves.toBeUndefined(); await session.close(); }); + test("cancels a cold start before it can issue a native turn", async () => { + const threadStart = deferred<{ thread: { id: string } }>(); + const appServer = createFakeCodexAppServer({ + "thread/start": () => threadStart.promise, + }); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + const resultPromise = session.run("Start after the delayed thread."); + + try { + await appServer.waitForRequest("thread/start"); + let interruptSettled = false; + const interruptPromise = session.interrupt().then(() => { + interruptSettled = true; + return undefined; + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(interruptSettled).toBe(false); + + threadStart.resolve({ thread: { id: "thread-1" } }); + await expect(interruptPromise).resolves.toBeUndefined(); + await expect(resultPromise).rejects.toThrow("interrupted before reaching Codex"); + expect(appServer.requests()).not.toContainEqual( + expect.objectContaining({ method: "turn/start" }), + ); + appServer.assertNoErrors(); + } finally { + threadStart.resolve({ thread: { id: "thread-1" } }); + const nativeStart = await appServer.waitForTurnStart().catch(() => null); + if (nativeStart) { + appServer.startsTurn({ threadId: "thread-1", turnId: "cleanup-turn" }); + appServer.completeTurn(); + } + await resultPromise.catch(() => undefined); + await session.close(); + } + }); + test("interrupts an autonomous Codex turn identified by live notifications", async () => { const session = createSession(); const requests: Array<{ method: string; params: unknown }> = []; diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 859318ffe6..f7e06cc005 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -857,6 +857,14 @@ function isDefinitiveCodexSteerRejection(error: unknown): boolean { ); } +function isCodexAlreadyIdleInterrupt(error: unknown): boolean { + return ( + error instanceof CodexAppServerRpcError && + error.code === -32600 && + error.message === "no active turn to interrupt" + ); +} + // Codex app-server API response types interface CodexReasoningEffortEntry { reasoningEffort?: string; @@ -3227,6 +3235,11 @@ export class CodexAppServerAgentSession implements AgentSession { promise: Promise; resolve: (turnId: string | null) => void; } | null = null; + private pendingForegroundStart: { + promise: Promise; + resolve: () => void; + cancelRequested: boolean; + } | null = null; private client: CodexAppServerClient | null = null; private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); private nextTurnOrdinal = 0; @@ -4081,10 +4094,20 @@ export class CodexAppServerAgentSession implements AgentSession { prompt: AgentPromptInput, options?: AgentRunOptions, ): Promise<{ turnId: string }> { - if (this.activeForegroundTurnId) { + if (this.activeForegroundTurnId || this.pendingForegroundStart) { throw new Error("A foreground turn is already active"); } + let resolveStart!: () => void; + const pendingStart = { + promise: new Promise((resolve) => { + resolveStart = resolve; + }), + resolve: () => resolveStart(), + cancelRequested: false, + }; + this.pendingForegroundStart = pendingStart; + this.dismissPendingPlanApprovals("Dismissed by a new prompt"); try { @@ -4129,6 +4152,9 @@ export class CodexAppServerAgentSession implements AgentSession { hasDeveloperInstructions: turnStart.hasDeveloperInstructions, hasCodexConfig: turnStart.hasCodexConfig, }); + if (pendingStart.cancelRequested) { + throw new Error("Codex turn start was interrupted before reaching Codex"); + } await this.client.request("turn/start", turnStart.params, TURN_START_TIMEOUT_MS); return { turnId }; } catch (error) { @@ -4137,6 +4163,11 @@ export class CodexAppServerAgentSession implements AgentSession { this.activeForegroundTurnId = null; this.activeClientMessageId = null; throw error; + } finally { + if (this.pendingForegroundStart === pendingStart) { + this.pendingForegroundStart = null; + } + pendingStart.resolve(); } } @@ -4617,7 +4648,19 @@ export class CodexAppServerAgentSession implements AgentSession { } async interrupt(): Promise { + const pendingStart = this.pendingForegroundStart; + if (pendingStart) { + pendingStart.cancelRequested = true; + await pendingStart.promise; + } if (!this.client || !this.currentThreadId) { + if ( + !this.activeForegroundTurnId && + !this.currentTurnId && + !this.pendingForegroundTurnIdentification + ) { + return; + } throw new Error("Cannot interrupt Codex before the active thread is initialized"); } let turnId = this.currentTurnId; @@ -4632,17 +4675,31 @@ export class CodexAppServerAgentSession implements AgentSession { ) { turnId = await pendingIdentification.promise; } + if (!turnId && !this.activeForegroundTurnId && !this.currentTurnId) { + return; + } if (!turnId || (foregroundTurnId && this.activeForegroundTurnId !== foregroundTurnId)) { throw new Error("Cannot interrupt Codex before turn/started identifies the active turn"); } - await this.client.request( - "turn/interrupt", - { - threadId: this.currentThreadId, - turnId, - }, - INTERRUPT_TIMEOUT_MS, - ); + try { + await this.client.request( + "turn/interrupt", + { + threadId: this.currentThreadId, + turnId, + }, + INTERRUPT_TIMEOUT_MS, + ); + } catch (error) { + if (!isCodexAlreadyIdleInterrupt(error)) { + throw error; + } + this.activeForegroundTurnId = null; + this.activeClientMessageId = null; + this.currentTurnId = null; + this.pendingForegroundTurnIdentification?.resolve(null); + this.pendingForegroundTurnIdentification = null; + } } async close(): Promise {