Fail active agents when their provider process exits (#2757)
* fix(server): fail agents when provider process exits The transport previously closed pending RPCs without terminating the active manager run, leaving the agent stuck in running state. Propagate unexpected process termination as a one-shot turn failure while suppressing it during intentional session close. * fix(server): keep idle provider exits recoverable A process death with no active foreground turn is not a failed turn and must not trigger turn-based lifecycle actions. Leave the session disconnected so the next prompt reconnects without publishing a false terminal event. * fix(server): preserve provider exit recovery Connection cleanup after a failed reconnect must not close the durable session boundary or discard AgentManager subscribers. Treat both foreground and autonomous root turns as active when propagating process death. * test(server): make process exit harness portable * fix(server): clear permissions after provider exit * fix(server): preserve plan approval after provider exit * fix(server): serialize provider reconnects * fix(server): stop reconnects after session close
This commit is contained in:
+547
@@ -0,0 +1,547 @@
|
|||||||
|
import { mkdtempSync, rmSync } from "node:fs";
|
||||||
|
import { tmpdir } from "node:os";
|
||||||
|
import { join } from "node:path";
|
||||||
|
|
||||||
|
import { expect, test } from "vitest";
|
||||||
|
|
||||||
|
import { createTestLogger } from "../../../test-utils/test-logger.js";
|
||||||
|
import { AgentManager, type AgentManagerEvent } from "../agent-manager.js";
|
||||||
|
import type {
|
||||||
|
AgentClient,
|
||||||
|
AgentLaunchContext,
|
||||||
|
AgentSession,
|
||||||
|
AgentSessionConfig,
|
||||||
|
} from "../agent-sdk-types.js";
|
||||||
|
import { CodexAppServerAgentClient, CodexAppServerAgentSession } from "./codex-app-server-agent.js";
|
||||||
|
import {
|
||||||
|
createFakeCodexAppServer,
|
||||||
|
type FakeCodexAppServer,
|
||||||
|
} from "./codex/test-utils/fake-app-server.js";
|
||||||
|
|
||||||
|
const logger = createTestLogger();
|
||||||
|
|
||||||
|
class ProcessExitCodexClient extends CodexAppServerAgentClient implements AgentClient {
|
||||||
|
constructor(private readonly appServers: FakeCodexAppServer[]) {
|
||||||
|
super(logger);
|
||||||
|
}
|
||||||
|
|
||||||
|
override async isAvailable(): Promise<boolean> {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
override async createSession(
|
||||||
|
config: AgentSessionConfig,
|
||||||
|
launchContext?: AgentLaunchContext,
|
||||||
|
): Promise<AgentSession> {
|
||||||
|
const session = new CodexAppServerAgentSession(
|
||||||
|
config,
|
||||||
|
null,
|
||||||
|
logger,
|
||||||
|
async () => {
|
||||||
|
const appServer = this.appServers.shift();
|
||||||
|
if (!appServer) {
|
||||||
|
throw new Error("No fake Codex app-server available");
|
||||||
|
}
|
||||||
|
return appServer.child;
|
||||||
|
},
|
||||||
|
{},
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
launchContext?.agentId,
|
||||||
|
);
|
||||||
|
await session.connect();
|
||||||
|
return session;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
test("unexpected Codex app-server exit fails the active run and agent", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-exit-"));
|
||||||
|
const appServer = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: { codex: new ProcessExitCodexClient([appServer]) },
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
const events: AgentManagerEvent[] = [];
|
||||||
|
const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false });
|
||||||
|
let agentId: string | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
agentId = agent.id;
|
||||||
|
const run = manager.runAgent(agent.id, "keep working");
|
||||||
|
const turnStart = await appServer.waitForTurnStart();
|
||||||
|
appServer.startsTurn({ threadId: String(turnStart.threadId), turnId: "native-turn-1" });
|
||||||
|
await manager.waitForAgentRunStart(agent.id);
|
||||||
|
|
||||||
|
appServer.child.stderr.write("provider crashed");
|
||||||
|
appServer.child.emit("exit", 17, null);
|
||||||
|
appServer.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
await expect(run).rejects.toThrow(
|
||||||
|
"Codex app-server exited with code 17 and signal null\nprovider crashed",
|
||||||
|
);
|
||||||
|
await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("error");
|
||||||
|
expect(manager.getAgent(agent.id)?.lastError).toBe(
|
||||||
|
"Codex app-server exited with code 17 and signal null\nprovider crashed",
|
||||||
|
);
|
||||||
|
expect(events).toContainEqual(
|
||||||
|
expect.objectContaining({
|
||||||
|
type: "agent_stream",
|
||||||
|
agentId: agent.id,
|
||||||
|
event: expect.objectContaining({
|
||||||
|
type: "turn_failed",
|
||||||
|
error: "Codex app-server exited with code 17 and signal null\nprovider crashed",
|
||||||
|
}),
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"),
|
||||||
|
).toHaveLength(1);
|
||||||
|
} finally {
|
||||||
|
if (agentId && manager.getAgent(agentId)) {
|
||||||
|
await manager.closeAgent(agentId);
|
||||||
|
}
|
||||||
|
unsubscribe();
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("intentional agent close stays clean while a Codex turn is active", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-close-"));
|
||||||
|
const appServer = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: { codex: new ProcessExitCodexClient([appServer]) },
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
const events: AgentManagerEvent[] = [];
|
||||||
|
const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
const run = manager.streamAgent(agent.id, "keep working");
|
||||||
|
const drainRun = (async () => {
|
||||||
|
const runEvents = [];
|
||||||
|
for await (const event of run) {
|
||||||
|
runEvents.push(event);
|
||||||
|
}
|
||||||
|
return runEvents;
|
||||||
|
})();
|
||||||
|
const turnStart = await appServer.waitForTurnStart();
|
||||||
|
appServer.startsTurn({ threadId: String(turnStart.threadId), turnId: "native-turn-1" });
|
||||||
|
await manager.waitForAgentRunStart(agent.id);
|
||||||
|
|
||||||
|
await manager.closeAgent(agent.id);
|
||||||
|
|
||||||
|
await expect(drainRun).resolves.toContainEqual(
|
||||||
|
expect.objectContaining({ type: "turn_canceled", reason: "agent closed" }),
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"),
|
||||||
|
).toHaveLength(0);
|
||||||
|
expect(events).toContainEqual(
|
||||||
|
expect.objectContaining({
|
||||||
|
type: "agent_state",
|
||||||
|
agent: expect.objectContaining({ id: agent.id, lifecycle: "closed" }),
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
} finally {
|
||||||
|
unsubscribe();
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("idle Codex app-server exit reconnects without publishing a failed turn", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-idle-exit-"));
|
||||||
|
const exitedAppServer = createFakeCodexAppServer();
|
||||||
|
const replacementAppServer = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: {
|
||||||
|
codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]),
|
||||||
|
},
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
const events: AgentManagerEvent[] = [];
|
||||||
|
const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false });
|
||||||
|
let agentId: string | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
agentId = agent.id;
|
||||||
|
|
||||||
|
exitedAppServer.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||||
|
expect(
|
||||||
|
events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"),
|
||||||
|
).toHaveLength(0);
|
||||||
|
|
||||||
|
const run = manager.runAgent(agent.id, "continue after reconnect");
|
||||||
|
const turnStart = await replacementAppServer.waitForTurnStart();
|
||||||
|
replacementAppServer.startsTurn({
|
||||||
|
threadId: String(turnStart.threadId),
|
||||||
|
turnId: "native-turn-2",
|
||||||
|
});
|
||||||
|
replacementAppServer.completeTurn({ threadId: String(turnStart.threadId) });
|
||||||
|
|
||||||
|
await expect(run).resolves.toMatchObject({ sessionId: "thread-1" });
|
||||||
|
expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||||
|
} finally {
|
||||||
|
if (agentId && manager.getAgent(agentId)) {
|
||||||
|
await manager.closeAgent(agentId);
|
||||||
|
}
|
||||||
|
unsubscribe();
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("failed reconnect preserves manager events for a later successful run", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-reconnect-failure-"));
|
||||||
|
const exitedAppServer = createFakeCodexAppServer();
|
||||||
|
const failedReconnect = createFakeCodexAppServer({
|
||||||
|
initialize: async () => {
|
||||||
|
throw new Error("reconnect failed");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
const successfulReconnect = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: {
|
||||||
|
codex: new ProcessExitCodexClient([exitedAppServer, failedReconnect, successfulReconnect]),
|
||||||
|
},
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
let agentId: string | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
agentId = agent.id;
|
||||||
|
exitedAppServer.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
await expect(manager.runAgent(agent.id, "first reconnect")).rejects.toThrow("reconnect failed");
|
||||||
|
|
||||||
|
const run = manager.runAgent(agent.id, "second reconnect");
|
||||||
|
const turnStart = await successfulReconnect.waitForTurnStart();
|
||||||
|
successfulReconnect.startsTurn({
|
||||||
|
threadId: String(turnStart.threadId),
|
||||||
|
turnId: "native-turn-3",
|
||||||
|
});
|
||||||
|
successfulReconnect.completeTurn({ threadId: String(turnStart.threadId) });
|
||||||
|
|
||||||
|
await expect(run).resolves.toMatchObject({ sessionId: "thread-1" });
|
||||||
|
expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||||
|
} finally {
|
||||||
|
if (agentId && manager.getAgent(agentId)) {
|
||||||
|
await manager.closeAgent(agentId);
|
||||||
|
}
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("unexpected exit fails an autonomous Codex turn", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-autonomous-exit-"));
|
||||||
|
const appServer = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: { codex: new ProcessExitCodexClient([appServer]) },
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
const events: AgentManagerEvent[] = [];
|
||||||
|
const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false });
|
||||||
|
let agentId: string | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
agentId = agent.id;
|
||||||
|
appServer.startsTurn({ threadId: "thread-1", turnId: "autonomous-turn" });
|
||||||
|
await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("running");
|
||||||
|
|
||||||
|
appServer.child.stderr.write("autonomous provider crashed");
|
||||||
|
appServer.child.emit("exit", 23, null);
|
||||||
|
|
||||||
|
await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("error");
|
||||||
|
expect(manager.getAgent(agent.id)?.lastError).toBe(
|
||||||
|
"Codex app-server exited with code 23 and signal null\nautonomous provider crashed",
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"),
|
||||||
|
).toHaveLength(1);
|
||||||
|
} finally {
|
||||||
|
if (agentId && manager.getAgent(agentId)) {
|
||||||
|
await manager.closeAgent(agentId);
|
||||||
|
}
|
||||||
|
unsubscribe();
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("provider permissions do not survive an unexpected exit and reconnect", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-permission-exit-"));
|
||||||
|
const exitedAppServer = createFakeCodexAppServer();
|
||||||
|
const replacementAppServer = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: {
|
||||||
|
codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]),
|
||||||
|
},
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
let agentId: string | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
agentId = agent.id;
|
||||||
|
|
||||||
|
const failedRun = manager.runAgent(agent.id, "run a command");
|
||||||
|
const failedTurnStart = await exitedAppServer.waitForTurnStart();
|
||||||
|
const failedThreadId = String(failedTurnStart.threadId);
|
||||||
|
exitedAppServer.startsTurn({ threadId: failedThreadId, turnId: "native-turn-1" });
|
||||||
|
exitedAppServer.requestCommandApproval({
|
||||||
|
itemId: "stale-command",
|
||||||
|
threadId: failedThreadId,
|
||||||
|
turnId: "native-turn-1",
|
||||||
|
command: "git status",
|
||||||
|
cwd: workdir,
|
||||||
|
reason: "confirm command",
|
||||||
|
});
|
||||||
|
await expect
|
||||||
|
.poll(() => manager.getPendingPermissions(agent.id).map((request) => request.id))
|
||||||
|
.toEqual(["permission-stale-command"]);
|
||||||
|
|
||||||
|
exitedAppServer.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
await expect(failedRun).rejects.toThrow("Codex app-server exited with code 17");
|
||||||
|
expect(manager.getPendingPermissions(agent.id)).toEqual([]);
|
||||||
|
|
||||||
|
const recoveredRun = manager.runAgent(agent.id, "continue after reconnect");
|
||||||
|
const recoveredTurnStart = await replacementAppServer.waitForTurnStart();
|
||||||
|
const recoveredThreadId = String(recoveredTurnStart.threadId);
|
||||||
|
replacementAppServer.startsTurn({
|
||||||
|
threadId: recoveredThreadId,
|
||||||
|
turnId: "native-turn-2",
|
||||||
|
});
|
||||||
|
replacementAppServer.requestCommandApproval({
|
||||||
|
itemId: "new-command",
|
||||||
|
threadId: recoveredThreadId,
|
||||||
|
turnId: "native-turn-2",
|
||||||
|
command: "git diff",
|
||||||
|
cwd: workdir,
|
||||||
|
reason: "confirm command",
|
||||||
|
});
|
||||||
|
await expect
|
||||||
|
.poll(() => manager.getPendingPermissions(agent.id).map((request) => request.id))
|
||||||
|
.toEqual(["permission-new-command"]);
|
||||||
|
|
||||||
|
await manager.respondToPermission(agent.id, "permission-new-command", {
|
||||||
|
behavior: "allow",
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(manager.getPendingPermissions(agent.id)).toEqual([]);
|
||||||
|
replacementAppServer.completeTurn({ threadId: recoveredThreadId });
|
||||||
|
await expect(recoveredRun).resolves.toMatchObject({ sessionId: "thread-1" });
|
||||||
|
expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||||
|
} finally {
|
||||||
|
if (agentId && manager.getAgent(agentId)) {
|
||||||
|
await manager.closeAgent(agentId);
|
||||||
|
}
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("idle provider exit preserves a synthetic plan approval across reconnect", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-plan-exit-"));
|
||||||
|
const exitedAppServer = createFakeCodexAppServer();
|
||||||
|
const replacementAppServer = createFakeCodexAppServer();
|
||||||
|
const manager = new AgentManager({
|
||||||
|
clients: {
|
||||||
|
codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]),
|
||||||
|
},
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
let agentId: string | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const agent = await manager.createAgent(
|
||||||
|
{
|
||||||
|
provider: "codex",
|
||||||
|
cwd: workdir,
|
||||||
|
modeId: "auto",
|
||||||
|
model: "gpt-5.4",
|
||||||
|
featureValues: { plan_mode: true },
|
||||||
|
},
|
||||||
|
undefined,
|
||||||
|
{ workspaceId: undefined },
|
||||||
|
);
|
||||||
|
agentId = agent.id;
|
||||||
|
|
||||||
|
const planRun = manager.runAgent(agent.id, "make a plan");
|
||||||
|
const planTurnStart = await exitedAppServer.waitForTurnStart();
|
||||||
|
const planThreadId = String(planTurnStart.threadId);
|
||||||
|
exitedAppServer.startsTurn({ threadId: planThreadId, turnId: "plan-turn" });
|
||||||
|
exitedAppServer.updatesPlan({
|
||||||
|
threadId: planThreadId,
|
||||||
|
steps: ["Inspect the provider lifecycle", "Implement the fix"],
|
||||||
|
});
|
||||||
|
exitedAppServer.completeTurn({ threadId: planThreadId });
|
||||||
|
await expect(planRun).resolves.toMatchObject({ sessionId: "thread-1" });
|
||||||
|
const [planApproval] = manager.getPendingPermissions(agent.id);
|
||||||
|
expect(planApproval).toMatchObject({
|
||||||
|
name: "CodexPlanApproval",
|
||||||
|
kind: "plan",
|
||||||
|
});
|
||||||
|
|
||||||
|
exitedAppServer.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||||
|
expect(manager.getPendingPermissions(agent.id)).toEqual([planApproval]);
|
||||||
|
const approval = await manager.respondToPermission(agent.id, planApproval!.id, {
|
||||||
|
behavior: "allow",
|
||||||
|
selectedActionId: "implement",
|
||||||
|
});
|
||||||
|
expect(approval).toMatchObject({
|
||||||
|
followUpPrompt: expect.stringContaining("Implement the fix"),
|
||||||
|
});
|
||||||
|
|
||||||
|
const implementationRun = manager.runAgent(agent.id, approval!.followUpPrompt!);
|
||||||
|
const implementationTurnStart = await replacementAppServer.waitForTurnStart();
|
||||||
|
const implementationThreadId = String(implementationTurnStart.threadId);
|
||||||
|
replacementAppServer.startsTurn({
|
||||||
|
threadId: implementationThreadId,
|
||||||
|
turnId: "implementation-turn",
|
||||||
|
});
|
||||||
|
replacementAppServer.completeTurn({ threadId: implementationThreadId });
|
||||||
|
await expect(implementationRun).resolves.toMatchObject({ sessionId: "thread-1" });
|
||||||
|
expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||||
|
} finally {
|
||||||
|
if (agentId && manager.getAgent(agentId)) {
|
||||||
|
await manager.closeAgent(agentId);
|
||||||
|
}
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("concurrent session APIs share one reconnect after an idle provider exit", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-concurrent-reconnect-"));
|
||||||
|
const appServers = [
|
||||||
|
createFakeCodexAppServer(),
|
||||||
|
createFakeCodexAppServer(),
|
||||||
|
createFakeCodexAppServer(),
|
||||||
|
];
|
||||||
|
let spawnCount = 0;
|
||||||
|
let releaseReconnect: (() => void) | undefined;
|
||||||
|
const reconnectGate = new Promise<void>((resolve) => {
|
||||||
|
releaseReconnect = resolve;
|
||||||
|
});
|
||||||
|
const session = new CodexAppServerAgentSession(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
null,
|
||||||
|
logger,
|
||||||
|
async () => {
|
||||||
|
const appServer = appServers[spawnCount];
|
||||||
|
if (!appServer) {
|
||||||
|
throw new Error("No fake Codex app-server available");
|
||||||
|
}
|
||||||
|
spawnCount += 1;
|
||||||
|
if (spawnCount > 1) {
|
||||||
|
await reconnectGate;
|
||||||
|
}
|
||||||
|
return appServer.child;
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
try {
|
||||||
|
await session.connect();
|
||||||
|
appServers[0]!.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
const runtimeInfo = session.getRuntimeInfo();
|
||||||
|
const turnStart = session.startTurn("continue after reconnect");
|
||||||
|
const reconnectSpawnCount = spawnCount - 1;
|
||||||
|
releaseReconnect?.();
|
||||||
|
|
||||||
|
await expect(runtimeInfo).resolves.toMatchObject({ provider: "codex" });
|
||||||
|
await expect(turnStart).resolves.toMatchObject({ turnId: expect.any(String) });
|
||||||
|
expect(reconnectSpawnCount).toBe(1);
|
||||||
|
} finally {
|
||||||
|
releaseReconnect?.();
|
||||||
|
await session.close();
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("session close disposes a provider that arrives from an in-flight reconnect", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "codex-process-close-reconnect-"));
|
||||||
|
const initialAppServer = createFakeCodexAppServer();
|
||||||
|
const lateAppServer = createFakeCodexAppServer();
|
||||||
|
let spawnCount = 0;
|
||||||
|
let markReconnectSpawned: (() => void) | undefined;
|
||||||
|
const reconnectSpawned = new Promise<void>((resolve) => {
|
||||||
|
markReconnectSpawned = resolve;
|
||||||
|
});
|
||||||
|
let releaseReconnect: (() => void) | undefined;
|
||||||
|
const reconnectGate = new Promise<void>((resolve) => {
|
||||||
|
releaseReconnect = resolve;
|
||||||
|
});
|
||||||
|
const session = new CodexAppServerAgentSession(
|
||||||
|
{ provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" },
|
||||||
|
null,
|
||||||
|
logger,
|
||||||
|
async () => {
|
||||||
|
spawnCount += 1;
|
||||||
|
if (spawnCount === 1) {
|
||||||
|
return initialAppServer.child;
|
||||||
|
}
|
||||||
|
markReconnectSpawned?.();
|
||||||
|
await reconnectGate;
|
||||||
|
return lateAppServer.child;
|
||||||
|
},
|
||||||
|
);
|
||||||
|
const events: AgentManagerEvent[] = [];
|
||||||
|
session.subscribe((event) => {
|
||||||
|
events.push({ type: "agent_stream", agentId: "test-agent", event });
|
||||||
|
});
|
||||||
|
const lateExit = new Promise<void>((resolve) => {
|
||||||
|
lateAppServer.child.once("exit", () => resolve());
|
||||||
|
});
|
||||||
|
|
||||||
|
try {
|
||||||
|
await session.connect();
|
||||||
|
initialAppServer.child.emit("exit", 17, null);
|
||||||
|
|
||||||
|
const turnStart = session.startTurn("continue after reconnect");
|
||||||
|
await reconnectSpawned;
|
||||||
|
await session.close();
|
||||||
|
releaseReconnect?.();
|
||||||
|
|
||||||
|
await expect(turnStart).rejects.toThrow("Codex app-server session is closed");
|
||||||
|
await expect(lateExit).resolves.toBeUndefined();
|
||||||
|
expect(events.filter((event) => event.event.type === "turn_failed")).toHaveLength(0);
|
||||||
|
|
||||||
|
await expect(session.connect()).rejects.toThrow("Codex app-server session is closed");
|
||||||
|
expect(spawnCount).toBe(2);
|
||||||
|
} finally {
|
||||||
|
releaseReconnect?.();
|
||||||
|
await session.close();
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
@@ -3164,6 +3164,8 @@ export class CodexAppServerAgentSession implements AgentSession {
|
|||||||
private unpairedCompactionNotificationCompletions = 0;
|
private unpairedCompactionNotificationCompletions = 0;
|
||||||
private unpairedCompactionItemCompletions = 0;
|
private unpairedCompactionItemCompletions = 0;
|
||||||
private connected = false;
|
private connected = false;
|
||||||
|
private connectionPromise: Promise<void> | null = null;
|
||||||
|
private closed = false;
|
||||||
private collaborationModes: Array<{
|
private collaborationModes: Array<{
|
||||||
name: string;
|
name: string;
|
||||||
mode?: string | null;
|
mode?: string | null;
|
||||||
@@ -3229,15 +3231,49 @@ export class CodexAppServerAgentSession implements AgentSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async connect(): Promise<void> {
|
async connect(): Promise<void> {
|
||||||
|
if (this.closed) {
|
||||||
|
throw this.createClosedError();
|
||||||
|
}
|
||||||
if (this.connected) return;
|
if (this.connected) return;
|
||||||
|
if (this.connectionPromise) {
|
||||||
|
await this.connectionPromise;
|
||||||
|
if (this.closed) {
|
||||||
|
throw this.createClosedError();
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const connectionPromise = this.establishConnection();
|
||||||
|
this.connectionPromise = connectionPromise;
|
||||||
|
try {
|
||||||
|
await connectionPromise;
|
||||||
|
} finally {
|
||||||
|
if (this.connectionPromise === connectionPromise) {
|
||||||
|
this.connectionPromise = null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (this.closed) {
|
||||||
|
throw this.createClosedError();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private async establishConnection(): Promise<void> {
|
||||||
const child = await this.spawnAppServer();
|
const child = await this.spawnAppServer();
|
||||||
this.client = new CodexAppServerClient(child, this.logger, () => this.traceContext());
|
const client = new CodexAppServerClient(child, this.logger, () => this.traceContext());
|
||||||
this.client.setNotificationHandler((method, params) => this.handleNotification(method, params));
|
if (this.closed) {
|
||||||
|
await client.dispose();
|
||||||
|
throw this.createClosedError();
|
||||||
|
}
|
||||||
|
this.client = client;
|
||||||
|
client.setUnexpectedTerminationHandler((error) => {
|
||||||
|
this.handleUnexpectedTermination(error);
|
||||||
|
});
|
||||||
|
client.setNotificationHandler((method, params) => this.handleNotification(method, params));
|
||||||
this.registerRequestHandlers();
|
this.registerRequestHandlers();
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await this.client.request("initialize", buildCodexAppServerInitializeParams());
|
await client.request("initialize", buildCodexAppServerInitializeParams());
|
||||||
this.client.notify("initialized", {});
|
client.notify("initialized", {});
|
||||||
|
|
||||||
await this.loadCollaborationModes();
|
await this.loadCollaborationModes();
|
||||||
await this.loadSkills();
|
await this.loadSkills();
|
||||||
@@ -3249,20 +3285,31 @@ export class CodexAppServerAgentSession implements AgentSession {
|
|||||||
await this.loadPersistedHistory();
|
await this.loadPersistedHistory();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (this.closed) {
|
||||||
|
throw this.createClosedError();
|
||||||
|
}
|
||||||
this.connected = true;
|
this.connected = true;
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
try {
|
try {
|
||||||
await this.close();
|
if (this.client === client) {
|
||||||
} catch (closeError) {
|
await this.disposeClient();
|
||||||
|
} else {
|
||||||
|
await client.dispose();
|
||||||
|
}
|
||||||
|
} catch (disposeError) {
|
||||||
this.logger.warn(
|
this.logger.warn(
|
||||||
{ err: closeError, connectError: error },
|
{ err: disposeError, connectError: error },
|
||||||
"Failed to close Codex app-server after connection failure",
|
"Failed to dispose Codex app-server client after connection failure",
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private createClosedError(): Error {
|
||||||
|
return new Error("Codex app-server session is closed");
|
||||||
|
}
|
||||||
|
|
||||||
private traceContext(): CodexAppServerTraceContext {
|
private traceContext(): CodexAppServerTraceContext {
|
||||||
return {
|
return {
|
||||||
agentId: this.agentId,
|
agentId: this.agentId,
|
||||||
@@ -3271,6 +3318,24 @@ export class CodexAppServerAgentSession implements AgentSession {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private handleUnexpectedTermination(error: Error): void {
|
||||||
|
this.connected = false;
|
||||||
|
const hasActiveRootTurn = this.activeForegroundTurnId !== null || this.currentTurnId !== null;
|
||||||
|
this.clearPendingPermissions({ preservePlanApprovals: !hasActiveRootTurn });
|
||||||
|
if (hasActiveRootTurn) {
|
||||||
|
this.emitEvent({
|
||||||
|
type: "turn_failed",
|
||||||
|
provider: CODEX_PROVIDER,
|
||||||
|
error: error.message,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
this.activeForegroundTurnId = null;
|
||||||
|
this.activeClientMessageId = null;
|
||||||
|
this.currentTurnId = null;
|
||||||
|
this.pendingForegroundTurnIdentification?.resolve(null);
|
||||||
|
this.pendingForegroundTurnIdentification = null;
|
||||||
|
}
|
||||||
|
|
||||||
private async loadCollaborationModes(): Promise<void> {
|
private async loadCollaborationModes(): Promise<void> {
|
||||||
if (!this.client) return;
|
if (!this.client) return;
|
||||||
try {
|
try {
|
||||||
@@ -4308,25 +4373,39 @@ export class CodexAppServerAgentSession implements AgentSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async close(): Promise<void> {
|
async close(): Promise<void> {
|
||||||
for (const pending of this.pendingPermissionHandlers.values()) {
|
this.closed = true;
|
||||||
pending.resolve({ decision: "cancel" });
|
this.clearPendingPermissions();
|
||||||
}
|
|
||||||
this.pendingPermissionHandlers.clear();
|
|
||||||
this.pendingPermissions.clear();
|
|
||||||
this.resolvedPermissionRequests.clear();
|
|
||||||
this.pendingSubAgentNotificationsByThreadId.clear();
|
this.pendingSubAgentNotificationsByThreadId.clear();
|
||||||
this.subscribers.clear();
|
this.subscribers.clear();
|
||||||
this.activeForegroundTurnId = null;
|
this.activeForegroundTurnId = null;
|
||||||
this.activeClientMessageId = null;
|
this.activeClientMessageId = null;
|
||||||
this.pendingForegroundTurnIdentification?.resolve(null);
|
this.pendingForegroundTurnIdentification?.resolve(null);
|
||||||
this.pendingForegroundTurnIdentification = null;
|
this.pendingForegroundTurnIdentification = null;
|
||||||
if (this.client) {
|
await this.disposeClient();
|
||||||
await this.client.dispose();
|
this.currentThreadId = null;
|
||||||
|
}
|
||||||
|
|
||||||
|
private clearPendingPermissions(options?: { preservePlanApprovals?: boolean }): void {
|
||||||
|
for (const [requestId, pending] of this.pendingPermissionHandlers) {
|
||||||
|
if (options?.preservePlanApprovals && pending.kind === "plan") {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
pending.resolve({ decision: "cancel" });
|
||||||
|
this.pendingPermissionHandlers.delete(requestId);
|
||||||
|
this.pendingPermissions.delete(requestId);
|
||||||
}
|
}
|
||||||
|
this.mcpElicitationPermissionIds.clear();
|
||||||
|
this.resolvedPermissionRequests.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
private async disposeClient(): Promise<void> {
|
||||||
|
const client = this.client;
|
||||||
this.client = null;
|
this.client = null;
|
||||||
this.connected = false;
|
this.connected = false;
|
||||||
this.currentThreadId = null;
|
|
||||||
this.currentTurnId = null;
|
this.currentTurnId = null;
|
||||||
|
if (client) {
|
||||||
|
await client.dispose();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async listCommands(): Promise<AgentSlashCommand[]> {
|
async listCommands(): Promise<AgentSlashCommand[]> {
|
||||||
|
|||||||
@@ -35,6 +35,7 @@ interface PendingRequest {
|
|||||||
|
|
||||||
type RequestHandler = (params: unknown, requestId: number) => unknown;
|
type RequestHandler = (params: unknown, requestId: number) => unknown;
|
||||||
type NotificationHandler = (method: string, params: unknown) => void;
|
type NotificationHandler = (method: string, params: unknown) => void;
|
||||||
|
type UnexpectedTerminationHandler = (error: Error) => void;
|
||||||
|
|
||||||
export interface CodexThreadForkParams {
|
export interface CodexThreadForkParams {
|
||||||
threadId: string;
|
threadId: string;
|
||||||
@@ -159,6 +160,7 @@ export class CodexAppServerClient {
|
|||||||
private readonly pending = new Map<number, PendingRequest>();
|
private readonly pending = new Map<number, PendingRequest>();
|
||||||
private readonly requestHandlers = new Map<string, RequestHandler>();
|
private readonly requestHandlers = new Map<string, RequestHandler>();
|
||||||
private notificationHandler: NotificationHandler | null = null;
|
private notificationHandler: NotificationHandler | null = null;
|
||||||
|
private unexpectedTerminationHandler: UnexpectedTerminationHandler | null = null;
|
||||||
private nextId = 1;
|
private nextId = 1;
|
||||||
private disposed = false;
|
private disposed = false;
|
||||||
private stderrBuffer = "";
|
private stderrBuffer = "";
|
||||||
@@ -184,12 +186,7 @@ export class CodexAppServerClient {
|
|||||||
|
|
||||||
child.on("error", (err) => {
|
child.on("error", (err) => {
|
||||||
this.logger.error({ err }, "Codex app-server child process error");
|
this.logger.error({ err }, "Codex app-server child process error");
|
||||||
for (const pending of this.pending.values()) {
|
this.handleUnexpectedTermination(err);
|
||||||
clearTimeout(pending.timer);
|
|
||||||
pending.reject(err);
|
|
||||||
}
|
|
||||||
this.pending.clear();
|
|
||||||
this.disposed = true;
|
|
||||||
});
|
});
|
||||||
|
|
||||||
child.on("exit", (code, signal) => {
|
child.on("exit", (code, signal) => {
|
||||||
@@ -198,15 +195,14 @@ export class CodexAppServerClient {
|
|||||||
? "Codex app-server exited"
|
? "Codex app-server exited"
|
||||||
: `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`;
|
: `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`;
|
||||||
const error = new Error(`${message}\n${this.stderrBuffer}`.trim());
|
const error = new Error(`${message}\n${this.stderrBuffer}`.trim());
|
||||||
for (const pending of this.pending.values()) {
|
this.handleUnexpectedTermination(error);
|
||||||
clearTimeout(pending.timer);
|
|
||||||
pending.reject(error);
|
|
||||||
}
|
|
||||||
this.pending.clear();
|
|
||||||
this.disposed = true;
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
setUnexpectedTerminationHandler(handler: UnexpectedTerminationHandler): void {
|
||||||
|
this.unexpectedTerminationHandler = handler;
|
||||||
|
}
|
||||||
|
|
||||||
setNotificationHandler(handler: NotificationHandler): void {
|
setNotificationHandler(handler: NotificationHandler): void {
|
||||||
this.notificationHandler = handler;
|
this.notificationHandler = handler;
|
||||||
}
|
}
|
||||||
@@ -251,6 +247,7 @@ export class CodexAppServerClient {
|
|||||||
async dispose(): Promise<void> {
|
async dispose(): Promise<void> {
|
||||||
if (this.disposed) return;
|
if (this.disposed) return;
|
||||||
this.disposed = true;
|
this.disposed = true;
|
||||||
|
this.unexpectedTerminationHandler = null;
|
||||||
this.rl.close();
|
this.rl.close();
|
||||||
try {
|
try {
|
||||||
this.child.stdin.end();
|
this.child.stdin.end();
|
||||||
@@ -275,6 +272,29 @@ export class CodexAppServerClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private handleUnexpectedTermination(error: Error): void {
|
||||||
|
if (this.disposed) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
this.disposed = true;
|
||||||
|
this.rl.close();
|
||||||
|
for (const pending of this.pending.values()) {
|
||||||
|
clearTimeout(pending.timer);
|
||||||
|
pending.reject(error);
|
||||||
|
}
|
||||||
|
this.pending.clear();
|
||||||
|
const handler = this.unexpectedTerminationHandler;
|
||||||
|
this.unexpectedTerminationHandler = null;
|
||||||
|
if (!handler) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
handler(error);
|
||||||
|
} catch (handlerError) {
|
||||||
|
this.logger.warn({ err: handlerError }, "Codex app-server termination handler threw");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private writeJsonRpcResponse(response: JsonRpcResponse): void {
|
private writeJsonRpcResponse(response: JsonRpcResponse): void {
|
||||||
if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) {
|
if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) {
|
||||||
return;
|
return;
|
||||||
|
|||||||
@@ -51,6 +51,7 @@ export interface FakeCodexAppServer {
|
|||||||
waitForTurnStart(): Promise<JsonObject>;
|
waitForTurnStart(): Promise<JsonObject>;
|
||||||
nextResponse(): Promise<string>;
|
nextResponse(): Promise<string>;
|
||||||
startsTurn(params: { threadId: string; turnId?: string }): void;
|
startsTurn(params: { threadId: string; turnId?: string }): void;
|
||||||
|
updatesPlan(params: { threadId: string; steps: string[] }): void;
|
||||||
completeTurn(params?: { threadId?: string }): void;
|
completeTurn(params?: { threadId?: string }): void;
|
||||||
startsSubAgent(params: {
|
startsSubAgent(params: {
|
||||||
callId: string;
|
callId: string;
|
||||||
@@ -319,6 +320,17 @@ export function createFakeCodexAppServer(
|
|||||||
})}\n`,
|
})}\n`,
|
||||||
);
|
);
|
||||||
},
|
},
|
||||||
|
updatesPlan(params) {
|
||||||
|
child.stdout.write(
|
||||||
|
`${JSON.stringify({
|
||||||
|
method: "turn/plan/updated",
|
||||||
|
params: {
|
||||||
|
threadId: params.threadId,
|
||||||
|
plan: params.steps.map((step) => ({ step, status: "pending" })),
|
||||||
|
},
|
||||||
|
})}\n`,
|
||||||
|
);
|
||||||
|
},
|
||||||
completeTurn(params = {}) {
|
completeTurn(params = {}) {
|
||||||
child.stdout.write(
|
child.stdout.write(
|
||||||
`${JSON.stringify({
|
`${JSON.stringify({
|
||||||
|
|||||||
Reference in New Issue
Block a user