From 4415db9696680d49007374ddbb1489cb12df2359 Mon Sep 17 00:00:00 2001 From: Code Queue Review Date: Sun, 24 May 2026 08:16:24 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20Codex=20stdio=20?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=20turn=20=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 基于 HWLAB #426 修复 app-server retry 通知误判，并从 repo-owned DEV Responses env 派生 Codex provider base_url，避免子进程继续使用过期挂载配置。 验证真实 completed reply 仅记录非敏感状态和长度，不打印 assistant 正文或 Secret。 --- .../code-agent-session-registry.test.mjs | 89 ++++++++++++++- internal/cloud/codex-stdio-session.mjs | 105 +++++++++++++++--- 2 files changed, 174 insertions(+), 20 deletions(-) diff --git a/internal/cloud/code-agent-session-registry.test.mjs b/internal/cloud/code-agent-session-registry.test.mjs index 4d9a78fe..dd13e042 100644 --- a/internal/cloud/code-agent-session-registry.test.mjs +++ b/internal/cloud/code-agent-session-registry.test.mjs @@ -7,7 +7,11 @@ import test from "node:test"; import { createCodeAgentSessionRegistry } from "./code-agent-session-registry.mjs"; import { handleCodeAgentChat, validateCodeAgentChatSchema } from "./code-agent-chat.mjs"; -import { createCodexStdioSessionManager } from "./codex-stdio-session.mjs"; +import { + codexAppServerArgs, + codexAppServerProviderBaseUrl, + createCodexStdioSessionManager +} from "./codex-stdio-session.mjs"; import { classifyCodexRunnerCapability, classifyCodeAgentChatReadiness @@ -1524,7 +1528,7 @@ async function prepareFakeCodexHome(root = null) { return codexHome; } -function createFakeAppServerClient({ calls, responses = ["first stdio reply", "second stdio reply"], failedTurn = null } = {}) { +function createFakeAppServerClient({ calls, responses = ["first stdio reply", "second stdio reply"], failedTurn = null, retryingTurn = null } = {}) { let notificationHandler = null; let turn = 0; return { @@ -1550,6 +1554,17 @@ function createFakeAppServerClient({ calls, responses = ["first stdio reply", "s turn += 1; const turnId = `turn_stdio_${turn}`; notificationHandler?.({ method: "turn/started", params: { turn: { id: turnId } } }); + if (retryingTurn && turn === retryingTurn.turn) { + notificationHandler?.({ + method: "error", + params: { + threadId: args.threadId, + turnId, + willRetry: true, + error: { message: retryingTurn.message } + } + }); + } if (failedTurn && turn === failedTurn.turn) { notificationHandler?.({ method: "turn/completed", @@ -1713,6 +1728,23 @@ test("Codex stdio feasibility verifies writable workspace, CODEX_HOME, and versi } }); +test("Codex app-server args pin repo-owned DEV responses egress without leaking provider env", () => { + const env = { + HWLAB_CODE_AGENT_OPENAI_BASE_URL: "http://172.26.26.227:17680/v1/responses", + OPENAI_API_KEY: "test-openai-key-material" + }; + + assert.equal(codexAppServerProviderBaseUrl(env), "http://172.26.26.227:17680/v1"); + const args = codexAppServerArgs(env); + assert.deepEqual(args.slice(0, 1), ["app-server"]); + assert.deepEqual(args.slice(-2), ["--listen", "stdio://"]); + assert.ok(args.includes("model_provider=\"OpenAI\"")); + assert.ok(args.includes("model_providers.OpenAI.base_url=\"http://172.26.26.227:17680/v1\"")); + assert.ok(args.includes("model_providers.OpenAI.wire_api=\"responses\"")); + assert.equal(args.join(" ").includes("test-openai-key-material"), false); + assert.equal(args.join(" ").includes("/v1/responses"), false); +}); + test("repo-owned Codex stdio manager creates and reuses long-lived sessions with trace evidence", async () => { const calls = []; const fakeCodex = await createFakeCodexCommand(); @@ -1870,6 +1902,59 @@ test("Codex stdio runner startup failure marks session failed and blocks long-li await rm(codexHome, { recursive: true, force: true }); }); +test("Codex app-server retrying error notification waits for completed turn", async () => { + const calls = []; + const fakeCodex = await createFakeCodexCommand(); + const codexHome = await prepareFakeCodexHome(); + const manager = createCodexStdioSessionManager({ + idFactory: () => "ses_stdio_retrying_turn", + createRpcClient: async () => createFakeAppServerClient({ + calls, + responses: ["retry recovered stdio reply"], + retryingTurn: { turn: 1, message: "Reconnecting... 1/5" } + }) + }); + const env = { + PATH: process.env.PATH, + OPENAI_API_KEY: "test-openai-key-material", + CODEX_HOME: codexHome, + HWLAB_CODE_AGENT_PROVIDER: "codex-stdio", + HWLAB_CODE_AGENT_MODEL: "gpt-test", + HWLAB_CODE_AGENT_CODEX_COMMAND: fakeCodex.command, + HWLAB_CODE_AGENT_CODEX_STDIO_ENABLED: "1", + HWLAB_CODE_AGENT_CODEX_STDIO_SUPERVISOR: "repo-owned", + HWLAB_CODE_AGENT_WORKSPACE: process.cwd(), + HWLAB_CODE_AGENT_OPENAI_BASE_URL: "http://172.26.26.227:17680/v1/responses" + }; + + const payload = await handleCodeAgentChat( + { + conversationId: "cnv_stdio_retrying_turn", + traceId: "trc_stdio_retrying_turn", + message: "需要长会话" + }, + { + now: () => "2026-05-23T00:08:45.000Z", + codexStdioManager: manager, + env + } + ); + + validateCodeAgentChatSchema(payload); + assert.equal(payload.status, "completed"); + assert.equal(payload.reply.content, "retry recovered stdio reply"); + assert.equal(payload.session.status, "idle"); + assert.equal(payload.longLivedSessionGate.status, "pass"); + assert.equal(payload.runnerTrace.valuesPrinted, false); + assert.ok(payload.runnerTrace.events.some((event) => event.label === "app-server:retrying")); + assert.ok(payload.runnerTrace.events.some((event) => event.label === "turn:completed:completed")); + assert.equal(payload.runnerTrace.events.some((event) => event.label === "app-server:error"), false); + assert.deepEqual(calls.filter((call) => call.method !== "initialize").map((call) => call.method), ["thread/start", "turn/start"]); + assert.equal(classifyCodexRunnerCapability(payload, { httpStatus: 200 }).capabilityPass, true); + await rm(fakeCodex.root, { recursive: true, force: true }); + await rm(codexHome, { recursive: true, force: true }); +}); + test("Codex app-server failed turn is not wrapped as completed assistant output", async () => { const calls = []; const fakeCodex = await createFakeCodexCommand(); diff --git a/internal/cloud/codex-stdio-session.mjs b/internal/cloud/codex-stdio-session.mjs index 3abb4713..2cfe04e1 100644 --- a/internal/cloud/codex-stdio-session.mjs +++ b/internal/cloud/codex-stdio-session.mjs @@ -704,15 +704,18 @@ export function createCodexStdioSessionManager(options = {}) { let probeClient = null; const startedAt = timestampFor(params.now ?? nowDefault); try { - probeClient = options.createProbeRpcClient - ? await options.createProbeRpcClient({ env, availability }) - : options.createRpcClient - ? await options.createRpcClient({ env, availability, probe: true }) - : createCodexAppServerJsonLineClient({ - command: availability.command, - env: childProcessEnv(env), - cwd: availability.workspace - }); + if (options.createProbeRpcClient) { + probeClient = await options.createProbeRpcClient({ env, availability }); + } else if (options.createRpcClient) { + probeClient = await options.createRpcClient({ env, availability, probe: true }); + } else { + probeClient = createCodexAppServerJsonLineClient({ + command: availability.command, + args: codexAppServerArgs(env), + env: childProcessEnv(env), + cwd: availability.workspace + }); + } const timeoutMs = positiveInteger(params.probeTimeoutMs, 10000); if (typeof probeClient.initialize === "function") await probeClient.initialize(timeoutMs); protocolProbe = { @@ -904,6 +907,7 @@ export function createCodexStdioSessionManager(options = {}) { ? await options.createRpcClient({ env, availability }) : createCodexAppServerJsonLineClient({ command: availability.command, + args: codexAppServerArgs(env), env: childProcessEnv(env), cwd: availability.workspace }); @@ -1094,8 +1098,55 @@ export function createCodexStdioSessionManager(options = {}) { }; } -export function createCodexAppServerJsonLineClient({ command = DEFAULT_CODEX_STDIO_COMMAND, env = process.env, cwd = repoRoot, onNotification = null } = {}) { - const child = spawn(command, ["app-server", "--listen", "stdio://"], { +export function codexAppServerArgs(env = process.env) { + return [ + "app-server", + ...codexAppServerProviderConfigArgs(env), + "--listen", + "stdio://" + ]; +} + +export function codexAppServerProviderBaseUrl(env = process.env) { + const configured = firstNonEmpty(env.HWLAB_CODE_AGENT_OPENAI_BASE_URL, env.OPENAI_BASE_URL); + if (!configured) return null; + try { + const url = new URL(configured); + url.hash = ""; + url.search = ""; + const normalizedPath = url.pathname.replace(/\/+$/u, ""); + if (normalizedPath.endsWith("/responses")) { + url.pathname = normalizedPath.slice(0, -"/responses".length) || "/"; + } + const serialized = url.toString(); + return serialized.endsWith("/") && url.pathname === "/" ? serialized.slice(0, -1) : serialized.replace(/\/$/u, ""); + } catch { + return null; + } +} + +function codexAppServerProviderConfigArgs(env = process.env) { + const baseUrl = codexAppServerProviderBaseUrl(env); + if (!baseUrl) return []; + return [ + "-c", + "model_provider=\"OpenAI\"", + "-c", + `model_providers.OpenAI.base_url=${tomlString(baseUrl)}`, + "-c", + "model_providers.OpenAI.wire_api=\"responses\"", + "-c", + "model_providers.OpenAI.requires_openai_auth=true" + ]; +} + +function tomlString(value) { + return JSON.stringify(String(value ?? "")); +} + +export function createCodexAppServerJsonLineClient({ command = DEFAULT_CODEX_STDIO_COMMAND, args = null, env = process.env, cwd = repoRoot, onNotification = null } = {}) { + const spawnArgs = Array.isArray(args) ? args : codexAppServerArgs(env); + const child = spawn(command, spawnArgs, { cwd, env, stdio: ["pipe", "pipe", "pipe"] @@ -1526,20 +1577,24 @@ function createAppServerTurnState({ traceRecorder, session } = {}) { if (method === "error") { const error = extractAppServerRecord(params?.error); const message = typeof error?.message === "string" ? error.message : "Codex app-server error"; + const willRetry = params?.willRetry === true; + setThreadId(extractAppServerString(params, "threadId")); + setTurnId(extractAppServerString(params, "turnId")); traceRecorder?.append({ - type: "error", - status: "failed", - label: "app-server:error", - errorCode: "codex_stdio_failed", + type: willRetry ? "provider_retry" : "error", + status: willRetry ? "retrying" : "failed", + label: willRetry ? "app-server:retrying" : "app-server:error", + errorCode: willRetry ? "codex_stdio_provider_retry" : "codex_stdio_failed", message: redactText(message), sessionId: session?.sessionId, sessionStatus: session?.status, turn: session?.turn, threadId, turnId, - waitingFor: "user-retry" + waitingFor: willRetry ? "turn/completed" : "user-retry", + terminal: !willRetry }); - finish("failed", message); + if (!willRetry) finish("failed", message); return; } if (method === "turn/completed") { @@ -3615,7 +3670,7 @@ function codexHomeContractState(codexHomeInfo, codexHome, codexHomeFiles = null) } function childProcessEnv(env = process.env) { - const noProxy = mergeNoProxy(env.NO_PROXY, env.no_proxy, CODEX_CHILD_NO_PROXY_REQUIRED); + const noProxy = mergeNoProxy(env.NO_PROXY, env.no_proxy, CODEX_CHILD_NO_PROXY_REQUIRED, codexProviderNoProxyEntries(env)); return { PATH: Object.hasOwn(env, "PATH") ? env.PATH : process.env.PATH || "/usr/local/bin:/usr/bin:/bin", HOME: env.HOME || process.env.HOME || os.homedir(), @@ -3651,6 +3706,7 @@ function childProcessEnvState(env = process.env) { const sourceForbiddenPresent = CODEX_CHILD_STRIPPED_ENV_KEYS.filter((key) => hasEnvValue(env, key)); const noProxyEntries = splitNoProxy(child.NO_PROXY); const requiredMissing = CODEX_CHILD_NO_PROXY_REQUIRED.filter((entry) => !noProxyEntries.includes(entry.toLowerCase())); + const providerNoProxy = codexProviderNoProxyEntries(env); return { status: inheritedForbidden.length === 0 && requiredMissing.length === 0 ? "ready" : "blocked", forbiddenSourceEnvKeysPresent: sourceForbiddenPresent, @@ -3660,6 +3716,7 @@ function childProcessEnvState(env = process.env) { noProxy: { present: Boolean(child.NO_PROXY), required: [...CODEX_CHILD_NO_PROXY_REQUIRED], + provider: providerNoProxy, missing: requiredMissing, merged: noProxyEntries }, @@ -3669,6 +3726,18 @@ function childProcessEnvState(env = process.env) { }; } +function codexProviderNoProxyEntries(env = process.env) { + const baseUrl = codexAppServerProviderBaseUrl(env); + if (!baseUrl) return []; + try { + const url = new URL(baseUrl); + const hostname = url.hostname.replace(/^\[|\]$/gu, ""); + return [hostname, url.host].filter((entry, index, items) => entry && items.indexOf(entry) === index); + } catch { + return []; + } +} + function mergeNoProxy(...parts) { const entries = []; const seen = new Set();