diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 2358b7e4d..01e392ffa 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -12,6 +12,10 @@ initializing → idle → running → idle (or error → closed) Each agent in `AgentManager` carries a `lastStatus` of `initializing`, `idle`, `running`, `error`, or `closed`. State transitions persist to disk and stream to subscribed clients via WebSocket. +### 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. + ## Relationships Agents can launch other agents via the agent-scoped `create_agent` MCP tool. Agent-scoped creation is always asynchronous. `relationship` and `workspace` are separate decisions: diff --git a/packages/app/src/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index f6af2e78d..ad29711a5 100644 --- a/packages/app/src/composer/actions.test.ts +++ b/packages/app/src/composer/actions.test.ts @@ -227,6 +227,7 @@ describe("cancelComposerAgent", () => { isAgentRunning: boolean; isCancellingAgent: boolean; isConnected: boolean; + onCancelFailed: (error: unknown) => void; } { const canceledIds: string[] = []; return { @@ -240,6 +241,7 @@ describe("cancelComposerAgent", () => { isAgentRunning: true, isCancellingAgent: false, isConnected: true, + onCancelFailed: () => undefined, }; } @@ -250,6 +252,24 @@ describe("cancelComposerAgent", () => { expect(input.client.canceledIds).toEqual(["agent"]); }); + it("reports a rejected cancel so the composer can leave its canceling state", async () => { + const cancellationError = new Error("Provider rejected the interrupt"); + const failures: unknown[] = []; + const input = baseInput(); + input.client.cancelAgent = async () => { + throw cancellationError; + }; + + const result = cancelComposerAgent({ + ...input, + onCancelFailed: (error: unknown) => failures.push(error), + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(result).toBe(true); + expect(failures).toEqual([cancellationError]); + }); + it("does nothing when the agent is not running", () => { const input = baseInput(); const result = cancelComposerAgent({ ...input, isAgentRunning: false }); diff --git a/packages/app/src/composer/actions.ts b/packages/app/src/composer/actions.ts index 816736ac0..f2a90c8b7 100644 --- a/packages/app/src/composer/actions.ts +++ b/packages/app/src/composer/actions.ts @@ -142,12 +142,18 @@ export interface CancelComposerAgentInput { isAgentRunning: boolean; isCancellingAgent: boolean; isConnected: boolean; + onCancelFailed: (error: unknown) => void; } export function cancelComposerAgent(input: CancelComposerAgentInput): boolean { if (!input.isAgentRunning || input.isCancellingAgent) return false; if (!input.isConnected || !input.client) return false; - void input.client.cancelAgent(input.agentId); + try { + void Promise.resolve(input.client.cancelAgent(input.agentId)).catch(input.onCancelFailed); + } catch (error) { + input.onCancelFailed(error); + return false; + } return true; } diff --git a/packages/app/src/composer/index.tsx b/packages/app/src/composer/index.tsx index 396b279ad..3cd4760f7 100644 --- a/packages/app/src/composer/index.tsx +++ b/packages/app/src/composer/index.tsx @@ -416,7 +416,7 @@ function renderComposerAttachmentPill(args: RenderComposerAttachmentPillArgs): R ); } -function resolveVoiceStartErrorMessage(error: unknown): string | null { +function resolveErrorMessage(error: unknown): string | null { if (error instanceof Error) return error.message; if (typeof error === "string") return error; return null; @@ -438,7 +438,7 @@ function attemptStartRealtimeVoice(args: AttemptStartRealtimeVoiceArgs): void { if (voice.isVoiceModeForAgent(serverId, agentId)) return; void voice.startVoice(serverId, agentId).catch((error) => { console.error("[Composer] Failed to start voice mode", error); - const message = resolveVoiceStartErrorMessage(error); + const message = resolveErrorMessage(error); if (message && message.trim().length > 0) { toastErrorRef.current(message); } @@ -1466,6 +1466,13 @@ export function Composer({ isAgentRunning, isCancellingAgent, isConnected, + onCancelFailed: (error) => { + setIsCancellingAgent(false); + const message = resolveErrorMessage(error); + if (message && message.trim().length > 0) { + toastErrorRef.current(message); + } + }, }); if (!didCancel) return; setIsCancellingAgent(true); diff --git a/packages/app/src/composer/input/input.tsx b/packages/app/src/composer/input/input.tsx index 4c68a2c40..b35a2811d 100644 --- a/packages/app/src/composer/input/input.tsx +++ b/packages/app/src/composer/input/input.tsx @@ -66,7 +66,12 @@ import { resolveVoiceAccessibilityLabel, resolveVoiceTooltipText, } from "./labels"; -import { computeCanStartDictation, runAlternateSendAction, runDefaultSendAction } from "./state"; +import { + computeCanStartDictation, + runAlternateSendAction, + runDefaultSendAction, + stopRealtimeVoice, +} from "./state"; const DEFAULT_SEND_KEYS: ShortcutKey[][] = [["Enter"]]; @@ -929,31 +934,6 @@ async function startDictationIfAvailableImpl(ctx: StartDictationContext): Promis await ctx.startDictation(); } -interface StopRealtimeVoiceContext { - voice: { stopVoice: () => Promise } | null | undefined; - isRealtimeVoiceForCurrentAgent: boolean; - isAgentRunning: boolean; - client: { cancelAgent: (agentId: string) => Promise } | null; - voiceAgentId: string | undefined; -} - -async function stopRealtimeVoiceImpl(ctx: StopRealtimeVoiceContext): Promise { - if (!ctx.voice || !ctx.isRealtimeVoiceForCurrentAgent) return; - - const tasks: Promise[] = []; - if (ctx.isAgentRunning && ctx.client && ctx.voiceAgentId) { - tasks.push(ctx.client.cancelAgent(ctx.voiceAgentId)); - } - tasks.push(ctx.voice.stopVoice()); - - const results = await Promise.allSettled(tasks); - results.forEach((result) => { - if (result.status === "rejected") { - console.error("[MessageInput] Failed to stop realtime voice", result.reason); - } - }); -} - interface VoicePressContext { isRealtimeVoiceForCurrentAgent: boolean; voice: { toggleMute: () => void } | null | undefined; @@ -1502,17 +1482,23 @@ export const MessageInput = forwardRef( discardFailedDictation(); }, [discardFailedDictation]); - const handleStopRealtimeVoice = useCallback( - () => - stopRealtimeVoiceImpl({ + const handleStopRealtimeVoice = useCallback(async () => { + try { + await stopRealtimeVoice({ voice, isRealtimeVoiceForCurrentAgent, isAgentRunning, client, voiceAgentId, - }), - [client, isAgentRunning, isRealtimeVoiceForCurrentAgent, voice, voiceAgentId], - ); + }); + } catch (error) { + console.error("[MessageInput] Failed to stop realtime voice", error); + const message = extractErrorMessage(error); + if (message && message.trim().length > 0) { + toast.error(message); + } + } + }, [client, isAgentRunning, isRealtimeVoiceForCurrentAgent, toast, voice, voiceAgentId]); const handleToggleRealtimeVoiceShortcut = useCallback(() => { toggleRealtimeVoiceImpl({ diff --git a/packages/app/src/composer/input/state.test.ts b/packages/app/src/composer/input/state.test.ts index ebea86bbb..8a11806ed 100644 --- a/packages/app/src/composer/input/state.test.ts +++ b/packages/app/src/composer/input/state.test.ts @@ -1,5 +1,10 @@ -import { describe, expect, it } from "vitest"; -import { computeCanStartDictation, runAlternateSendAction, runDefaultSendAction } from "./state"; +import { describe, expect, it, vi } from "vitest"; +import { + computeCanStartDictation, + runAlternateSendAction, + runDefaultSendAction, + stopRealtimeVoice, +} from "./state"; const connected = { isConnected: true } as never; const disconnected = { isConnected: false } as never; @@ -149,3 +154,45 @@ describe("composer send behavior", () => { expect(alternateAction.calls).toEqual(["send"]); }); }); + +describe("stopRealtimeVoice", () => { + it("keeps voice mode active when the running agent refuses cancellation", async () => { + const cancellationError = new Error("active run cancellation was not acknowledged"); + const cancelAgent = vi.fn().mockRejectedValue(cancellationError); + const stopVoice = vi.fn().mockResolvedValue(undefined); + + await expect( + stopRealtimeVoice({ + voice: { stopVoice }, + isRealtimeVoiceForCurrentAgent: true, + isAgentRunning: true, + client: { cancelAgent }, + voiceAgentId: "agent-1", + }), + ).rejects.toBe(cancellationError); + + expect(stopVoice).not.toHaveBeenCalled(); + }); + + it("stops voice mode after the running agent acknowledges cancellation", async () => { + const calls: string[] = []; + + await stopRealtimeVoice({ + voice: { + stopVoice: async () => { + calls.push("stop voice"); + }, + }, + isRealtimeVoiceForCurrentAgent: true, + isAgentRunning: true, + client: { + cancelAgent: async () => { + calls.push("cancel agent"); + }, + }, + voiceAgentId: "agent-1", + }); + + expect(calls).toEqual(["cancel agent", "stop voice"]); + }); +}); diff --git a/packages/app/src/composer/input/state.ts b/packages/app/src/composer/input/state.ts index 9eaa554fc..e2f714623 100644 --- a/packages/app/src/composer/input/state.ts +++ b/packages/app/src/composer/input/state.ts @@ -3,6 +3,14 @@ import type { MessagePayload } from "@/composer/types"; export type SendBehavior = "interrupt" | "queue"; +interface StopRealtimeVoiceContext { + voice: { stopVoice: () => Promise } | null | undefined; + isRealtimeVoiceForCurrentAgent: boolean; + isAgentRunning: boolean; + client: { cancelAgent: (agentId: string) => Promise } | null; + voiceAgentId: string | undefined; +} + interface SendActionContext { defaultSendBehavior: SendBehavior; isAgentRunning: boolean; @@ -41,3 +49,16 @@ export function runAlternateSendAction(ctx: SendActionContext): void { ctx.handleQueueMessage(); } } + +export async function stopRealtimeVoice(ctx: StopRealtimeVoiceContext): Promise { + if (!ctx.voice || !ctx.isRealtimeVoiceForCurrentAgent) return; + + if (ctx.isAgentRunning) { + if (!ctx.client || !ctx.voiceAgentId) { + throw new Error("Cannot stop the running voice agent while the host is unavailable"); + } + await ctx.client.cancelAgent(ctx.voiceAgentId); + } + + await ctx.voice.stopVoice(); +} diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index e5fb01e84..7d2ab8074 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -476,6 +476,15 @@ function applyToolErrorToMessages( ); } +function notifyVoiceAbortFailure( + data: Extract["payload"], + notifyError: (message: string) => void, +): void { + if (data.type === "error" && data.metadata?.voiceAbortFailed === true) { + notifyError(data.content); + } +} + interface SessionProviderSharedProps { children: ReactNode; serverId: string; @@ -1570,6 +1579,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setMessages(serverId, applyToolError); } + notifyVoiceAbortFailure(data, toast.error); + let activityType: "system" | "info" | "success" | "error" = "info"; if (data.type === "error") activityType = "error"; @@ -1805,6 +1816,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider applyWorkspaceSetupProgress, applyTimelineResponse, updateSessionServerInfo, + toast, voiceRuntime, voiceAudioEngine, ]); diff --git a/packages/cli/src/commands/agent/delete.test.ts b/packages/cli/src/commands/agent/delete.test.ts new file mode 100644 index 000000000..56aa83dc6 --- /dev/null +++ b/packages/cli/src/commands/agent/delete.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it, vi } from "vitest"; + +import { runDeleteCommand } from "./delete.js"; + +const agent = { + id: "11111111-1111-4111-8111-111111111111", + status: "running", + archivedAt: null, + cwd: "/tmp/project", +}; +const cancelAgent = vi.fn(async () => { + throw new Error("active run cancellation was not acknowledged"); +}); +const deleteAgent = vi.fn(async () => undefined); +const close = vi.fn(async () => undefined); + +vi.mock("../../utils/client.js", () => ({ + connectToDaemon: vi.fn(async () => ({ + fetchAgents: vi.fn(async () => ({ entries: [{ agent }] })), + fetchAgent: vi.fn(async () => ({ agent })), + cancelAgent, + deleteAgent, + close, + })), + getDaemonHost: vi.fn(() => "ws://127.0.0.1:6767"), +})); + +describe("runDeleteCommand", () => { + it("force-deletes a running agent when graceful cancellation is refused", async () => { + const result = await runDeleteCommand(agent.id, {}, {} as never); + + expect(cancelAgent).toHaveBeenCalledWith(agent.id); + expect(deleteAgent).toHaveBeenCalledWith(agent.id); + expect(result.data).toEqual({ + deletedCount: 1, + agentIds: [agent.id], + }); + }); +}); diff --git a/packages/cli/src/commands/agent/delete.ts b/packages/cli/src/commands/agent/delete.ts index dafb26e4c..c8502c9a1 100644 --- a/packages/cli/src/commands/agent/delete.ts +++ b/packages/cli/src/commands/agent/delete.ts @@ -92,7 +92,7 @@ export async function runDeleteCommand( agents.map(async (agent) => { try { if (agent.status === "running") { - await client.cancelAgent(agent.id); + await client.cancelAgent(agent.id).catch(() => {}); } await client.deleteAgent(agent.id); return { ok: true as const, id: agent.id }; diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 1f827a0b3..9f5f3d7c3 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -2664,7 +2664,7 @@ export class DaemonClient { agentId, requestId, }); - await this.sendRequest({ + const payload = await this.sendRequest({ requestId, message, options: { skipQueue: true }, @@ -2678,6 +2678,9 @@ export class DaemonClient { return msg.payload; }, }); + if (payload.error) { + throw new Error(payload.error); + } } async setAgentMode(agentId: string, modeId: string): Promise { diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index ec05be176..1718e1117 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3139,6 +3139,7 @@ export const CancelAgentResponseMessageSchema = z.object({ requestId: z.string(), agentId: z.string(), agent: AgentSnapshotPayloadSchema.nullable(), + error: z.string().nullable().optional(), }), }); diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 5cbff4b03..759f9b2f4 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -54,6 +54,28 @@ function deferred(): Deferred { return { promise, resolve, reject }; } +function waitForAgentLifecycle( + manager: AgentManager, + agentId: string, + lifecycle: ManagedAgent["lifecycle"], +): Promise { + return new Promise((resolve) => { + const unsubscribe = manager.subscribe( + (event) => { + if ( + event.type === "agent_state" && + event.agent.id === agentId && + event.agent.lifecycle === lifecycle + ) { + unsubscribe(); + resolve(); + } + }, + { agentId, replayState: false }, + ); + }); +} + const TEST_CAPABILITIES = { supportsStreaming: false, supportsSessionPersistence: false, @@ -402,6 +424,83 @@ class TestAgentSession implements AgentSession { async close(): Promise {} } +class ControlledInterruptSession extends TestAgentSession { + interruptCalled = false; + + constructor( + config: AgentSessionConfig, + readonly turnId: string, + private readonly interruptBehavior: (session: ControlledInterruptSession) => Promise, + ) { + super(config); + } + + override async startTurn(): Promise<{ turnId: string }> { + setTimeout(() => { + this.pushEvent({ type: "turn_started", provider: this.provider, turnId: this.turnId }); + }, 0); + return { turnId: this.turnId }; + } + + override async interrupt(): Promise { + this.interruptCalled = true; + await this.interruptBehavior(this); + } +} + +interface ControlledInterruptFixture { + agentId: string; + manager: AgentManager; + session: ControlledInterruptSession; + startForegroundRun(): Promise; + cleanup(): void; +} + +async function createControlledInterruptFixture(options: { + name: string; + agentId: string; + turnId: string; + interrupt: (session: ControlledInterruptSession) => Promise; +}): Promise { + const workdir = mkdtempSync(join(tmpdir(), `agent-manager-${options.name}-`)); + const session = new ControlledInterruptSession( + { provider: "codex", cwd: workdir }, + options.turnId, + options.interrupt, + ); + const client = new (class extends TestAgentClient { + override async createSession(): Promise { + return session; + } + })(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: new AgentStorage(join(workdir, "agents"), logger), + logger, + rescueTimeouts: { interruptSessionMs: 10 }, + idFactory: () => options.agentId, + }); + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + + return { + agentId: agent.id, + manager, + session, + async startForegroundRun() { + const run = manager.streamAgent(agent.id, "exercise cancellation"); + void (async () => { + for await (const _event of run) { + // Keep the foreground stream subscribed until the controlled turn settles. + } + })(); + await manager.waitForAgentRunStart(agent.id); + }, + cleanup: () => rmSync(workdir, { recursive: true, force: true }), + }; +} + class HeldRuntimeInfoSession extends TestAgentSession { private readonly runtimeInfoRequested = deferred(); private readonly runtimeInfoAllowed = deferred(); @@ -1357,77 +1456,124 @@ test("reloadAgentSession completes when the previous session close hangs", async } }); -test("cancelAgentRun completes when provider interrupt hangs", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-interrupt-timeout-")); - const storagePath = join(workdir, "agents"); - const storage = new AgentStorage(storagePath, logger); - - class HangingInterruptSession extends TestAgentSession { - interruptCalled = false; - - override async interrupt(): Promise { - this.interruptCalled = true; - await new Promise(() => {}); - } - } - - class HangingInterruptClient extends TestAgentClient { - readonly session = new HangingInterruptSession({ - provider: "codex", - cwd: workdir, - }); - - override async createSession(): Promise { - return this.session; - } - } - - const client = new HangingInterruptClient(); - const manager = new AgentManager({ - clients: { - codex: client, - }, - registry: storage, - logger, - rescueTimeouts: { interruptSessionMs: 10 }, - idFactory: () => "00000000-0000-4000-8000-000000000303", +test("cancelAgentRun preserves running state when the provider interrupt hangs", async () => { + const fixture = await createControlledInterruptFixture({ + name: "interrupt-timeout", + agentId: "00000000-0000-4000-8000-000000000303", + turnId: "hanging-interrupt-turn", + interrupt: async () => await new Promise(() => {}), }); try { - const snapshot = await manager.createAgent( - { - provider: "codex", - cwd: workdir, - }, - undefined, - { workspaceId: undefined }, - ); + const running = waitForAgentLifecycle(fixture.manager, fixture.agentId, "running"); + fixture.session.pushEvent({ + type: "turn_started", + provider: "codex", + turnId: "hanging-interrupt-turn", + }); + await running; - await new Promise((resolve) => { - const unsubscribe = manager.subscribe( - (event) => { - if ( - event.type === "agent_state" && - event.agent.id === snapshot.id && - event.agent.lifecycle === "running" - ) { - unsubscribe(); - resolve(); - } - }, - { agentId: snapshot.id, replayState: false }, - ); - client.session.pushEvent({ - type: "turn_started", - provider: "codex", - turnId: "hanging-interrupt-turn", - }); + await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({ + status: "refused", + }); + expect(fixture.session.interruptCalled).toBe(true); + expect(fixture.manager.getAgent(fixture.agentId)?.lifecycle).toBe("running"); + } finally { + fixture.cleanup(); + } +}); + +test("cancelAgentRun preserves the active turn when the provider rejects the interrupt", async () => { + const fixture = await createControlledInterruptFixture({ + name: "interrupt-rejected", + agentId: "00000000-0000-4000-8000-000000000304", + turnId: "provider-still-active-turn", + interrupt: async () => { + throw new Error("A foreground turn is already active"); + }, + }); + + try { + await fixture.startForegroundRun(); + + await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({ + status: "refused", + }); + expect(fixture.manager.getAgent(fixture.agentId)).toMatchObject({ + lifecycle: "running", + activeForegroundTurnId: "provider-still-active-turn", }); - await expect(manager.cancelAgentRun(snapshot.id)).resolves.toBe(true); - expect(client.session.interruptCalled).toBe(true); + fixture.session.pushEvent({ + type: "turn_completed", + provider: "codex", + turnId: "provider-still-active-turn", + }); } finally { - rmSync(workdir, { recursive: true, force: true }); + fixture.cleanup(); + } +}); + +test("cancelAgentRun succeeds when the foreground turn finishes before the provider rejects the interrupt", async () => { + let fixture!: ControlledInterruptFixture; + fixture = await createControlledInterruptFixture({ + name: "interrupt-after-completion", + agentId: "00000000-0000-4000-8000-000000000305", + turnId: "naturally-completed-turn", + interrupt: async (session) => { + const settled = waitForAgentLifecycle(fixture.manager, fixture.agentId, "idle"); + session.pushEvent({ + type: "turn_completed", + provider: session.provider, + turnId: "naturally-completed-turn", + }); + await settled; + throw new Error("turn already completed"); + }, + }); + + try { + await fixture.startForegroundRun(); + + await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({ + status: "settled", + }); + expect(fixture.manager.getAgent(fixture.agentId)).toMatchObject({ + lifecycle: "idle", + activeForegroundTurnId: null, + }); + } finally { + fixture.cleanup(); + } +}); + +test("cancelAgentRun succeeds when the provider queues completion before rejecting the interrupt", async () => { + const fixture = await createControlledInterruptFixture({ + name: "interrupt-queued-completion", + agentId: "00000000-0000-4000-8000-000000000306", + turnId: "queued-completion-turn", + interrupt: async (session) => { + session.pushEvent({ + type: "turn_completed", + provider: session.provider, + turnId: "queued-completion-turn", + }); + throw new Error("turn already completed"); + }, + }); + + try { + await fixture.startForegroundRun(); + + await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({ + status: "settled", + }); + expect(fixture.manager.getAgent(fixture.agentId)).toMatchObject({ + lifecycle: "idle", + activeForegroundTurnId: null, + }); + } finally { + fixture.cleanup(); } }); @@ -4290,7 +4436,7 @@ test("replaceAgentRun does not emit idle or resolve waiters between interrupted await manager.waitForAgentRunStart(snapshot.id); const waitPromise = manager.waitForAgentEvent(snapshot.id); - const secondRun = manager.replaceAgentRun(snapshot.id, "second run"); + const secondRun = await manager.replaceAgentRun(snapshot.id, "second run"); const secondRunDrain = (async () => { for await (const _event of secondRun) { // Drain replacement run. @@ -4421,12 +4567,7 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the const replaceUpdatesStart = stateUpdates.length; const beforeReplaceUpdatedAt = manager.getAgent(snapshot.id)?.updatedAt.getTime() ?? 0; - const secondRun = manager.replaceAgentRun(snapshot.id, "replacement run"); - const secondRunDrain = (async () => { - for await (const _event of secondRun) { - // Drain replacement run. - } - })(); + const secondRunPromise = manager.replaceAgentRun(snapshot.id, "replacement run"); await interruptStarted.promise; const replacementUpdates = stateUpdates.slice(replaceUpdatesStart); @@ -4438,15 +4579,28 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the expect(replacementUpdates.map((update) => update.lifecycle)).not.toContain("idle"); allowInterruptToFinish.resolve(); + const secondRun = await secondRunPromise; + const secondRunDrain = (async () => { + for await (const _event of secondRun) { + // Drain replacement run. + } + })(); await secondStartEntered.promise; const replaceGapSnapshot = manager.getAgent(snapshot.id) as | { pendingReplacement: boolean; activeForegroundTurnId: string | null; lifecycle: string } | undefined; - expect(replaceGapSnapshot?.pendingReplacement).toBe(false); + expect(replaceGapSnapshot?.pendingReplacement).toBe(true); expect(replaceGapSnapshot?.activeForegroundTurnId).toBeNull(); expect(replaceGapSnapshot?.lifecycle).toBe("running"); + const replacementStart = manager.waitForAgentRunStart(snapshot.id); + const prematureStart = await Promise.race([ + replacementStart.then(() => "resolved"), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + expect(prematureStart).toBe("pending"); + capturedSession!.pushEvent({ type: "turn_completed", provider: "codex", turnId: "turn-1" }); await new Promise((resolve) => setTimeout(resolve, 0)); @@ -4458,7 +4612,7 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the allowSecondStartToResolve.resolve(); - await manager.waitForAgentRunStart(snapshot.id); + await replacementStart; await firstRunDrain; await secondRunDrain; unsubscribe(); @@ -4546,16 +4700,18 @@ test("applies live autonomous events while no foreground run is active", async ( expect(lifecycleUpdates).toContain("idle"); }); -test("cancelAgentRun can interrupt autonomous running state without a foreground activeForegroundTurnId", async () => { +test("cancelAgentRun waits for an acknowledged autonomous interrupt to settle", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-cancel-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); class LiveInterruptSession extends TestAgentSession { public interruptCount = 0; + readonly interruptCalled = deferred(); override async interrupt(): Promise { this.interruptCount += 1; + this.interruptCalled.resolve(undefined); } } @@ -4618,9 +4774,80 @@ test("cancelAgentRun can interrupt autonomous running state without a foreground expect(beforeCancel?.lifecycle).toBe("running"); expect(beforeCancel?.activeForegroundTurnId).toBeNull(); - const cancelled = await manager.cancelAgentRun(snapshot.id); - expect(cancelled).toBe(true); + let cancelSettled = false; + const cancelPromise = manager.cancelAgentRun(snapshot.id).finally(() => { + cancelSettled = true; + }); + await capturedSession.interruptCalled.promise; + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(cancelSettled).toBe(false); expect(client.lastSession?.interruptCount).toBe(1); + + capturedSession.pushEvent({ + type: "turn_canceled", + provider: "codex", + turnId: "autonomous-cancel-1", + reason: "interrupted", + }); + + await expect(cancelPromise).resolves.toEqual({ status: "settled" }); + expect(manager.getAgent(snapshot.id)?.lifecycle).toBe("idle"); +}); + +test("failed replacement cancellation preserves an autonomous running state", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-replace-rejected-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + + class RejectingLiveInterruptSession extends TestAgentSession { + override async interrupt(): Promise { + throw new Error("provider still owns the autonomous turn"); + } + } + + class RejectingLiveInterruptClient extends TestAgentClient { + readonly session = new RejectingLiveInterruptSession({ + provider: "codex", + cwd: workdir, + }); + + override async createSession(): Promise { + return this.session; + } + } + + const client = new RejectingLiveInterruptClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + rescueTimeouts: { interruptSessionMs: 10 }, + idFactory: () => "00000000-0000-4000-8000-000000000130", + }); + + try { + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + const running = waitForAgentLifecycle(manager, agent.id, "running"); + + client.session.pushEvent({ + type: "turn_started", + provider: "codex", + turnId: "autonomous-replace-1", + }); + await running; + + await expect(manager.replaceAgentRun(agent.id, "replacement prompt")).rejects.toThrow( + `Cannot replace agent ${agent.id} because its active run cancellation was not acknowledged`, + ); + expect(manager.getAgent(agent.id)).toMatchObject({ + lifecycle: "running", + activeForegroundTurnId: null, + }); + } finally { + rmSync(workdir, { recursive: true, force: true }); + } }); test("waitForAgentEvent waitForActive resolves for autonomous live-event run", async () => { @@ -7217,7 +7444,7 @@ test("replaceAgentRun succeeds when foreground turn terminal event is never deli // Replace the hung run. cancelAgentRun will time out after 2s because // no terminal event arrives. After the fix, it should force-clear the // stale foreground state so streamAgent can proceed. - const secondRun = manager.replaceAgentRun(snapshot.id, "replacement prompt"); + const secondRun = await manager.replaceAgentRun(snapshot.id, "replacement prompt"); const collectedEvents: AgentStreamEvent[] = []; const secondRunDrain = (async () => { for await (const event of secondRun) { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index f74bbdc99..fa34cbaa1 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -59,7 +59,7 @@ import { AgentStreamCoalescer, } from "./agent-stream-coalescer.js"; import { limitAgentTimelineItemContent } from "./agent-timeline-content.js"; -import { ForegroundRunState, type ForegroundTurnWaiter } from "./foreground-run-state.js"; +import { AgentRunState, type ForegroundTurnWaiter } from "./agent-run-state.js"; import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest"; import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js"; import { isSystemInjectedEnvelope } from "./agent-prompt.js"; @@ -95,6 +95,20 @@ export class AgentManagerShuttingDownError extends Error { } } +export class AgentRunCancellationError extends Error { + constructor(agentId: string, action: "reload" | "replace" | "rewind" | "stop") { + super( + `Cannot ${action} agent ${agentId} because its active run cancellation was not acknowledged`, + ); + this.name = "AgentRunCancellationError"; + } +} + +export type AgentRunCancellationResult = + | { status: "not_running" } + | { status: "settled" } + | { status: "refused" }; + interface PreparedSessionConfig { storedConfig: AgentSessionConfig; launchConfig: AgentSessionConfig; @@ -543,7 +557,7 @@ export class AgentManager { private readonly providerSubagents = new ProviderSubagentStore(); private readonly agentsAwaitingInitialSnapshotPersist = new Set(); private readonly sessionEventTails = new Map>(); - private readonly foregroundRuns = new ForegroundRunState(); + private readonly runs = new AgentRunState(); private readonly subscribers = new Set(); private readonly idFactory: () => string; private readonly registry?: AgentStorage; @@ -722,7 +736,7 @@ export class AgentManager { return ( agent.lifecycle === "running" || Boolean(agent.activeForegroundTurnId) || - this.foregroundRuns.hasPendingRun(agentId) + this.runs.hasRun(agentId) ); } @@ -1161,7 +1175,7 @@ export class AgentManager { this.assertAcceptingAgentRegistrations(); let existing = this.requireSessionAgent(agentId); if (this.hasInFlightRun(agentId)) { - await this.cancelAgentRun(agentId); + await this.cancelAgentRunBefore(agentId, "reload"); existing = this.requireSessionAgent(agentId); } const rehydrateFromDisk = options?.rehydrateFromDisk ?? false; @@ -1836,13 +1850,13 @@ export class AgentManager { turnId: existingAgent.activeForegroundTurnId ?? undefined, lifecycle: existingAgent.lifecycle, activeForegroundTurnId: existingAgent.activeForegroundTurnId, - hasPendingForegroundRun: this.foregroundRuns.hasPendingRun(agentId), + hasTrackedRun: this.runs.hasRun(agentId), promptType: typeof prompt === "string" ? "string" : "structured", hasRunOptions: Boolean(options), }, "agent.manager.stream.request", ); - if (existingAgent.activeForegroundTurnId || this.foregroundRuns.hasPendingRun(agentId)) { + if (existingAgent.activeForegroundTurnId || this.runs.hasRun(agentId)) { this.logger.trace( { agentId, @@ -1850,7 +1864,7 @@ export class AgentManager { sessionId: existingAgent.persistence?.sessionId ?? undefined, turnId: existingAgent.activeForegroundTurnId ?? undefined, lifecycle: existingAgent.lifecycle, - hasPendingForegroundRun: this.foregroundRuns.hasPendingRun(agentId), + hasTrackedRun: this.runs.hasRun(agentId), }, "agent.manager.stream.reject", ); @@ -1858,18 +1872,19 @@ export class AgentManager { } const agent = existingAgent; - agent.pendingReplacement = false; + const isReplacement = agent.pendingReplacement; agent.lastError = undefined; - const pendingRun = this.foregroundRuns.createPendingRun(agentId); + const pendingRun = this.runs.createPendingRun(agentId); const streamForwarder = async function* streamForwarder(this: AgentManager) { let turnId: string; - let turnStream: ReturnType | null = null; + 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", @@ -1877,11 +1892,15 @@ export class AgentManager { error: errorMsg, }); this.finalizeForegroundTurn(agent); - this.foregroundRuns.settlePendingRun(agentId, pendingRun.token); + this.runs.settleForegroundRun(agentId, pendingRun.token); throw error; } pendingRun.started = true; + pendingRun.turnId = turnId; + if (isReplacement) { + agent.pendingReplacement = false; + } agent.activeForegroundTurnId = turnId; agent.lifecycle = "running"; this.touchUpdatedAt(agent); @@ -1898,8 +1917,8 @@ export class AgentManager { "agent.manager.stream.start", ); - turnStream = this.foregroundRuns.createTurnStream(turnId); - this.foregroundRuns.addWaiter(agent, turnStream.waiter); + turnStream = this.runs.createTurnStream(turnId); + this.runs.addWaiter(agent, turnStream.waiter); try { for await (const event of turnStream.events(isTurnTerminalEvent)) { @@ -1907,9 +1926,9 @@ export class AgentManager { } } finally { if (turnStream) { - this.foregroundRuns.deleteWaiter(agent, turnStream.waiter); + this.runs.deleteWaiter(agent, turnStream.waiter); } - this.foregroundRuns.settlePendingRun(agentId, pendingRun.token); + this.runs.settleForegroundRun(agentId, pendingRun.token); if (!agent.activeForegroundTurnId) { await this.refreshRuntimeInfo(agent); } @@ -1922,7 +1941,7 @@ export class AgentManager { private finalizeForegroundTurn(agent: ActiveManagedAgent, turnId?: string): void { const mutableAgent = agent; if (turnId) { - this.foregroundRuns.rememberFinalizedTurn(mutableAgent, turnId); + this.runs.rememberFinalizedTurn(mutableAgent, turnId); } mutableAgent.activeForegroundTurnId = null; const terminalError = mutableAgent.lastError; @@ -1962,16 +1981,16 @@ export class AgentManager { } } - replaceAgentRun( + async replaceAgentRun( agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions, - ): AsyncGenerator { + ): Promise> { const snapshot = this.requireAgent(agentId); if ( snapshot.lifecycle !== "running" && !snapshot.activeForegroundTurnId && - !this.foregroundRuns.hasPendingRun(agentId) + !this.runs.hasRun(agentId) ) { return this.streamAgent(agentId, prompt, options); } @@ -1982,27 +2001,16 @@ export class AgentManager { this.touchUpdatedAt(agent); this.emitState(agent); - return async function* replaceRunForwarder(this: AgentManager) { - try { - await this.cancelAgentRun(agentId); - const nextRun = this.streamAgent(agentId, prompt, options); - for await (const event of nextRun) { - yield event; - } - } catch (error) { - const latest = this.agents.get(agentId); - if (latest) { - const latestActive = latest; - latestActive.pendingReplacement = false; - if (!latestActive.activeForegroundTurnId && latestActive.lifecycle === "running") { - (latestActive as ActiveManagedAgent).lifecycle = "idle"; - this.touchUpdatedAt(latestActive); - this.emitState(latestActive); - } - } - throw error; + try { + await this.cancelAgentRunBefore(agentId, "replace"); + return this.streamAgent(agentId, prompt, options); + } catch (error) { + const latest = this.agents.get(agentId); + if (latest) { + latest.pendingReplacement = false; } - }.call(this); + throw error; + } } async waitForAgentRunStart(agentId: string, options?: WaitForAgentStartOptions): Promise { @@ -2011,7 +2019,7 @@ export class AgentManager { throw new Error(`Agent ${agentId} not found`); } - const pendingRun = this.foregroundRuns.getPendingRun(agentId); + const pendingRun = this.runs.getPendingRun(agentId); if ((snapshot.lifecycle === "running" || pendingRun?.started) && !snapshot.pendingReplacement) { return; } @@ -2075,7 +2083,7 @@ export class AgentManager { return true; } - const currentPendingRun = this.foregroundRuns.getPendingRun(agentId); + const currentPendingRun = this.runs.getPendingRun(agentId); if ( (current.lifecycle === "running" || currentPendingRun?.started) && !current.pendingReplacement @@ -2146,106 +2154,70 @@ export class AgentManager { } } - async cancelAgentRun(agentId: string): Promise { + async cancelAgentRun(agentId: string): Promise { const agent = this.requireSessionAgent(agentId); - const pendingRun = this.foregroundRuns.getPendingRun(agentId); - const foregroundTurnId = agent.activeForegroundTurnId; - const hasForegroundTurn = Boolean(foregroundTurnId); - const isAutonomousRunning = agent.lifecycle === "running" && !hasForegroundTurn && !pendingRun; - - if (!hasForegroundTurn && !isAutonomousRunning && !pendingRun) { - return false; + const run = + this.runs.getRun(agentId) ?? + (agent.lifecycle === "running" ? this.runs.trackAutonomousRun(agentId, null) : null); + if (!run) { + return { status: "not_running" }; } - await this.interruptSession(agent.session, agentId); + const interruptAcknowledged = await this.interruptSession(agent.session, agentId); + const settlement = await this.waitWithTimeout({ + operation: run.settledPromise, + timeoutMs: interruptAcknowledged + ? INTERRUPT_SESSION_TIMEOUT_MS + : this.rescueTimeouts.interruptSessionMs, + }); - // The interrupt will produce a turn_canceled/turn_failed event via subscribe(), - // which flows through the session event dispatcher and settles the foreground turn waiter. - // Wait briefly for the event to propagate if there's an active foreground turn. - if (foregroundTurnId) { - const waiter = Array.from(agent.foregroundTurnWaiters).find( - (candidate) => candidate.turnId === foregroundTurnId, - ); - const timeout = new Promise((resolvePromise) => setTimeout(resolvePromise, 2000)); - if (waiter) { - await Promise.race([waiter.settledPromise, timeout]); - } else if (agent.activeForegroundTurnId === foregroundTurnId) { - await Promise.race([ - new Promise((resolvePromise) => { - const unsubscribe = this.subscribe( - (event) => { - if ( - event.type === "agent_state" && - event.agent.id === agentId && - !event.agent.activeForegroundTurnId - ) { - unsubscribe(); - resolvePromise(); - } - }, - { agentId, replayState: false }, - ); - }), - timeout, - ]); - } - // The waiter settling wakes up the streamForwarder generator, but its - // finally block (which deletes the pendingForegroundRun) runs asynchronously. - // Wait for the pending run to be fully cleaned up so the next streamAgent - // call doesn't see a stale entry and reject with "already has an active run". - if (pendingRun && !pendingRun.settled) { - await Promise.race([pendingRun.settledPromise, timeout]); - } - } else if (pendingRun) { - const timeout = new Promise((resolvePromise) => setTimeout(resolvePromise, 2000)); - await Promise.race([pendingRun.settledPromise, timeout]); + if (!interruptAcknowledged) { + return { status: settlement === "completed" ? "settled" : "refused" }; } - // If the foreground turn is still stuck after the timeout, force-dispatch a - // synthetic turn_canceled so the normal event pipeline cleans up - // activeForegroundTurnId, settles waiters, and unblocks the streamForwarder. - if (foregroundTurnId && agent.activeForegroundTurnId === foregroundTurnId) { + if (settlement === "timed_out" && run.turnId) { this.logger.warn( - { agentId, foregroundTurnId }, - "cancelAgentRun: foreground turn still active after timeout, force-canceling", + { agentId, turnId: run.turnId, kind: run.kind }, + "cancelAgentRun: acknowledged turn still active after timeout, force-canceling", ); - void this.dispatchSessionEvent(agent, { + await this.dispatchSessionEvent(agent, { + type: "turn_canceled", + provider: agent.provider, + reason: "interrupted", + turnId: run.turnId, + }); + await run.settledPromise; + } else if (settlement === "timed_out" && run.kind === "autonomous") { + this.logger.warn( + { agentId, 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: foregroundTurnId, }); - // The synthetic event unblocks the streamForwarder generator, whose finally - // block settles the pending foreground run asynchronously. Wait for it. - const staleRun = this.foregroundRuns.getPendingRun(agentId); - if (staleRun && !staleRun.settled) { - await staleRun.settledPromise; - } } - // Clear any pending permissions that weren't cleaned up by handleStreamEvent. if (agent.pendingPermissions.size > 0) { - for (const [requestId] of agent.pendingPermissions) { - this.dispatchStream( - agent.id, - { - type: "permission_resolved", - provider: agent.provider, - requestId, - resolution: { behavior: "deny", message: "Interrupted" }, - }, - { timestamp: new Date().toISOString() }, - ); - } - agent.pendingPermissions.clear(); + this.resolvePendingPermissionsForAgent(agent, agent.provider, undefined, "Interrupted"); this.touchUpdatedAt(agent); this.emitState(agent); } - - return true; + return { status: "settled" }; } - private async interruptSession(session: AgentSession, agentId: string): Promise { + private async cancelAgentRunBefore( + agentId: string, + action: "reload" | "replace" | "rewind", + ): Promise { + const result = await this.cancelAgentRun(agentId); + if (result.status === "refused") { + throw new AgentRunCancellationError(agentId, action); + } + } + + private async interruptSession(session: AgentSession, agentId: string): Promise { try { const result = await this.waitWithTimeout({ operation: session.interrupt(), @@ -2263,9 +2235,12 @@ export class AgentManager { { agentId, timeoutMs: this.rescueTimeouts.interruptSessionMs }, "Timed out interrupting session during cancel", ); + return false; } + return true; } catch (error) { this.logger.error({ err: error, agentId }, "Failed to interrupt session"); + return false; } } @@ -2294,13 +2269,11 @@ export class AgentManager { async rewind(agentId: string, messageId: string, mode: RewindMode): Promise { const agent = this.requireSessionAgent(agentId); - const hadActiveRun = - Boolean(agent.activeForegroundTurnId) || this.foregroundRuns.hasPendingRun(agentId); - if (hadActiveRun) { - await this.cancelAgentRun(agentId); + if (this.hasInFlightRun(agentId)) { + await this.cancelAgentRunBefore(agentId, "rewind"); } - const lock = this.foregroundRuns.createPendingRun(agentId); + const lock = this.runs.createPendingRun(agentId); try { this.logger.info( { agentId, provider: agent.provider, messageId, mode }, @@ -2323,7 +2296,7 @@ export class AgentManager { ); throw error; } finally { - this.foregroundRuns.settlePendingRun(agentId, lock.token); + this.runs.settleForegroundRun(agentId, lock.token); } } @@ -2421,7 +2394,7 @@ export class AgentManager { throw new Error(`Agent ${agentId} not found`); } - const pendingForegroundRun = this.foregroundRuns.getPendingRun(agentId); + const pendingForegroundRun = this.runs.getPendingRun(agentId); const hasForegroundTurn = Boolean(snapshot.activeForegroundTurnId) || Boolean(pendingForegroundRun); @@ -2793,13 +2766,13 @@ export class AgentManager { agent.unsubscribeSession(); agent.unsubscribeSession = null; } - this.foregroundRuns.cancelWaiters(agent, (turnId) => ({ + this.runs.cancelWaiters(agent, (turnId) => ({ type: "turn_canceled", provider: agent.provider, reason: cancelReason, turnId, })); - this.foregroundRuns.settlePendingRun(agent.id); + this.runs.clearAgentRun(agent.id); return { ...agent, lifecycle: "closed", @@ -2883,7 +2856,7 @@ export class AgentManager { return; } const turnId = getAgentStreamEventTurnId(event); - const matchingWaiters = this.foregroundRuns.getMatchingWaiters(agent, turnId); + const matchingWaiters = this.runs.getMatchingWaiters(agent, turnId); this.logger.trace( { agentId: agent.id, @@ -2902,7 +2875,7 @@ export class AgentManager { return; } - this.foregroundRuns.notifyWaiters(matchingWaiters, event, { + this.runs.notifyWaiters(matchingWaiters, event, { terminal: isTurnTerminalEvent(event), }); this.logger.trace( @@ -3121,7 +3094,7 @@ export class AgentManager { return; } - this.foregroundRuns.notifyAgentWaiters(agent, event); + this.runs.notifyAgentWaiters(agent, event); this.logger.trace( { agentId, @@ -3151,7 +3124,7 @@ export class AgentManager { if ( eventTurnId && isTurnTerminalEvent(event) && - this.foregroundRuns.hasFinalizedTurn(agent, eventTurnId) + this.runs.hasFinalizedTurn(agent, eventTurnId) ) { return false; } @@ -3180,8 +3153,11 @@ export class AgentManager { await dispatchPromise; } - if (!options?.fromHistory && isForegroundEvent && isTurnTerminalEvent(event)) { - this.finalizeForegroundTurn(agent, eventTurnId); + if (!options?.fromHistory && isTurnTerminalEvent(event)) { + this.runs.settleTerminalRun(agent.id, eventTurnId); + if (isForegroundEvent) { + this.finalizeForegroundTurn(agent, eventTurnId); + } } if (!options?.fromHistory && flags.shouldDispatchEvent) { @@ -3496,6 +3472,7 @@ export class AgentManager { "agent.manager.turn.started", ); if (!isForegroundEvent) { + this.runs.trackAutonomousRun(agent.id, eventTurnId ?? null); agent.lifecycle = "running"; this.emitState(agent); } diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index a88a8b9a3..d726509a8 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -1,4 +1,5 @@ import { expect, it, test, vi } from "vitest"; +import pino, { type Logger } from "pino"; import { createTestLogger } from "../../test-utils/test-logger.js"; import { AgentManager } from "./agent-manager.js"; @@ -11,8 +12,34 @@ import { } from "./agent-prompt.js"; import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js"; +interface CapturedLogger { + logger: Logger; + records: Array>; + nextRecord: Promise; +} + +function createCapturedLogger(): CapturedLogger { + const records: Array> = []; + let resolveNextRecord!: () => void; + const nextRecord = new Promise((resolve) => { + resolveNextRecord = resolve; + }); + const logger = pino( + { level: "error" }, + { + write(line: string) { + records.push(JSON.parse(line) as Record); + resolveNextRecord(); + }, + }, + ); + return { logger, records, nextRecord }; +} + interface FinishNotificationScenarioOptions { childLastAssistantMessage?: string | null; + parentPromptError?: Error; + logger?: Logger; } interface FinishNotificationScenario { @@ -56,11 +83,15 @@ function createFinishNotificationScenario( return options?.childLastAssistantMessage ?? null; }); Reflect.set(agentManager, "tryRunOutOfBand", () => false); - Reflect.set(agentManager, "hasInFlightRun", () => false); + Reflect.set(agentManager, "hasInFlightRun", () => Boolean(options?.parentPromptError)); Reflect.set(agentManager, "streamAgent", (_agentId: string, prompt: string) => { resolveParentPrompt?.(prompt); return (async function* noop() {})(); }); + Reflect.set(agentManager, "replaceAgentRun", async (_agentId: string, prompt: string) => { + resolveParentPrompt?.(prompt); + throw options?.parentPromptError; + }); const agentStorage: AgentStorage = Object.create(AgentStorage.prototype); Reflect.set(agentStorage, "get", async (agentId: string) => { @@ -77,7 +108,7 @@ function createFinishNotificationScenario( agentStorage, childAgentId: "child-agent", callerAgentId: "caller-agent", - logger: createTestLogger(), + logger: options?.logger ?? createTestLogger(), }); }, async finishChildAndReadParentPrompt() { @@ -161,6 +192,28 @@ test("finish notifications tell the parent the child's last assistant message", ); }); +test("finish notifications log a rejected parent prompt without an unhandled rejection", async () => { + const captured = createCapturedLogger(); + const scenario = createFinishNotificationScenario({ + parentPromptError: new Error("parent provider rejected replacement"), + logger: captured.logger, + }); + + scenario.startWatchingChild(); + await scenario.finishChildAndReadParentPrompt(); + await captured.nextRecord; + + expect(captured.records).toEqual([ + expect.objectContaining({ + msg: "Failed to notify caller agent", + childAgentId: "child-agent", + callerAgentId: "caller-agent", + reason: "finished", + err: expect.objectContaining({ message: "parent provider rejected replacement" }), + }), + ]); +}); + it("does not notify archived callers", async () => { let subscriber: ((event: AgentManagerEvent) => void) | null = null; diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 74664b630..0178ee578 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -15,13 +15,13 @@ export interface StartAgentRunOptions { runOptions?: AgentRunOptions; } -export function startAgentRun( +export async function startAgentRun( agentManager: AgentRunController, agentId: string, prompt: AgentPromptInput, logger: Logger, options?: StartAgentRunOptions, -): { outOfBand: boolean } { +): Promise<{ outOfBand: boolean }> { const snapshot = agentManager.getAgent(agentId); logger.trace( { @@ -44,7 +44,7 @@ export function startAgentRun( const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); const runOptions = options?.runOptions; const iterator = shouldReplace - ? agentManager.replaceAgentRun(agentId, prompt, runOptions) + ? await agentManager.replaceAgentRun(agentId, prompt, runOptions) : agentManager.streamAgent(agentId, prompt, runOptions); logger.trace( { @@ -197,7 +197,7 @@ export async function sendPromptToAgent( ? { ...params.runOptions, messageId: params.messageId } : params.runOptions; - return startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, { + return await startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, { replaceRunning: true, runOptions, }); @@ -215,7 +215,7 @@ export async function startCreatedAgentInitialPrompt( return currentSnapshot; } - const dispatchResult = startAgentRun( + const dispatchResult = await startAgentRun( params.agentManager, params.agentId, params.prompt, @@ -298,6 +298,15 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): }); } + function notifySafely(reason: "finished" | "errored" | "needs permission"): void { + void notify(reason).catch((error) => { + logger.error( + { err: error, childAgentId, callerAgentId, reason }, + "Failed to notify caller agent", + ); + }); + } + unsubscribe = agentManager.subscribe( (event) => { if (fired) { @@ -310,11 +319,11 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): return; } if (event.agent.lifecycle === "error") { - void notify("errored"); + notifySafely("errored"); return; } if (event.agent.lifecycle === "idle" && hasSeenRunning) { - void notify("finished"); + notifySafely("finished"); return; } if (event.agent.lifecycle === "closed") { @@ -326,7 +335,7 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): } if (event.event.type === "permission_requested") { - void notify("needs permission"); + notifySafely("needs permission"); } }, { agentId: childAgentId, replayState: false }, @@ -345,6 +354,6 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): if (childSnapshot.lifecycle === "running") { hasSeenRunning = true; } else if (childSnapshot.lifecycle === "error") { - void notify("errored"); + notifySafely("errored"); } } diff --git a/packages/server/src/server/agent/foreground-run-state.ts b/packages/server/src/server/agent/agent-run-state.ts similarity index 65% rename from packages/server/src/server/agent/foreground-run-state.ts rename to packages/server/src/server/agent/agent-run-state.ts index 705c59a38..e74731718 100644 --- a/packages/server/src/server/agent/foreground-run-state.ts +++ b/packages/server/src/server/agent/agent-run-state.ts @@ -12,45 +12,102 @@ export interface ForegroundTurnWaiter { export interface PendingForegroundRun { token: string; + kind: "foreground"; + turnId: string | null; started: boolean; settled: boolean; settledPromise: Promise; resolveSettled: () => void; } +export interface AutonomousAgentRun { + token: string; + kind: "autonomous"; + turnId: string | null; + started: true; + settled: boolean; + settledPromise: Promise; + resolveSettled: () => void; +} + +export type TrackedAgentRun = PendingForegroundRun | AutonomousAgentRun; + export interface ForegroundRunAgentState { foregroundTurnWaiters: Set; finalizedForegroundTurnIds: Set; } -export class ForegroundRunState { - private readonly pendingRuns = new Map(); +export class AgentRunState { + private readonly runs = new Map(); createPendingRun(agentId: string): PendingForegroundRun { const pendingRun = createPendingForegroundRun(); - this.pendingRuns.set(agentId, pendingRun); + this.runs.set(agentId, pendingRun); return pendingRun; } getPendingRun(agentId: string): PendingForegroundRun | null { - return this.pendingRuns.get(agentId) ?? null; + const run = this.runs.get(agentId); + return run?.kind === "foreground" ? run : null; } hasPendingRun(agentId: string): boolean { - return this.pendingRuns.has(agentId); + return this.getPendingRun(agentId) !== null; } - settlePendingRun(agentId: string, token?: string): void { - const pendingRun = this.pendingRuns.get(agentId); - if (!pendingRun) { + getRun(agentId: string): TrackedAgentRun | null { + return this.runs.get(agentId) ?? null; + } + + hasRun(agentId: string): boolean { + return this.runs.has(agentId); + } + + trackAutonomousRun(agentId: string, turnId: string | null): TrackedAgentRun { + const current = this.runs.get(agentId); + if (current) { + return current; + } + + const run = createTrackedRun({ kind: "autonomous", turnId, started: true }); + this.runs.set(agentId, run); + return run; + } + + settleTerminalRun(agentId: string, turnId: string | undefined): void { + const run = this.runs.get(agentId); + if (!run) { return; } - if (token && pendingRun.token !== token) { + if (run.kind === "foreground" && (run.turnId === null || run.turnId !== turnId)) { + return; + } + if ( + run.kind === "autonomous" && + run.turnId !== null && + turnId !== undefined && + run.turnId !== turnId + ) { return; } - this.pendingRuns.delete(agentId); - settlePendingForegroundRun(pendingRun); + this.clearRun(agentId, run); + } + + settleForegroundRun(agentId: string, token: string): void { + const run = this.runs.get(agentId); + if (run?.kind !== "foreground" || run.token !== token) { + return; + } + + this.clearRun(agentId, run); + } + + clearAgentRun(agentId: string): void { + const run = this.runs.get(agentId); + if (run) { + this.clearRun(agentId, run); + } } createTurnStream(turnId: string): ForegroundTurnStream { @@ -120,14 +177,6 @@ export class ForegroundRunState { agent.foregroundTurnWaiters.clear(); } - clearAgent(agentId: string, agent: ForegroundRunAgentState): void { - for (const waiter of agent.foregroundTurnWaiters) { - this.settleWaiter(waiter); - } - agent.foregroundTurnWaiters.clear(); - this.settlePendingRun(agentId); - } - rememberFinalizedTurn(agent: ForegroundRunAgentState, turnId: string): void { agent.finalizedForegroundTurnIds.add(turnId); if (agent.finalizedForegroundTurnIds.size <= 50) { @@ -143,6 +192,11 @@ export class ForegroundRunState { hasFinalizedTurn(agent: ForegroundRunAgentState, turnId: string): boolean { return agent.finalizedForegroundTurnIds.has(turnId); } + + private clearRun(agentId: string, run: TrackedAgentRun): void { + this.runs.delete(agentId); + settleTrackedRun(run); + } } export class ForegroundTurnStream { @@ -205,24 +259,42 @@ export class ForegroundTurnStream { } function createPendingForegroundRun(): PendingForegroundRun { + return createTrackedRun({ kind: "foreground", turnId: null, started: false }); +} + +function createTrackedRun(input: { + kind: "foreground"; + turnId: null; + started: false; +}): PendingForegroundRun; +function createTrackedRun(input: { + kind: "autonomous"; + turnId: string | null; + started: true; +}): AutonomousAgentRun; +function createTrackedRun( + input: + | { kind: "foreground"; turnId: null; started: false } + | { kind: "autonomous"; turnId: string | null; started: true }, +): TrackedAgentRun { let resolveSettled!: () => void; const settledPromise = new Promise((resolvePromise) => { resolveSettled = resolvePromise; }); return { token: randomUUID(), - started: false, + ...input, settled: false, settledPromise, resolveSettled, }; } -function settlePendingForegroundRun(pendingRun: PendingForegroundRun): void { - if (pendingRun.settled) { +function settleTrackedRun(run: TrackedAgentRun): void { + if (run.settled) { return; } - pendingRun.settled = true; - pendingRun.resolveSettled(); + run.settled = true; + run.resolveSettled(); } diff --git a/packages/server/src/server/agent/lifecycle-command.test.ts b/packages/server/src/server/agent/lifecycle-command.test.ts index 9a3da4aa6..6e5714d48 100644 --- a/packages/server/src/server/agent/lifecycle-command.test.ts +++ b/packages/server/src/server/agent/lifecycle-command.test.ts @@ -43,6 +43,8 @@ class FakeLifecycleAgentManager implements LifecycleAgentManager { readonly modeUpdates: Array<{ agentId: string; modeId: string }> = []; readonly detachedAgentIds: string[] = []; inFlightAgentIds = new Set(); + readonly settledDuringCancellationAgentIds = new Set(); + readonly rejectedCancellationAgentIds = new Set(); constructor(private readonly storage: FakeLifecycleAgentStorage) {} @@ -54,9 +56,18 @@ class FakeLifecycleAgentManager implements LifecycleAgentManager { return this.inFlightAgentIds.has(agentId); } - async cancelAgentRun(agentId: string): Promise { + async cancelAgentRun(agentId: string) { this.cancelledAgentIds.push(agentId); - return this.inFlightAgentIds.delete(agentId); + if (this.settledDuringCancellationAgentIds.delete(agentId)) { + this.inFlightAgentIds.delete(agentId); + return { status: "not_running" } as const; + } + if (this.rejectedCancellationAgentIds.has(agentId)) { + return { status: "refused" } as const; + } + return this.inFlightAgentIds.delete(agentId) + ? ({ status: "settled" } as const) + : ({ status: "not_running" } as const); } async clearAgentAttention(agentId: string): Promise { @@ -168,6 +179,21 @@ describe("agent lifecycle commands", () => { expect(manager.cancelledAgentIds).toEqual(["agent-1"]); }); + test("accepts a stop when the run settles during cancellation", async () => { + const storage = new FakeLifecycleAgentStorage(); + const manager = new FakeLifecycleAgentManager(storage); + manager.liveAgents.set("agent-1", managedAgent("agent-1", "running")); + manager.inFlightAgentIds.add("agent-1"); + manager.settledDuringCancellationAgentIds.add("agent-1"); + + await expect( + cancelAgentRunCommand({ agentManager: manager, logger }, "agent-1"), + ).resolves.toEqual({ + agent: manager.liveAgents.get("agent-1"), + cancelled: false, + }); + }); + test("archives a live agent after canceling and clearing attention", async () => { const storage = new FakeLifecycleAgentStorage(); const manager = new FakeLifecycleAgentManager(storage); @@ -193,6 +219,21 @@ describe("agent lifecycle commands", () => { expect(manager.archivedAgentIds).toEqual(["agent-1"]); }); + test("archives a live agent when its graceful cancellation is rejected", async () => { + const storage = new FakeLifecycleAgentStorage(); + const manager = new FakeLifecycleAgentManager(storage); + manager.liveAgents.set("agent-1", managedAgent("agent-1", "running")); + manager.inFlightAgentIds.add("agent-1"); + manager.rejectedCancellationAgentIds.add("agent-1"); + storage.records.set("agent-1", storedAgent("agent-1")); + + await expect( + archiveAgentCommand({ agentManager: manager, agentStorage: storage, logger }, "agent-1"), + ).resolves.toMatchObject({ agentId: "agent-1" }); + expect(manager.cancelledAgentIds).toEqual(["agent-1"]); + expect(manager.archivedAgentIds).toEqual(["agent-1"]); + }); + test("archives a stored agent when no live agent exists", async () => { const storage = new FakeLifecycleAgentStorage(); const manager = new FakeLifecycleAgentManager(storage); diff --git a/packages/server/src/server/agent/lifecycle-command.ts b/packages/server/src/server/agent/lifecycle-command.ts index b6318c383..cb172ce62 100644 --- a/packages/server/src/server/agent/lifecycle-command.ts +++ b/packages/server/src/server/agent/lifecycle-command.ts @@ -1,6 +1,10 @@ import type { Logger } from "pino"; -import type { ManagedAgent } from "./agent-manager.js"; +import { + AgentRunCancellationError, + type AgentRunCancellationResult, + type ManagedAgent, +} from "./agent-manager.js"; import type { StoredAgentRecord } from "./agent-storage.js"; import type { AgentProviderNotice } from "./agent-sdk-types.js"; @@ -9,7 +13,7 @@ export type LifecycleAgentSnapshot = Pick; + cancelAgentRun(agentId: string): Promise; clearAgentAttention(agentId: string): Promise; archiveAgent(agentId: string): Promise<{ archivedAt: string }>; archiveSnapshot(agentId: string, archivedAt: string): Promise; @@ -47,10 +51,14 @@ export interface CancelAgentRunResult { cancelled: boolean; } -export async function cancelAgentRunCommand( +interface RequestedAgentRunCancellation extends CancelAgentRunResult { + cancellation: AgentRunCancellationResult; +} + +async function requestAgentRunCancellation( dependencies: Pick, agentId: string, -): Promise { +): Promise { const { agentManager, logger } = dependencies; const agent = agentManager.getAgent(agentId); if (!agent) { @@ -64,7 +72,7 @@ export async function cancelAgentRunCommand( { agentId, lifecycle: agent.lifecycle, hasInFlightRun }, "cancelAgentRunCommand: skipping because agent is not running", ); - return { agent, cancelled: false }; + return { agent, cancelled: false, cancellation: { status: "not_running" } }; } logger.debug( @@ -72,23 +80,33 @@ export async function cancelAgentRunCommand( "cancelAgentRunCommand: interrupting", ); const startedAt = Date.now(); - const cancelled = await agentManager.cancelAgentRun(agentId); + const cancellation = await agentManager.cancelAgentRun(agentId); logger.debug( - { agentId, cancelled, durationMs: Date.now() - startedAt }, + { agentId, cancellation: cancellation.status, durationMs: Date.now() - startedAt }, "cancelAgentRunCommand: cancelAgentRun completed", ); - if (!cancelled) { - logger.warn( + return { + agent, + cancelled: cancellation.status === "settled", + cancellation, + }; +} + +export async function cancelAgentRunCommand( + dependencies: Pick, + agentId: string, +): Promise { + const result = await requestAgentRunCancellation(dependencies, agentId); + if (result.cancellation.status === "refused") { + dependencies.logger.warn( { agentId }, "cancelAgentRunCommand: reported running but no active run was cancelled", ); + throw new AgentRunCancellationError(agentId, "stop"); } - return { - agent, - cancelled, - }; + return { agent: result.agent, cancelled: result.cancelled }; } export interface ArchiveAgentResult { @@ -104,7 +122,7 @@ export async function archiveAgentCommand( const liveAgent = dependencies.agentManager.getAgent(agentId); let record: StoredAgentRecord | null; if (liveAgent) { - await cancelAgentRunCommand(dependencies, agentId); + await requestAgentRunCancellation(dependencies, agentId); await dependencies.agentManager.clearAgentAttention(agentId).catch(() => undefined); await dependencies.agentManager.archiveAgent(agentId); record = await dependencies.agentStorage.get(agentId); diff --git a/packages/server/src/server/agent/permission-response.test.ts b/packages/server/src/server/agent/permission-response.test.ts index 89b320b00..a5cf477e6 100644 --- a/packages/server/src/server/agent/permission-response.test.ts +++ b/packages/server/src/server/agent/permission-response.test.ts @@ -53,11 +53,11 @@ class FakePermissionAgentManager { return emptyAgentStream(); } - replaceAgentRun( + async replaceAgentRun( agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions, - ): AsyncGenerator { + ): Promise> { this.replacementRuns.push({ agentId, prompt, options }); return emptyAgentStream(); } diff --git a/packages/server/src/server/agent/permission-response.ts b/packages/server/src/server/agent/permission-response.ts index 5d473e128..7cb58e6d2 100644 --- a/packages/server/src/server/agent/permission-response.ts +++ b/packages/server/src/server/agent/permission-response.ts @@ -33,7 +33,7 @@ export async function respondToAgentPermission( if (result?.followUpPrompt) { logger.debug({ agentId }, "Permission response requires follow-up turn, starting agent stream"); - startAgentRun(agentManager, agentId, result.followUpPrompt, logger, { + await startAgentRun(agentManager, agentId, result.followUpPrompt, logger, { replaceRunning: true, }); } 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 8f77da542..4db6c748c 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 @@ -32,6 +32,7 @@ import { type FakeCodexAppServer, waitForNextPermission, waitForNextTimelineItem, + waitForProviderSubagent, waitForTimelineToolCall, } from "./codex/test-utils/fake-app-server.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; @@ -2457,6 +2458,158 @@ describe("Codex app-server provider", () => { expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1); }); + test("discovers a MultiAgentV2 child from a legacy-only lifecycle notification", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the investigation."); + await appServer.waitForTurnStart(); + const child = waitForProviderSubagent(session, "legacy-only-child-thread"); + const spawn = waitForTimelineToolCall(session, "spawn-legacy-only-child"); + + appServer.startsTurn({ threadId: "thread-1", turnId: "turn-with-legacy-only-child" }); + appServer.startsLegacyOnlySubAgent({ + callId: "spawn-legacy-only-child", + threadId: "legacy-only-child-thread", + agentPath: "/root/legacy-only-child", + }); + + await expect(child).resolves.toMatchObject({ + type: "provider_subagent", + provider: "codex", + turnId: "codex-turn-0", + event: { + type: "upsert", + id: "legacy-only-child-thread", + status: "running", + }, + }); + await expect(spawn).resolves.toMatchObject({ + type: "timeline", + provider: "codex", + turnId: "codex-turn-0", + item: { + type: "tool_call", + callId: "spawn-legacy-only-child", + status: "running", + detail: { + type: "sub_agent", + description: "/root/legacy-only-child", + }, + }, + }); + + appServer.completeTurn(); + await resultPromise; + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("reports when Codex rejects a foreground turn interrupt", async () => { + const appServer = createFakeCodexAppServer({ + "turn/interrupt": async () => { + throw new Error("A foreground turn is already active"); + }, + }); + 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-waiting-for-child" }); + + await expect(session.interrupt()).rejects.toThrow("A foreground turn is already active"); + + appServer.completeTurn(); + await resultPromise; + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("rejects an interrupt until Codex identifies the accepted turn", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Start working."); + await appServer.waitForTurnStart(); + + await expect(session.interrupt()).rejects.toThrow( + "Cannot interrupt Codex before turn/started identifies the active turn", + ); + + appServer.startsTurn({ threadId: "thread-1", turnId: "turn-identified-late" }); + appServer.completeTurn(); + await resultPromise; + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("rejects an interrupt before Codex initializes the thread", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + await expect(session.interrupt()).rejects.toThrow( + "Cannot interrupt Codex before the active thread is initialized", + ); + + await session.close(); + }); + + test("interrupts an autonomous Codex turn identified by live notifications", async () => { + const session = createSession(); + const requests: Array<{ method: string; params: unknown }> = []; + session.activeForegroundTurnId = null; + session.client = { + request: async (method, params) => { + requests.push({ method, params }); + return {}; + }, + }; + + asInternals(session).handleNotification("turn/started", { + threadId: "test-thread", + turn: { id: "autonomous-turn" }, + }); + + await session.interrupt(); + + expect(requests).toContainEqual({ + method: "turn/interrupt", + params: { + threadId: "test-thread", + turnId: "autonomous-turn", + }, + }); + }); + test("never replaces the root identity with an early child thread start", () => { const session = createSession(); 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 dc77fdb7e..48aeddb90 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 @@ -1647,6 +1647,10 @@ function readCodexSubAgentActivity(item: unknown): CodexSubAgentActivity | null }; } +function shouldIgnoreMirroredLifecycleItem(source: "item" | "codex_event", item: unknown): boolean { + return source === "codex_event" && !readCodexSubAgentActivity(item); +} + function settleHistoricalSubAgentActivity( item: ToolCallTimelineItem, kind: CodexSubAgentActivity["kind"], @@ -3800,6 +3804,7 @@ export class CodexAppServerAgentSession implements AgentSession { const turnId = this.createTurnId(); this.activeForegroundTurnId = turnId; + this.currentTurnId = null; try { this.logTurnStartSummary({ @@ -4191,19 +4196,20 @@ export class CodexAppServerAgentSession implements AgentSession { } async interrupt(): Promise { - if (!this.client || !this.currentThreadId || !this.currentTurnId) return; - try { - await this.client.request( - "turn/interrupt", - { - threadId: this.currentThreadId, - turnId: this.currentTurnId, - }, - INTERRUPT_TIMEOUT_MS, - ); - } catch (error) { - this.logger.warn({ error }, "Failed to interrupt Codex turn"); + if (!this.client || !this.currentThreadId) { + throw new Error("Cannot interrupt Codex before the active thread is initialized"); } + if (!this.currentTurnId) { + throw new Error("Cannot interrupt Codex before turn/started identifies the active turn"); + } + await this.client.request( + "turn/interrupt", + { + threadId: this.currentThreadId, + turnId: this.currentTurnId, + }, + INTERRUPT_TIMEOUT_MS, + ); } async close(): Promise { @@ -5528,9 +5534,10 @@ export class CodexAppServerAgentSession implements AgentSession { parsed: Extract, ): void { // Codex emits mirrored lifecycle notifications via both `codex/event/item_*` - // and canonical `item/*`. We render only the canonical channel to avoid - // duplicated assistant/reasoning rows. - if (parsed.source === "codex_event") { + // and canonical `item/*`. Render ordinary items only from the canonical + // channel, but accept a legacy-only child announcement so it can establish + // the provider-subagent route. + if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) { return; } if (this.isUserMessageItem(parsed.item)) { @@ -5684,7 +5691,7 @@ export class CodexAppServerAgentSession implements AgentSession { private handleItemStartedNotification( parsed: Extract, ): void { - if (parsed.source === "codex_event") { + if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) { return; } if (this.isUserMessageItem(parsed.item)) { diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index 9ef0c354e..0e7488e9a 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -58,6 +58,12 @@ export interface FakeCodexAppServer { agentPath: string; parentThreadId?: string; }): void; + startsLegacyOnlySubAgent(params: { + callId: string; + threadId: string; + agentPath: string; + parentThreadId?: string; + }): void; beginsSubAgentActivity(params: FakeSubAgentActivity): void; completesSubAgentActivity(params: FakeSubAgentActivity): void; completesCompaction(params: { threadId: string; itemId: string }): void; @@ -324,6 +330,18 @@ export function createFakeCodexAppServer( startsSubAgent(params) { writeSubAgentActivity("item/completed", { ...params, kind: "started" }); }, + startsLegacyOnlySubAgent(params) { + writeLegacyEvent(params.parentThreadId ?? "thread-1", "codex/event/item_completed", { + type: "item_completed", + item: { + type: "subAgentActivity", + id: params.callId, + kind: "started", + agentThreadId: params.threadId, + agentPath: params.agentPath, + }, + }); + }, beginsSubAgentActivity(params) { writeSubAgentActivity("item/started", params); }, @@ -534,6 +552,7 @@ function waitForNextEvent( } type TimelineEvent = StreamEventOfType<"timeline">; +type ProviderSubagentEvent = StreamEventOfType<"provider_subagent">; export function waitForNextPermission( session: AgentSession, @@ -555,3 +574,10 @@ export function waitForTimelineToolCall( (event) => event.item.type === "tool_call" && event.item.callId === callId, ); } + +export function waitForProviderSubagent( + session: AgentSession, + id: string, +): Promise { + return waitForNextEvent(session, "provider_subagent", (event) => event.event.id === id); +} diff --git a/packages/server/src/server/agent/rewind/rewind.test.ts b/packages/server/src/server/agent/rewind/rewind.test.ts index 15efcb1ad..a321fb79f 100644 --- a/packages/server/src/server/agent/rewind/rewind.test.ts +++ b/packages/server/src/server/agent/rewind/rewind.test.ts @@ -100,6 +100,35 @@ describe("AgentManager rewind", () => { expect(session.recordedRewinds).toEqual([{ mode: "files", messageId: "message-1" }]); }); + test("does not rewind when the in-flight turn rejects cancellation", async () => { + class RejectingInterruptSession extends FakeRewindSession { + override async interrupt(): Promise { + throw new Error("provider still owns the active turn"); + } + } + + const session = new RejectingInterruptSession(); + const manager = new AgentManager({ + clients: { claude: new FakeRewindClient(session) }, + logger: createTestLogger(), + idFactory: () => "00000000-0000-4000-8000-000000000902", + }); + const agent = await manager.createAgent({ provider: "claude", cwd: process.cwd() }, undefined, { + workspaceId: undefined, + }); + const run = manager.streamAgent(agent.id, "keep working"); + await run.next(); + + await expect(manager.rewind(agent.id, "message-1", "files")).rejects.toThrow( + `Cannot rewind agent ${agent.id} because its active run cancellation was not acknowledged`, + ); + expect(session.recordedRewinds).toEqual([]); + expect(manager.getAgent(agent.id)).toMatchObject({ + lifecycle: "running", + activeForegroundTurnId: "turn-1", + }); + }); + test("blocks new prompts until the rehydrate epoch broadcasts", async () => { const historyGate = new RewindHistoryGate(); historyGate.hold(); diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index 89f66f351..944ca4465 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -210,105 +210,120 @@ test("createAgent with background initialPrompt returns a running snapshot befor } }); +interface StubAgentOptions { + sessionId: string; + supportsStreaming: boolean; + startError?: string; + interruptError?: string; +} + +class StubAgentSession implements AgentSession { + readonly provider = "codex" as const; + readonly capabilities; + private activeTurnId: string | null = null; + + constructor(private readonly options: StubAgentOptions) { + this.capabilities = { + supportsStreaming: options.supportsStreaming, + supportsSessionPersistence: true, + supportsDynamicModes: false, + supportsMcpServers: false, + supportsReasoningStream: false, + supportsToolInvocations: false, + supportsRewindConversation: false, + supportsRewindFiles: false, + supportsRewindBoth: false, + } as const; + } + + get id(): string { + return this.options.sessionId; + } + + async run(): Promise { + return { sessionId: this.id, finalText: "", timeline: [] }; + } + + async startTurn(): Promise<{ turnId: string }> { + if (this.options.startError) throw new Error(this.options.startError); + if (this.activeTurnId) throw new Error("A foreground turn is already active"); + this.activeTurnId = "provider-owned-turn"; + return { turnId: this.activeTurnId }; + } + + subscribe(): () => void { + return () => undefined; + } + + async *streamHistory(): AsyncGenerator {} + + async getRuntimeInfo() { + return { + provider: this.provider, + sessionId: this.id, + model: "gpt-5.4-mini", + modeId: "full-access", + }; + } + + async getAvailableModes() { + return [{ id: "full-access", label: "Full access", description: "No prompts" }]; + } + + async getCurrentMode(): Promise { + return "full-access"; + } + + async setMode(): Promise {} + getPendingPermissions() { + return []; + } + async respondToPermission(): Promise {} + describePersistence(): AgentPersistenceHandle { + return { provider: this.provider, sessionId: this.id }; + } + async interrupt(): Promise { + if (this.options.interruptError) throw new Error(this.options.interruptError); + } + async close(): Promise { + this.activeTurnId = null; + } +} + +class StubAgentClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities; + + constructor(private readonly options: StubAgentOptions) { + this.capabilities = new StubAgentSession(options).capabilities; + } + + async isAvailable(): Promise { + return true; + } + async createSession(): Promise { + return new StubAgentSession(this.options); + } + async resumeSession(): Promise { + return new StubAgentSession(this.options); + } + async fetchCatalog() { + return { + models: [{ id: "gpt-5.4-mini", label: "GPT-5.4 mini", provider: this.provider }], + modes: [{ id: "full-access", label: "Full access", description: "No prompts" }], + }; + } +} + test("createAgent fails when the initial turn cannot start", async () => { - class StartTurnFailureSession implements AgentSession { - readonly provider = "codex" as const; - readonly id = "start-turn-failure-session"; - readonly capabilities = { - supportsStreaming: false, - supportsSessionPersistence: true, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: false, - supportsToolInvocations: false, - supportsRewindConversation: false, - supportsRewindFiles: false, - supportsRewindBoth: false, - } as const; - - async run(): Promise { - return { - sessionId: this.id, - finalText: "", - timeline: [], - }; - } - - async startTurn(): Promise<{ turnId: string }> { - throw new Error("Initial turn failed to start"); - } - - subscribe(): () => void { - return () => undefined; - } - - async *streamHistory(): AsyncGenerator { - yield* []; - } - - async getRuntimeInfo() { - return { - provider: "codex" as const, - sessionId: this.id, - model: "gpt-5.4-mini", - modeId: "full-access", - }; - } - - async getAvailableModes(): Promise> { - return [{ id: "full-access", label: "Full access", description: "No prompts" }]; - } - - async getCurrentMode(): Promise { - return "full-access"; - } - - async setMode(): Promise {} - - getPendingPermissions() { - return []; - } - - async respondToPermission(): Promise {} - - describePersistence(): AgentPersistenceHandle | null { - return { provider: "codex", sessionId: this.id }; - } - - async interrupt(): Promise {} - - async close(): Promise {} - } - - class StartTurnFailureClient implements AgentClient { - readonly provider = "codex" as const; - readonly capabilities = { - supportsStreaming: false, - supportsSessionPersistence: true, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: false, - supportsToolInvocations: false, - supportsRewindConversation: false, - supportsRewindFiles: false, - supportsRewindBoth: false, - } as const; - - async isAvailable(): Promise { - return true; - } - - async createSession(_config: AgentSessionConfig): Promise { - return new StartTurnFailureSession(); - } - - async resumeSession(): Promise { - return new StartTurnFailureSession(); - } - } + const testAgent = new StubAgentClient({ + sessionId: "start-turn-failure-session", + supportsStreaming: false, + startError: "Initial turn failed to start", + }); const daemon = await createTestPaseoDaemon({ - agentClients: { codex: new StartTurnFailureClient() }, + agentClients: { codex: testAgent }, }); const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, @@ -335,6 +350,58 @@ test("createAgent fails when the initial turn cannot start", async () => { } }); +function createUninterruptibleClient(): AgentClient { + return new StubAgentClient({ + sessionId: "uninterruptible-session", + supportsStreaming: true, + interruptError: "Provider did not acknowledge cancellation", + }); +} + +test("DaemonClient rejects a replacement prompt when cancellation is not acknowledged", async () => { + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { codex: createUninterruptibleClient() }, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + + try { + await client.connect(); + const agent = await client.createAgent({ provider: "codex", cwd }); + await client.sendMessage(agent.id, "Keep working on the first prompt."); + + await expect(client.sendMessage(agent.id, "Replace it with this prompt.")).rejects.toThrow( + `Cannot replace agent ${agent.id} because its active run cancellation was not acknowledged`, + ); + } finally { + await client.close(); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } +}, 30_000); + +test("DaemonClient rejects Stop when cancellation is not acknowledged", async () => { + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { codex: createUninterruptibleClient() }, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + + try { + await client.connect(); + const agent = await client.createAgent({ provider: "codex", cwd }); + await client.sendMessage(agent.id, "Keep working until stopped."); + + await expect(client.cancelAgent(agent.id)).rejects.toThrow( + `Cannot stop agent ${agent.id} because its active run cancellation was not acknowledged`, + ); + } finally { + await client.close(); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } +}, 30_000); + function waitForSignal( timeoutMs: number, setup: (resolve: (value: T) => void, reject: (error: Error) => void) => () => void, diff --git a/packages/server/src/server/loop-service.test.ts b/packages/server/src/server/loop-service.test.ts index 67edfd173..37280535e 100644 --- a/packages/server/src/server/loop-service.test.ts +++ b/packages/server/src/server/loop-service.test.ts @@ -1033,6 +1033,173 @@ describe("LoopService", () => { expect(finalLoop.logs.some((entry) => entry.text.includes("Stop requested"))).toBe(true); }); + test("force-closes a loop worker when graceful cancellation is refused", async () => { + let release: (() => void) | null = null; + const blocker = new Promise((resolve) => { + release = resolve; + }); + const cancelledAgentIds: string[] = []; + const closedAgentIds: string[] = []; + const manager = new AgentManager({ + clients: { + claude: new ScriptedAgentClient("claude", { + async onRun({ config }) { + if (config.title?.includes("worker")) { + await blocker; + return "finished"; + } + return '{"passed":true,"reason":"ok"}'; + }, + }), + }, + registry: storage, + logger, + }); + manager.cancelAgentRun = async (agentId) => { + cancelledAgentIds.push(agentId); + return { status: "refused" }; + }; + const closeAgent = manager.closeAgent.bind(manager); + manager.closeAgent = async (agentId) => { + closedAgentIds.push(agentId); + await closeAgent(agentId); + }; + const service = createLoopService({ + paseoHome, + agentManager: manager, + agentStorage: storage, + logger, + }); + await service.initialize(); + + const loop = await service.runLoop({ + prompt: "Wait forever", + cwd: workspaceDir, + model: "test-model", + verifyChecks: ["test -f never.txt"], + }); + const workerAgentId = await waitForActiveWorkerRun(service, manager, loop.id); + const stopPromise = service.stopLoop(loop.id); + let closeWaitError: unknown; + + try { + await waitForCancelledAgent(cancelledAgentIds, workerAgentId); + await waitForCancelledAgent(closedAgentIds, workerAgentId); + } catch (error) { + closeWaitError = error; + } finally { + release?.(); + } + + const stopped = await stopPromise; + if (closeWaitError) { + throw closeWaitError; + } + expect(stopped.status).toBe("stopped"); + expect(cancelledAgentIds).toEqual([workerAgentId]); + expect(closedAgentIds).toContain(workerAgentId); + }); + + test("tolerates a loop worker closing while graceful cancellation is refused", async () => { + let release: (() => void) | null = null; + const blocker = new Promise((resolve) => { + release = resolve; + }); + const cancelledAgentIds: string[] = []; + const manager = new AgentManager({ + clients: { + claude: new ScriptedAgentClient("claude", { + async onRun({ config }) { + if (config.title?.includes("worker")) { + await blocker; + return "finished"; + } + return '{"passed":true,"reason":"ok"}'; + }, + }), + }, + registry: storage, + logger, + }); + const closeAgent = manager.closeAgent.bind(manager); + manager.cancelAgentRun = async (agentId) => { + cancelledAgentIds.push(agentId); + await closeAgent(agentId); + return { status: "refused" }; + }; + const service = createLoopService({ + paseoHome, + agentManager: manager, + agentStorage: storage, + logger, + }); + await service.initialize(); + + const loop = await service.runLoop({ + prompt: "Finish while Stop is canceling", + cwd: workspaceDir, + model: "test-model", + verifyChecks: ["test -f never.txt"], + }); + const workerAgentId = await waitForActiveWorkerRun(service, manager, loop.id); + const stopPromise = service.stopLoop(loop.id); + + await waitForCancelledAgent(cancelledAgentIds, workerAgentId); + release?.(); + + await expect(stopPromise).resolves.toMatchObject({ status: "stopped" }); + }); + + test("reports unexpected loop worker cancellation errors", async () => { + let release: (() => void) | null = null; + const blocker = new Promise((resolve) => { + release = resolve; + }); + const manager = new AgentManager({ + clients: { + claude: new ScriptedAgentClient("claude", { + async onRun({ config }) { + if (config.title?.includes("worker")) { + await blocker; + return "finished"; + } + return '{"passed":true,"reason":"ok"}'; + }, + }), + }, + registry: storage, + logger, + }); + manager.cancelAgentRun = async () => { + throw new Error("cancellation transport failed"); + }; + const service = createLoopService({ + paseoHome, + agentManager: manager, + agentStorage: storage, + logger, + }); + await service.initialize(); + + const loop = await service.runLoop({ + prompt: "Fail while Stop is canceling", + cwd: workspaceDir, + model: "test-model", + verifyChecks: ["test -f never.txt"], + }); + await waitForActiveWorkerRun(service, manager, loop.id); + const execution = ( + service as unknown as { running: Map }> } + ).running.get(loop.id)?.promise; + + try { + await expect(service.stopLoop(loop.id)).rejects.toThrow("cancellation transport failed"); + } finally { + release?.(); + await execution; + } + }); + test("stops while waiting for loop workspace provisioning without starting a worker", async () => { let resolveWorkspace: ((workspaceId: string) => void) | null = null; const workspaceProvisioned = new Promise((resolve) => { diff --git a/packages/server/src/server/loop-service.ts b/packages/server/src/server/loop-service.ts index e8e24b93f..99e5f8ff3 100644 --- a/packages/server/src/server/loop-service.ts +++ b/packages/server/src/server/loop-service.ts @@ -10,7 +10,7 @@ import { type EnsureWorkspaceForCreate, formatProviderModel, } from "./agent/create-agent/create.js"; -import type { AgentManager } from "./agent/agent-manager.js"; +import type { AgentManager, AgentRunCancellationResult } from "./agent/agent-manager.js"; import { buildStructuredAgentResponsePrompt, getStructuredAgentResponse, @@ -232,9 +232,19 @@ function buildVerifierTitle(loop: LoopRecord, iterationIndex: number): string { return `${prefix} [loop ${iterationIndex} verifier]`; } +function isUnknownLoopAgentError(error: unknown, agentId: string): boolean { + return error instanceof Error && error.message === `Unknown agent '${agentId}'`; +} + type LoopAgentManager = Pick< AgentManager, - "archiveAgent" | "cancelAgentRun" | "closeAgent" | "runAgent" | "subscribe" | "waitForAgentEvent" + | "archiveAgent" + | "cancelAgentRun" + | "closeAgent" + | "getAgent" + | "runAgent" + | "subscribe" + | "waitForAgentEvent" >; interface LoopExecutionContext { @@ -523,10 +533,10 @@ export class LoopService { if (running) { if (loop.activeWorkerAgentId) { - await this.options.agentManager.cancelAgentRun(loop.activeWorkerAgentId).catch(() => {}); + await this.stopInternalAgent(loop.activeWorkerAgentId, loop.archive); } if (loop.activeVerifierAgentId) { - await this.options.agentManager.cancelAgentRun(loop.activeVerifierAgentId).catch(() => {}); + await this.stopInternalAgent(loop.activeVerifierAgentId, loop.archive); } await running.promise.catch(() => {}); } else { @@ -539,6 +549,35 @@ export class LoopService { return cloneLoop(loop); } + private async stopInternalAgent(agentId: string, archive: boolean): Promise { + let cancellation: AgentRunCancellationResult; + try { + cancellation = await this.options.agentManager.cancelAgentRun(agentId); + } catch (error) { + if (isUnknownLoopAgentError(error, agentId)) { + return; + } + throw error; + } + if (cancellation.status !== "refused") { + return; + } + if (!this.options.agentManager.getAgent(agentId)) { + return; + } + try { + if (archive) { + await this.options.agentManager.archiveAgent(agentId); + return; + } + await this.options.agentManager.closeAgent(agentId); + } catch (error) { + if (!isUnknownLoopAgentError(error, agentId)) { + throw error; + } + } + } + private async executeLoop(loopId: string, signal: AbortSignal): Promise { const loop = this.requireLoop(loopId); const deadline = loop.maxTimeMs ? Date.now() + loop.maxTimeMs : null; diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index d85ce7f2b..ba7fabe7e 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -43,6 +43,7 @@ import type { } from "../services/github-service.js"; interface SessionHandlerInternals { + interruptAgentIfRunning(agentId: string): Promise; handleSendAgentMessage( agentId: string, text: string, @@ -89,6 +90,85 @@ function createBinaryMessageHandler( }; } +test("interruptAgentIfRunning rejects when graceful cancellation is refused", async () => { + const agentId = "11111111-1111-4111-8111-111111111111"; + const session = createSessionForTest({ + agentManager: { + getAgent: vi.fn(() => ({ id: agentId, provider: "codex", lifecycle: "running" })), + hasInFlightRun: vi.fn(() => true), + cancelAgentRun: vi.fn(async () => ({ status: "refused" as const })), + }, + }); + + await expect(asSessionInternals(session).interruptAgentIfRunning(agentId)).rejects.toThrow( + "active run cancellation was not acknowledged", + ); +}); + +test("cancel_agent_request reports refusal only through its response", async () => { + const agentId = "11111111-1111-4111-8111-111111111111"; + const messages: SessionOutboundMessage[] = []; + const getAgent = vi + .fn() + .mockReturnValueOnce({ id: agentId, provider: "codex", lifecycle: "running" }) + .mockReturnValue(null); + const session = createSessionForTest({ + messages, + agentManager: { + getAgent, + hasInFlightRun: vi.fn(() => true), + cancelAgentRun: vi.fn(async () => ({ status: "refused" as const })), + }, + }); + + await session.handleMessage({ + type: "cancel_agent_request", + agentId, + requestId: "cancel-refused", + }); + + expect(messages).toEqual([ + { + type: "cancel_agent_response", + payload: { + requestId: "cancel-refused", + agentId, + agent: null, + error: + "Cannot stop agent 11111111-1111-4111-8111-111111111111 because its active run cancellation was not acknowledged", + }, + }, + ]); +}); + +test("legacy cancel_agent_request reports refusal through the activity log", async () => { + const agentId = "11111111-1111-4111-8111-111111111111"; + const messages: SessionOutboundMessage[] = []; + const session = createSessionForTest({ + messages, + agentManager: { + getAgent: vi.fn(() => ({ id: agentId, provider: "codex", lifecycle: "running" })), + hasInFlightRun: vi.fn(() => true), + cancelAgentRun: vi.fn(async () => ({ status: "refused" as const })), + }, + }); + + await session.handleMessage({ type: "cancel_agent_request", agentId }); + + expect(messages).toEqual([ + { + type: "activity_log", + payload: { + id: expect.any(String), + timestamp: expect.any(Date), + type: "error", + content: + "Failed to cancel running agent on request: Cannot stop agent 11111111-1111-4111-8111-111111111111 because its active run cancellation was not acknowledged", + }, + }, + ]); +}); + const checkoutGitMocks = vi.hoisted(() => ({ checkoutResolvedBranch: vi.fn(), commitChanges: vi.fn(), diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index cc38733b2..69df4cd15 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -66,7 +66,7 @@ import { normalizeClientRestartRpcReason, } from "./lifecycle-reasons.js"; -import { AgentManager } from "./agent/agent-manager.js"; +import { AgentManager, AgentRunCancellationError } from "./agent/agent-manager.js"; import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import type { AgentManagerEvent, @@ -1085,16 +1085,17 @@ export class Session { ); const t0 = Date.now(); - const cancelled = await this.agentManager.cancelAgentRun(agentId); + const cancellation = await this.agentManager.cancelAgentRun(agentId); this.sessionLogger.debug( - { agentId, cancelled, durationMs: Date.now() - t0 }, + { agentId, cancellation: cancellation.status, durationMs: Date.now() - t0 }, "interruptAgentIfRunning: cancelAgentRun completed", ); - if (!cancelled) { + if (cancellation.status === "refused") { this.sessionLogger.warn( { agentId }, "interruptAgentIfRunning: reported running but no active run was cancelled", ); + throw new AgentRunCancellationError(agentId, "stop"); } } @@ -2857,11 +2858,30 @@ export class Session { requestId, agentId, agent: payload, + error: null, }, }); } } catch (error) { - this.handleAgentRunError(agentId, error, "Failed to cancel running agent on request"); + if (requestId) { + this.sessionLogger.error( + { err: error, agentId }, + `Failed to cancel running agent on request for agent ${agentId}`, + ); + const agent = this.agentManager.getAgent(agentId); + const payload = agent ? await this.buildAgentPayload(agent) : null; + this.emit({ + type: "cancel_agent_response", + payload: { + requestId, + agentId, + agent: payload, + error: errorToFriendlyMessage(error), + }, + }); + } else { + this.handleAgentRunError(agentId, error, "Failed to cancel running agent on request"); + } } } diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 5475a5854..f7aefff5c 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -1609,7 +1609,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy archivedAt: null, }; const killTerminal = vi.fn(); - const cancelAgentRun = vi.fn(async () => true); + const cancelAgentRun = vi.fn(async () => ({ status: "settled" as const })); const session = asTestSession( new Session({ clientId: "test-client", diff --git a/packages/server/src/server/session/voice/voice-session.test.ts b/packages/server/src/server/session/voice/voice-session.test.ts index 896546867..d6232ff4d 100644 --- a/packages/server/src/server/session/voice/voice-session.test.ts +++ b/packages/server/src/server/session/voice/voice-session.test.ts @@ -110,6 +110,32 @@ async function settle(): Promise { } describe("VoiceSession streaming transcription", () => { + test("surfaces a refused voice-mode agent interruption", async () => { + const { voiceSession, host } = createVoiceSession(); + host.interruptAgentIfRunning = vi.fn(async () => { + throw new Error("active run cancellation was not acknowledged"); + }); + + await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID); + + await expect(voiceSession.handleAbort()).rejects.toThrow( + "active run cancellation was not acknowledged", + ); + expect(host.interruptAgentIfRunning).toHaveBeenCalledWith(VOICE_AGENT_ID); + expect(host.emitted).toContainEqual( + expect.objectContaining({ + type: "activity_log", + payload: expect.objectContaining({ + type: "error", + content: "Voice interruption failed: active run cancellation was not acknowledged", + metadata: { voiceAbortFailed: true }, + }), + }), + ); + + await voiceSession.cleanup(); + }); + test("delivers the streaming final transcript to the agent exactly once", async () => { const { voiceSession, detector, sttSession, host } = createVoiceSession(); diff --git a/packages/server/src/server/session/voice/voice-session.ts b/packages/server/src/server/session/voice/voice-session.ts index 2ac9998ed..f6331dba3 100644 --- a/packages/server/src/server/session/voice/voice-session.ts +++ b/packages/server/src/server/session/voice/voice-session.ts @@ -1084,10 +1084,18 @@ export class VoiceSession { try { await this.host.interruptAgentIfRunning(this.voiceModeAgentId); } catch (error) { - this.sessionLogger.warn( - { err: error, agentId: this.voiceModeAgentId }, - "Failed to interrupt active voice-mode agent on abort", - ); + const message = `Voice interruption failed: ${getErrorMessage(error)}`; + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "error", + content: message, + metadata: { voiceAbortFailed: true }, + }, + }); + throw error; } }