mirrors/pikasTech-agentrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: keep codex stdio server alive across turns

Browse Source
This commit is contained in:
Codex
2026-06-02 01:35:31 +08:00
parent 4ddea73629
commit 1ff2f114b3
5 changed files with 171 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/codex-stdio.ts
+120 -39
View File
@@ -99,6 +99,7 @@ export class CodexStdioClient {
this.child = spawn(command, args, {
cwd: options.cwd,
env: options.env ?? process.env,
detached: true,
stdio: "pipe",
});
} catch (error) {
@@ -111,6 +112,10 @@ export class CodexStdioClient {
this.child.on("error", (error) => this.handleClose(127, null, spawnFailure(command, error)));
}
get isClosed(): boolean {
return this.closed;
}
request(method: string, params: JsonRecord, timeoutMs = requestTimeoutCapMs): Promise<unknown> {
if (this.closed) return Promise.reject(this.closeFailure ?? new CodexStdioFailure("backend-failed", "codex app-server is closed", `request:${method}`));
const id = this.nextId++;
@@ -135,12 +140,25 @@ export class CodexStdioClient {
stop(): void {
if (this.closed) return;
this.child.kill("SIGTERM");
this.kill("SIGTERM");
setTimeout(() => {
if (!this.closed) this.child.kill("SIGKILL");
if (!this.closed) this.kill("SIGKILL");
}, 1500).unref?.();
}
private kill(signal: NodeJS.Signals): void {
const pid = this.child.pid;
if (typeof pid === "number") {
try {
process.kill(-pid, signal);
return;
} catch {
// Fall back to killing the direct child when process-group termination is unavailable.
}
}
this.child.kill(signal);
}
private appendStderr(chunk: Buffer): void {
this.stderrBytes += chunk.byteLength;
const next = Buffer.concat([this.stderrTailBuffer, chunk]);
@@ -260,21 +278,84 @@ export class CodexStdioClient {
}
export async function runCodexStdioTurn(options: CodexStdioTurnOptions): Promise<BackendTurnResult> {
const session = new CodexStdioBackendSession();
const result = await session.runTurn(options);
const closeEvents = await session.close();
return { ...result, events: [...result.events, ...closeEvents] };
}
export class CodexStdioBackendSession {
private client: CodexStdioClient | null = null;
private clientKey: string | null = null;
async runTurn(options: CodexStdioTurnOptions): Promise<BackendTurnResult> {
return await runCodexStdioTurnWithSession(options, this);
}
async close(): Promise<BackendEvent[]> {
const client = this.client;
if (!client) return [];
this.client = null;
this.clientKey = null;
client.stop();
const closeInfo = await client.closedPromise;
return [{ type: "backend_status", payload: { phase: "codex-app-server-closed", appServerExit: closeEvent(closeInfo) } }];
}
async getClient(options: CodexStdioTurnOptions, env: NodeJS.ProcessEnv, events: BackendEvent[]): Promise<CodexStdioClient> {
const key = codexClientKey(options, env);
if (this.client && !this.client.isClosed && this.clientKey === key) {
events.push({ type: "backend_status", payload: { phase: "codex-app-server:reused", ...backendMetadata(options), protocol: codexProtocol } });
return this.client;
}
const closeEvents = await this.close();
events.push(...closeEvents);
events.push({
type: "backend_status",
payload: {
phase: "codex-app-server-starting",
...backendMetadata(options),
protocol: codexProtocol,
runtime: runtimeSummary(options, env, resolveCodexHome(options)),
},
});
const clientOptions: ConstructorParameters<typeof CodexStdioClient>[0] = {
cwd: options.cwd,
env,
onNotification: (message) => this.onNotification(message),
};
if (options.command) clientOptions.command = options.command;
if (options.args) clientOptions.args = options.args;
this.client = new CodexStdioClient(clientOptions);
this.clientKey = key;
const requestTimeoutMs = Math.min(positiveTimeout(options.timeoutMs), requestTimeoutCapMs);
const initializeResult = requireResponseRecord(await this.client.request("initialize", { clientInfo: { name: "agentrun", title: "AgentRun", version: "0.1.0" }, capabilities: { experimentalApi: true } }, requestTimeoutMs), "initialize");
validateInitializeResponse(initializeResult);
this.client.notify("initialized", {});
events.push({ type: "backend_status", payload: { phase: "initialize:completed", ...backendMetadata(options), protocol: codexProtocol } });
return this.client;
}
private notificationHandlers = new Set<(message: JsonRecord) => void>();
addNotificationHandler(handler: (message: JsonRecord) => void): () => void {
this.notificationHandlers.add(handler);
return () => this.notificationHandlers.delete(handler);
}
private onNotification(message: JsonRecord): void {
for (const handler of this.notificationHandlers) handler(message);
}
}
async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, session: CodexStdioBackendSession): Promise<BackendTurnResult> {
const codexHome = resolveCodexHome(options);
const projectionFailure = await prepareProjectedCodexHome(codexHome, options.env?.AGENTRUN_CODEX_SECRET_HOME ?? process.env.AGENTRUN_CODEX_SECRET_HOME);
if (projectionFailure) return projectionFailure;
const secretFailure = codexHomeReadiness(codexHome);
if (secretFailure) return secretFailure;
const env = childEnv(options, codexHome);
const events: BackendEvent[] = [{
type: "backend_status",
payload: {
phase: "codex-app-server-starting",
...backendMetadata(options),
protocol: codexProtocol,
runtime: runtimeSummary(options, env, codexHome),
},
}];
const events: BackendEvent[] = [];
if (options.abortSignal?.aborted) {
const cancelled = { status: "cancelled" as const, failureKind: "cancelled" as const, message: "cancel requested" };
events.push({ type: "backend_status", payload: { phase: "turn-cancelled", failureKind: "cancelled" } });
@@ -305,30 +386,20 @@ export async function runCodexStdioTurn(options: CodexStdioTurnOptions): Promise
client?.stop();
terminalResolve();
}, positiveTimeout(options.timeoutMs));
const stopNotifications = session.addNotificationHandler((message) => {
const normalized = normalizeCodexNotification(message);
if (normalized.threadId) threadId = normalized.threadId;
if (normalized.turnId) turnId = normalized.turnId;
if (normalized.assistantDelta) assistantText += normalized.assistantDelta;
if (typeof normalized.assistantFinal === "string" && normalized.assistantFinal.trim().length > 0) finalAssistantText = normalized.assistantFinal;
events.push(...normalized.events);
if (normalized.terminal && !terminal) {
terminal = normalized.terminal;
terminalResolve();
}
});
try {
const clientOptions: ConstructorParameters<typeof CodexStdioClient>[0] = {
cwd: options.cwd,
env,
onNotification: (message) => {
const normalized = normalizeCodexNotification(message);
if (normalized.threadId) threadId = normalized.threadId;
if (normalized.turnId) turnId = normalized.turnId;
if (normalized.assistantDelta) assistantText += normalized.assistantDelta;
if (typeof normalized.assistantFinal === "string" && normalized.assistantFinal.trim().length > 0) finalAssistantText = normalized.assistantFinal;
events.push(...normalized.events);
if (normalized.terminal && !terminal) {
terminal = normalized.terminal;
terminalResolve();
}
},
};
if (options.command) clientOptions.command = options.command;
if (options.args) clientOptions.args = options.args;
client = new CodexStdioClient(clientOptions);
const initializeResult = requireResponseRecord(await client.request("initialize", { clientInfo: { name: "agentrun", title: "AgentRun", version: "0.1.0" }, capabilities: { experimentalApi: true } }, requestTimeoutMs), "initialize");
validateInitializeResponse(initializeResult);
client.notify("initialized", {});
events.push({ type: "backend_status", payload: { phase: "initialize:completed", ...backendMetadata(options), protocol: codexProtocol } });
client = await session.getClient(options, env, events);
const startThread = async (phasePrefix = "thread/start"): Promise<string> => {
const response = requireResponseRecord(await client!.request("thread/start", withOptionalModel({ cwd: options.cwd, approvalPolicy: options.approvalPolicy, sandbox: options.sandbox, serviceName: "agentrun" }, options.model), requestTimeoutMs), "thread/start");
@@ -381,15 +452,12 @@ export async function runCodexStdioTurn(options: CodexStdioTurnOptions): Promise
events.push({ type: "error", payload: { failureKind: failure.failureKind, message: failure.message, phase: failure.phase, details: failure.details } });
}
} finally {
stopNotifications();
options.abortSignal?.removeEventListener("abort", abortTurn);
clearTimeout(timeout);
if (client) {
client.stop();
const closeInfo = await client.closedPromise;
events.push({ type: "backend_status", payload: { phase: "codex-app-server-closed", appServerExit: closeEvent(closeInfo) } });
}
}
if (!terminal) terminal = { status: "failed", failureKind: "backend-response-invalid", message: "codex app-server finished without terminal status" };
if (terminal.status !== "completed") events.push(...await session.close());
const reply = finalAssistantText.trim().length > 0 ? finalAssistantText : assistantText;
if (reply.trim().length > 0) events.push({ type: "assistant_message", payload: { text: reply } });
events.push({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } });
@@ -527,6 +595,19 @@ function childEnv(options: CodexStdioTurnOptions, codexHome: string): NodeJS.Pro
};
}
function codexClientKey(options: CodexStdioTurnOptions, env: NodeJS.ProcessEnv): string {
return JSON.stringify({
command: options.command ?? "codex",
args: options.args ?? defaultCodexArgs,
cwd: options.cwd,
codexHome: env.CODEX_HOME ?? resolveCodexHome(options),
backendProfile: options.backendProfile ?? "codex",
model: options.model ?? null,
approvalPolicy: options.approvalPolicy,
sandbox: options.sandbox,
});
}
function resolveCodexHome(options: CodexStdioTurnOptions): string {
return options.codexHome ?? options.env?.CODEX_HOME ?? `${options.env?.HOME ?? process.env.HOME ?? ""}/.codex`;
}