fix: 修复 Codex stdio 真实 turn 链路

基于 HWLAB #426 修复 app-server retry 通知误判，并从 repo-owned DEV Responses env 派生 Codex provider base_url，避免子进程继续使用过期挂载配置。验证真实 completed reply 仅记录非敏感状态和长度，不打印 assistant 正文或 Secret。
2026-05-24 08:16:24 +00:00
parent 3b6a723760
commit 4415db9696
2 changed files with 174 additions and 20 deletions
@@ -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();
@@ -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();