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

fix: isolate codex stdio turn routing

Browse Source
This commit is contained in:
Codex
2026-06-01 00:24:20 +08:00
parent 7b64427690
commit bbc3f8dca8
6 changed files with 422 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/codex-stdio-session.ts
+46 -35
View File
@@ -89,7 +89,6 @@ import {
gitConfigEnv,
childProcessEnvState,
codexProviderNoProxyEntries,
rpcClientConfigSignature,
mergeNoProxy,
splitNoProxy,
codexStdioSafety,
@@ -172,6 +171,7 @@ export const CODEX_STDIO_COMMAND_PROBE_TIMEOUT_MS = 3000;
export const CODEX_STDIO_COMMAND_PROBE_TRACE_ID = "trc_codex_stdio_command_probe";
export const CODEX_STDIO_COMMAND_PROBE_CONVERSATION_ID = "cnv_codex_stdio_command_probe";
export const CODEX_STDIO_FIRST_TOKEN_PROGRESS_MS = 15 * 1000;
export const CODEX_STDIO_NO_PROGRESS_DIAGNOSIS_MS = 60 * 1000;
export const CODEX_CHILD_STRIPPED_ENV_KEYS = Object.freeze([
"OPENAI_API_KEY",
"CODEX_API_KEY",
@@ -230,10 +230,9 @@ export function createCodexStdioSessionManager(options = {}) {
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
const sessions = new Map();
const conversations = new Map();
let rpcClient = options.rpcClient ?? null;
let rpcClientSignature = options.rpcClient ? "injected" : null;
let rpcStartedAt = null;
let rpcInitialized = false;
const activeRpcClients = new Map();
let protocolProbe = null;
let commandProbe = null;
@@ -548,7 +547,8 @@ export function createCodexStdioSessionManager(options = {}) {
});
try {
const client = await ensureRpcClient({ env, availability, timeoutMs: params.timeoutMs });
let client = null;
client = await createTurnRpcClient({ env, availability, timeoutMs: params.timeoutMs, sessionId: session.sessionId });
availability = describe({ ...params, env, workspace, sandbox });
traceRecorder.append({
type: "stdio",
@@ -678,6 +678,7 @@ export function createCodexStdioSessionManager(options = {}) {
idleMs: turnResult.idleMs,
lastActivityAt: turnResult.lastActivityAt,
waitingFor: turnResult.waitingFor,
diagnosis: turnResult.diagnosis,
providerTrace
});
}
@@ -744,6 +745,9 @@ export function createCodexStdioSessionManager(options = {}) {
turn: session.turn,
terminal: true
});
const providerTrace = codexStdioProviderTrace({ availability, toolName, turnResult, session });
closeTurnRpcClient(session.sessionId, client);
client = null;
return {
provider: CODEX_STDIO_PROVIDER,
model: firstNonEmpty(params.model, env.HWLAB_CODE_AGENT_MODEL, env.OPENAI_MODEL, "codex-default"),
@@ -778,10 +782,10 @@ export function createCodexStdioSessionManager(options = {}) {
outputTruncated: sidecar.outputTruncated
}),
capabilityLevel: CODEX_STDIO_CAPABILITY_LEVEL,
providerTrace: codexStdioProviderTrace({ availability, toolName, turnResult, session })
providerTrace
};
} catch (error) {
closeRpcClient();
closeTurnRpcClient(session?.sessionId);
const timeout = /timed out|timeout|超时/iu.test(String(error.message ?? ""));
const currentSession = sessions.get(session.sessionId) ?? null;
const canceled = currentSession?.status === "canceled";
@@ -808,6 +812,8 @@ export function createCodexStdioSessionManager(options = {}) {
idleMs: timeout ? error.idleMs : undefined,
lastActivityAt: timeout ? error.lastActivityAt : undefined,
hardTimeoutMs: timeout ? error.hardTimeoutMs : undefined,
diagnosis: error.diagnosis ?? undefined,
diagnosisCode: error.diagnosis?.code ?? error.diagnosisCode ?? undefined,
sessionId: session.sessionId,
sessionStatus: session.status,
turn: session.turn,
@@ -870,7 +876,7 @@ export function createCodexStdioSessionManager(options = {}) {
let availability = describe({ ...params, env, workspace, command, sandbox });
const startupBlockers = availability.blockers.filter((blocker) => blocker.code !== "stdio_protocol_not_wired");
if (startupBlockers.length > 0) return availability;
if (rpcClient && rpcInitialized === true) {
if (rpcInitialized === true) {
availability = describe({ ...params, env, workspace, command, sandbox, protocolInitialized: true });
return ensureCommandProbeReady({
env,
@@ -972,9 +978,7 @@ export function createCodexStdioSessionManager(options = {}) {
session.lastTraceId = optionalId(params.traceId) ?? session.lastTraceId;
session.currentTraceId = null;
session.statusReason = params.reason ?? "user_canceled";
if (rpcClient && typeof rpcClient.close === "function") {
closeRpcClient();
}
closeTurnRpcClient(session.sessionId);
return publicSession(session, { conversationId: params.conversationId, reused: true });
}
@@ -989,9 +993,7 @@ export function createCodexStdioSessionManager(options = {}) {
session.currentTraceId = null;
reaped.push(session.sessionId);
}
if (sessionsBusyCount() === 0 && reaped.length > 0) {
closeRpcClient();
}
if (sessionsBusyCount() === 0 && reaped.length > 0) closeAllRpcClients();
return {
reapedCount: reaped.length,
sessionIds: reaped
@@ -1006,14 +1008,28 @@ export function createCodexStdioSessionManager(options = {}) {
function clear() {
sessions.clear();
conversations.clear();
closeRpcClient();
closeAllRpcClients();
}
function closeRpcClient() {
if (rpcClient && typeof rpcClient.close === "function") {
rpcClient.close();
function closeTurnRpcClient(sessionId, expectedClient = null) {
const key = optionalId(sessionId);
const activeClient = key ? activeRpcClients.get(key) ?? null : null;
const client = expectedClient ?? activeClient;
if (client && (!activeClient || activeClient === client) && typeof client.close === "function") {
client.close();
}
rpcClient = null;
if (key && (!expectedClient || activeClient === expectedClient)) activeRpcClients.delete(key);
if (activeRpcClients.size === 0) {
rpcStartedAt = null;
rpcInitialized = false;
}
}
function closeAllRpcClients() {
for (const client of activeRpcClients.values()) {
if (client && typeof client.close === "function") client.close();
}
activeRpcClients.clear();
rpcStartedAt = null;
rpcInitialized = false;
protocolProbe = null;
@@ -1097,20 +1113,10 @@ export function createCodexStdioSessionManager(options = {}) {
return describe({ env, workspace, command, sandbox, protocolInitialized: availability.protocol?.initialized ?? rpcInitialized });
}
async function ensureRpcClient({ env, availability, timeoutMs } = {}) {
async function createTurnRpcClient({ env, availability, timeoutMs, sessionId } = {}) {
const args = codexAppServerArgs(env);
const childEnv = childProcessEnv(env);
const nextSignature = rpcClientConfigSignature({ availability, args, childEnv });
if (rpcClient && typeof rpcClient.startTurn === "function" && rpcClientSignature === nextSignature) return rpcClient;
if (rpcClient && rpcClientSignature !== nextSignature && typeof rpcClient.close === "function") {
rpcClient.close();
}
if (rpcClientSignature !== nextSignature) {
rpcClient = null;
rpcInitialized = false;
rpcClientSignature = null;
}
rpcClient = options.createRpcClient
const client = options.createRpcClient
? await options.createRpcClient({ env, availability })
: createCodexAppServerJsonLineClient({
command: availability.command,
@@ -1118,13 +1124,17 @@ export function createCodexStdioSessionManager(options = {}) {
env: childEnv,
cwd: availability.workspace
});
if (typeof rpcClient.initialize === "function") {
await rpcClient.initialize(effectiveTimeout(timeoutMs));
const key = requiredId(sessionId, "ses");
activeRpcClients.set(key, client);
try {
if (typeof client.initialize === "function") await client.initialize(effectiveTimeout(timeoutMs));
} catch (error) {
closeTurnRpcClient(key, client);
throw error;
}
rpcClientSignature = nextSignature;
rpcInitialized = true;
rpcStartedAt = timestampFor(nowDefault);
return rpcClient;
return client;
}
async function acquireSession(params = {}) {
@@ -1182,7 +1192,7 @@ export function createCodexStdioSessionManager(options = {}) {
session.updatedAt = timestamp;
session.currentTraceId = null;
session.statusReason = staleBusy.reason;
if (staleBusy.closeRpcClient) closeRpcClient();
if (staleBusy.closeRpcClient) closeTurnRpcClient(session.sessionId);
} else {
return blockedAcquire({
code: "session_busy",
@@ -1599,6 +1609,7 @@ export function createCodexAppServerJsonLineClient({ command = DEFAULT_CODEX_STD
}
function handleServerRequest(id, method, params = {}) {
notifyClientRequest(method, params, "received");
if (method === "item/commandExecution/requestApproval" || method === "item/fileChange/requestApproval" || method === "item/permissions/requestApproval") {
const approval = approvalResponseForServerRequest(params);
notifyClientRequest(method, params, approval.decision);