puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix hidden Codex subagents and stuck parent sessions (#2068)

Browse Source 
* fix(server): prevent hidden Codex subagents and stuck turns

Codex can announce native children only through its mirrored lifecycle stream, while rejected interrupts previously looked like successful local cancellation. Preserve those child announcements and keep manager state active until the provider acknowledges cancellation.

* fix(server): block actions after rejected cancellation

A provider can accept a turn before publishing the turn ID needed to interrupt it. Keep that interval non-cancelable, and prevent reload, replacement, or rewind from proceeding without an acknowledged cancellation.

* fix(server): surface rejected agent cancellations

Replacement prompts and Stop requests could appear accepted after the provider kept ownership of the active turn. Complete replacement cancellation before detaching the stream, and return cancellation failures through the client response.

* fix(server): handle turn completion during cancellation

* fix: finish cancellation lifecycle handling

* fix(server): close remaining cancellation races

* fix(server): settle autonomous cancellations

* fix: honor cancellation failures at call sites

* refactor(server): centralize agent run state

* fix(server): keep pending runs isolated from stale events

* fix: surface remaining cancellation failures

* fix(server): narrow cancellation error handling
This commit is contained in:
Mohamed Boudra
2026-07-15 00:01:58 +02:00
committed by GitHub
parent 50ed0d0ab1
commit d706c4339b
33 changed files with 1597 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/agent-lifecycle.md
View File

@@ -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:

20
packages/app/src/composer/actions.test.ts
View File

@@ -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 });

8
packages/app/src/composer/actions.ts
View File

@@ -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;
}

11
packages/app/src/composer/index.tsx
View File

@@ -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);

50
packages/app/src/composer/input/input.tsx
View File

@@ -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<unknown> } | null | undefined;
isRealtimeVoiceForCurrentAgent: boolean;
isAgentRunning: boolean;
client: { cancelAgent: (agentId: string) => Promise<unknown> } | null;
voiceAgentId: string | undefined;
}
async function stopRealtimeVoiceImpl(ctx: StopRealtimeVoiceContext): Promise<void> {
if (!ctx.voice || !ctx.isRealtimeVoiceForCurrentAgent) return;
const tasks: Promise<unknown>[] = [];
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<MessageInputRef, MessageInputProps>(
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({

51
packages/app/src/composer/input/state.test.ts
View File

@@ -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"]);
});
});

21
packages/app/src/composer/input/state.ts
View File

@@ -3,6 +3,14 @@ import type { MessagePayload } from "@/composer/types";
export type SendBehavior = "interrupt" | "queue";
interface StopRealtimeVoiceContext {
voice: { stopVoice: () => Promise<unknown> } | null | undefined;
isRealtimeVoiceForCurrentAgent: boolean;
isAgentRunning: boolean;
client: { cancelAgent: (agentId: string) => Promise<unknown> } | 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<void> {
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();
}

12
packages/app/src/contexts/session-context.tsx
View File

@@ -476,6 +476,15 @@ function applyToolErrorToMessages(
);
}
function notifyVoiceAbortFailure(
data: Extract<SessionOutboundMessage, { type: "activity_log" }>["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,
]);

39
packages/cli/src/commands/agent/delete.test.ts Normal file
View File

@@ -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],
});
});
});

2
packages/cli/src/commands/agent/delete.ts
View File

@@ -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 };

5
packages/client/src/daemon-client.ts
View File

@@ -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<AgentProviderNotice | null> {

1
packages/protocol/src/messages.ts
View File

@@ -3139,6 +3139,7 @@ export const CancelAgentResponseMessageSchema = z.object({
requestId: z.string(),
agentId: z.string(),
agent: AgentSnapshotPayloadSchema.nullable(),
error: z.string().nullable().optional(),
}),
});

381
packages/server/src/server/agent/agent-manager.test.ts
View File

@@ -54,6 +54,28 @@ function deferred<T>(): Deferred<T> {
return { promise, resolve, reject };
}
function waitForAgentLifecycle(
manager: AgentManager,
agentId: string,
lifecycle: ManagedAgent["lifecycle"],
): Promise<void> {
return new Promise<void>((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<void> {}
}
class ControlledInterruptSession extends TestAgentSession {
interruptCalled = false;
constructor(
config: AgentSessionConfig,
readonly turnId: string,
private readonly interruptBehavior: (session: ControlledInterruptSession) => Promise<void>,
) {
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<void> {
this.interruptCalled = true;
await this.interruptBehavior(this);
}
}
interface ControlledInterruptFixture {
agentId: string;
manager: AgentManager;
session: ControlledInterruptSession;
startForegroundRun(): Promise<void>;
cleanup(): void;
}
async function createControlledInterruptFixture(options: {
name: string;
agentId: string;
turnId: string;
interrupt: (session: ControlledInterruptSession) => Promise<void>;
}): Promise<ControlledInterruptFixture> {
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<AgentSession> {
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<void>();
private readonly runtimeInfoAllowed = deferred<void>();
@@ -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<void> {
this.interruptCalled = true;
await new Promise(() => {});
}
}
class HangingInterruptClient extends TestAgentClient {
readonly session = new HangingInterruptSession({
provider: "codex",
cwd: workdir,
});
override async createSession(): Promise<AgentSession> {
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<void>((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<void>();
override async interrupt(): Promise<void> {
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<void> {
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<AgentSession> {
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) {

249
packages/server/src/server/agent/agent-manager.ts
View File

@@ -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<string>();
private readonly sessionEventTails = new Map<string, Promise<void>>();
private readonly foregroundRuns = new ForegroundRunState();
private readonly runs = new AgentRunState();
private readonly subscribers = new Set<SubscriptionRecord>();
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<ForegroundRunState["createTurnStream"]> | null = null;
let turnStream: ReturnType<AgentRunState["createTurnStream"]> | 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<AgentStreamEvent> {
): Promise<AsyncGenerator<AgentStreamEvent>> {
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<void> {
@@ -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<boolean> {
async cancelAgentRun(agentId: string): Promise<AgentRunCancellationResult> {
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<void>((resolvePromise) => setTimeout(resolvePromise, 2000));
if (waiter) {
await Promise.race([waiter.settledPromise, timeout]);
} else if (agent.activeForegroundTurnId === foregroundTurnId) {
await Promise.race([
new Promise<void>((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<void>((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<void> {
private async cancelAgentRunBefore(
agentId: string,
action: "reload" | "replace" | "rewind",
): Promise<void> {
const result = await this.cancelAgentRun(agentId);
if (result.status === "refused") {
throw new AgentRunCancellationError(agentId, action);
}
}
private async interruptSession(session: AgentSession, agentId: string): Promise<boolean> {
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<void> {
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);
}

57
packages/server/src/server/agent/agent-prompt.test.ts
View File

@@ -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<Record<string, unknown>>;
nextRecord: Promise<void>;
}
function createCapturedLogger(): CapturedLogger {
const records: Array<Record<string, unknown>> = [];
let resolveNextRecord!: () => void;
const nextRecord = new Promise<void>((resolve) => {
resolveNextRecord = resolve;
});
const logger = pino(
{ level: "error" },
{
write(line: string) {
records.push(JSON.parse(line) as Record<string, unknown>);
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;

27
packages/server/src/server/agent/agent-prompt.ts
View File

@@ -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");
}
}

120
packages/server/src/server/agent/foreground-run-state.ts → packages/server/src/server/agent/agent-run-state.ts
View File

@@ -12,45 +12,102 @@ export interface ForegroundTurnWaiter {
export interface PendingForegroundRun {
token: string;
kind: "foreground";
turnId: string | null;
started: boolean;
settled: boolean;
settledPromise: Promise<void>;
resolveSettled: () => void;
}
export interface AutonomousAgentRun {
token: string;
kind: "autonomous";
turnId: string | null;
started: true;
settled: boolean;
settledPromise: Promise<void>;
resolveSettled: () => void;
}
export type TrackedAgentRun = PendingForegroundRun | AutonomousAgentRun;
export interface ForegroundRunAgentState {
foregroundTurnWaiters: Set<ForegroundTurnWaiter>;
finalizedForegroundTurnIds: Set<string>;
}
export class ForegroundRunState {
private readonly pendingRuns = new Map<string, PendingForegroundRun>();
export class AgentRunState {
private readonly runs = new Map<string, TrackedAgentRun>();
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<void>((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();
}

45
packages/server/src/server/agent/lifecycle-command.test.ts
View File

@@ -43,6 +43,8 @@ class FakeLifecycleAgentManager implements LifecycleAgentManager {
readonly modeUpdates: Array<{ agentId: string; modeId: string }> = [];
readonly detachedAgentIds: string[] = [];
inFlightAgentIds = new Set<string>();
readonly settledDuringCancellationAgentIds = new Set<string>();
readonly rejectedCancellationAgentIds = new Set<string>();
constructor(private readonly storage: FakeLifecycleAgentStorage) {}
@@ -54,9 +56,18 @@ class FakeLifecycleAgentManager implements LifecycleAgentManager {
return this.inFlightAgentIds.has(agentId);
}
async cancelAgentRun(agentId: string): Promise<boolean> {
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<void> {
@@ -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);

46
packages/server/src/server/agent/lifecycle-command.ts
View File

@@ -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<ManagedAgent, "id" | "cwd" | "lifecycl
export interface LifecycleAgentManager {
getAgent(agentId: string): LifecycleAgentSnapshot | null;
hasInFlightRun(agentId: string): boolean;
cancelAgentRun(agentId: string): Promise<boolean>;
cancelAgentRun(agentId: string): Promise<AgentRunCancellationResult>;
clearAgentAttention(agentId: string): Promise<void>;
archiveAgent(agentId: string): Promise<{ archivedAt: string }>;
archiveSnapshot(agentId: string, archivedAt: string): Promise<StoredAgentRecord>;
@@ -47,10 +51,14 @@ export interface CancelAgentRunResult {
cancelled: boolean;
}
export async function cancelAgentRunCommand(
interface RequestedAgentRunCancellation extends CancelAgentRunResult {
cancellation: AgentRunCancellationResult;
}
async function requestAgentRunCancellation(
dependencies: Pick<AgentLifecycleCommandDependencies, "agentManager" | "logger">,
agentId: string,
): Promise<CancelAgentRunResult> {
): Promise<RequestedAgentRunCancellation> {
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<AgentLifecycleCommandDependencies, "agentManager" | "logger">,
agentId: string,
): Promise<CancelAgentRunResult> {
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);

4
packages/server/src/server/agent/permission-response.test.ts
View File

@@ -53,11 +53,11 @@ class FakePermissionAgentManager {
return emptyAgentStream();
}
replaceAgentRun(
async replaceAgentRun(
agentId: string,
prompt: AgentPromptInput,
options?: AgentRunOptions,
): AsyncGenerator<AgentStreamEvent> {
): Promise<AsyncGenerator<AgentStreamEvent>> {
this.replacementRuns.push({ agentId, prompt, options });
return emptyAgentStream();
}

2
packages/server/src/server/agent/permission-response.ts
View File

@@ -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,
});
}

153
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
View File

@@ -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();

39
packages/server/src/server/agent/providers/codex-app-server-agent.ts
View File

@@ -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<void> {
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<void> {
@@ -5528,9 +5534,10 @@ export class CodexAppServerAgentSession implements AgentSession {
parsed: Extract<ParsedCodexNotification, { kind: "item_completed" }>,
): 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<ParsedCodexNotification, { kind: "item_started" }>,
): void {
if (parsed.source === "codex_event") {
if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) {
return;
}
if (this.isUserMessageItem(parsed.item)) {

26
packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts
View File

@@ -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<TType extends StreamEventType>(
}
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<ProviderSubagentEvent> {
return waitForNextEvent(session, "provider_subagent", (event) => event.event.id === id);
}

29
packages/server/src/server/agent/rewind/rewind.test.ts
View File

@@ -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<void> {
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();

259
packages/server/src/server/daemon-client.e2e.test.ts
View File

@@ -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<AgentRunResult> {
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<AgentStreamEvent> {}
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<string> {
return "full-access";
}
async setMode(): Promise<void> {}
getPendingPermissions() {
return [];
}
async respondToPermission(): Promise<void> {}
describePersistence(): AgentPersistenceHandle {
return { provider: this.provider, sessionId: this.id };
}
async interrupt(): Promise<void> {
if (this.options.interruptError) throw new Error(this.options.interruptError);
}
async close(): Promise<void> {
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<boolean> {
return true;
}
async createSession(): Promise<AgentSession> {
return new StubAgentSession(this.options);
}
async resumeSession(): Promise<AgentSession> {
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<AgentRunResult> {
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<AgentStreamEvent> {
yield* [];
}
async getRuntimeInfo() {
return {
provider: "codex" as const,
sessionId: this.id,
model: "gpt-5.4-mini",
modeId: "full-access",
};
}
async getAvailableModes(): Promise<Array<{ id: string; label: string; description: string }>> {
return [{ id: "full-access", label: "Full access", description: "No prompts" }];
}
async getCurrentMode(): Promise<string | null> {
return "full-access";
}
async setMode(): Promise<void> {}
getPendingPermissions() {
return [];
}
async respondToPermission(): Promise<void> {}
describePersistence(): AgentPersistenceHandle | null {
return { provider: "codex", sessionId: this.id };
}
async interrupt(): Promise<void> {}
async close(): Promise<void> {}
}
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<boolean> {
return true;
}
async createSession(_config: AgentSessionConfig): Promise<AgentSession> {
return new StartTurnFailureSession();
}
async resumeSession(): Promise<AgentSession> {
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<T>(
timeoutMs: number,
setup: (resolve: (value: T) => void, reject: (error: Error) => void) => () => void,

167
packages/server/src/server/loop-service.test.ts
View File

@@ -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<void>((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<void>((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<void>((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<string, { promise: Promise<void> }> }
).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<string>((resolve) => {

47
packages/server/src/server/loop-service.ts
View File

@@ -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<void> {
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<void> {
const loop = this.requireLoop(loopId);
const deadline = loop.maxTimeMs ? Date.now() + loop.maxTimeMs : null;

80
packages/server/src/server/session.test.ts
View File

@@ -43,6 +43,7 @@ import type {
} from "../services/github-service.js";
interface SessionHandlerInternals {
interruptAgentIfRunning(agentId: string): Promise<void>;
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(),

30
packages/server/src/server/session.ts
View File

@@ -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");
}
}
}

2
packages/server/src/server/session.workspaces.test.ts
View File

@@ -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",

26
packages/server/src/server/session/voice/voice-session.test.ts
View File

@@ -110,6 +110,32 @@ async function settle(): Promise<void> {
}
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();

16
packages/server/src/server/session/voice/voice-session.ts
View File

@@ -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;
}
}