diff --git a/internal/cloud/code-agent-chat.ts b/internal/cloud/code-agent-chat.ts index 2a110c7a..0a63f71d 100644 --- a/internal/cloud/code-agent-chat.ts +++ b/internal/cloud/code-agent-chat.ts @@ -214,6 +214,7 @@ export async function handleCodeAgentChat(params = {}, options = {}) { toolName: error.toolName }) }; + if (payload.error?.diagnosis) payload.diagnosis = payload.error.diagnosis; if (payload.error?.blocker) payload.blocker = payload.error.blocker; if (payload.error?.blockers) payload.blockers = payload.error.blockers; if (error.provider) payload.provider = error.provider; @@ -924,6 +925,7 @@ async function callCodexStdioRunner({ message, conversationId, sessionId, thread idleMs: error.idleMs, lastActivityAt: error.lastActivityAt, waitingFor: error.waitingFor, + diagnosis: error.diagnosis, providerTrace: error.providerTrace, stage: error.stage, missingTools: error.missingTools, @@ -1185,6 +1187,7 @@ function normalizeChatError(error, context = {}) { const idleMs = numberErrorField(error.idleMs ?? context.idleMs); const lastActivityAt = safeIsoLike(error.lastActivityAt ?? context.lastActivityAt); const waitingFor = redactText(error.waitingFor ?? context.waitingFor ?? ""); + const diagnosis = normalizeDiagnosis(error.diagnosis ?? context.diagnosis ?? error.blocker?.diagnosis ?? null); const blocker = structuredBlocker({ code, layer: taxonomy.layer, @@ -1209,7 +1212,8 @@ function normalizeChatError(error, context = {}) { hardTimeoutMs, idleMs, lastActivityAt, - waitingFor + waitingFor, + diagnosis }); const normalized = { code, @@ -1233,6 +1237,7 @@ function normalizeChatError(error, context = {}) { idleMs, lastActivityAt, waitingFor, + diagnosis, missingConfig: safeMissingConfig(error), blockers }; @@ -1429,7 +1434,7 @@ function errorTaxonomy(code, error = {}) { }; } -function structuredBlocker({ code, layer, message, userMessage, retryable, traceId, provider, backend, runner, capabilityLevel, route, toolName, category, blockers, sessionId = null, conversationId = null, stage = null, lastEvent = null, elapsedMs = null, timeoutMs = null, hardTimeoutMs = null, idleMs = null, lastActivityAt = null, waitingFor = null }) { +function structuredBlocker({ code, layer, message, userMessage, retryable, traceId, provider, backend, runner, capabilityLevel, route, toolName, category, blockers, sessionId = null, conversationId = null, stage = null, lastEvent = null, elapsedMs = null, timeoutMs = null, hardTimeoutMs = null, idleMs = null, lastActivityAt = null, waitingFor = null, diagnosis = null }) { return { code, layer, @@ -1454,6 +1459,7 @@ function structuredBlocker({ code, layer, message, userMessage, retryable, trace idleMs, lastActivityAt, waitingFor, + diagnosis, blockerCodes: Array.isArray(blockers) ? blockers.map((blocker) => blocker.code).filter(Boolean) : [] }; } @@ -1485,7 +1491,8 @@ function normalizeBlockers(blockers, fallback) { hardTimeoutMs: numberErrorField(blocker.hardTimeoutMs), idleMs: numberErrorField(blocker.idleMs), lastActivityAt: safeIsoLike(blocker.lastActivityAt), - waitingFor: blocker.waitingFor ? redactText(blocker.waitingFor) : null + waitingFor: blocker.waitingFor ? redactText(blocker.waitingFor) : null, + diagnosis: normalizeDiagnosis(blocker.diagnosis ?? null) })); if (normalized.length > 0) return normalized; return [{ @@ -1498,6 +1505,32 @@ function normalizeBlockers(blockers, fallback) { }]; } +function normalizeDiagnosis(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + return { + code: value.code ? redactText(value.code) : null, + layer: value.layer ? redactText(value.layer) : null, + kind: value.kind ? redactText(value.kind) : null, + summary: value.summary ? redactText(value.summary) : null, + waitingFor: value.waitingFor ? redactText(value.waitingFor) : null, + lastActivityAt: safeIsoLike(value.lastActivityAt), + idleMs: numberErrorField(value.idleMs), + threadId: value.threadId ? redactText(value.threadId) : null, + turnId: value.turnId ? redactText(value.turnId) : null, + lastRoutedEvent: value.lastRoutedEvent && typeof value.lastRoutedEvent === "object" ? { + type: value.lastRoutedEvent.type ?? null, + status: value.lastRoutedEvent.status ?? null, + label: value.lastRoutedEvent.label ?? null, + stage: value.lastRoutedEvent.stage ?? null, + toolName: value.lastRoutedEvent.toolName ?? null, + itemId: value.lastRoutedEvent.itemId ?? null, + waitingFor: value.lastRoutedEvent.waitingFor ?? null, + exitCode: numberErrorField(value.lastRoutedEvent.exitCode), + durationMs: numberErrorField(value.lastRoutedEvent.durationMs) + } : null + }; +} + function safeMissingConfig(error) { return [ ...(Array.isArray(error.missingEnv) ? error.missingEnv : []), diff --git a/internal/cloud/code-agent-session-registry.test.ts b/internal/cloud/code-agent-session-registry.test.ts index 438decff..740ae928 100644 --- a/internal/cloud/code-agent-session-registry.test.ts +++ b/internal/cloud/code-agent-session-registry.test.ts @@ -891,7 +891,7 @@ test("Code Agent PC gateway prompt reaches Codex stdio instead of internal hardw } }); -test("Codex stdio manager rebuilds app-server client when provider profile config changes", async () => { +test("Codex stdio manager uses a per-turn app-server client and closes each client", async () => { const calls = []; const createdModels = []; const closedModels = []; @@ -942,7 +942,7 @@ test("Codex stdio manager rebuilds app-server client when provider profile confi }); assert.deepEqual(createdModels, ["deepseek-chat", "gpt-5.5"]); - assert.deepEqual(closedModels, ["deepseek-chat"]); + assert.deepEqual(closedModels, ["deepseek-chat", "gpt-5.5"]); assert.equal(calls.filter((call) => call.method === "initialize").length, 2); assert.deepEqual(calls.filter((call) => call.method === "turn/start").map((call) => call.args.model), ["deepseek-chat", "gpt-5.5"]); } finally { @@ -951,6 +951,78 @@ test("Codex stdio manager rebuilds app-server client when provider profile confi } }); +test("Codex stdio concurrent turns do not share notification handlers", async () => { + const calls = []; + const createdClients = []; + const closedClients = []; + const fakeCodex = await createFakeCodexCommand(); + const codexHome = await prepareFakeCodexHome(); + const manager = createCodexStdioSessionManager({ + idFactory: () => `ses_concurrent_${createdClients.length + 1}`, + createRpcClient: async () => { + const clientIndex = createdClients.length + 1; + createdClients.push(clientIndex); + const client = createTimedFakeAppServerClient({ + calls, + text: `reply-from-client-${clientIndex}`, + eventEveryMs: 10, + eventCount: 1, + completeAfterMs: clientIndex === 1 ? 120 : 30 + }); + const originalClose = client.close; + client.close = () => { + closedClients.push(clientIndex); + originalClose(); + }; + return client; + } + }); + 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_CODEX_SANDBOX: "danger-full-access", + HWLAB_CODE_AGENT_WORKSPACE: process.cwd(), + HWLAB_CODE_AGENT_OPENAI_BASE_URL: "http://127.0.0.1:49280/responses" + }; + + try { + const first = manager.chat({ + conversationId: "cnv_concurrent_1", + traceId: "trc_concurrent_1", + message: "first concurrent turn", + env + }); + await new Promise((resolve) => setTimeout(resolve, 5)); + const second = manager.chat({ + conversationId: "cnv_concurrent_2", + traceId: "trc_concurrent_2", + message: "second concurrent turn", + env + }); + + const [firstPayload, secondPayload] = await Promise.all([first, second]); + assert.equal(firstPayload.content, "reply-from-client-1"); + assert.equal(secondPayload.content, "reply-from-client-2"); + assert.deepEqual(createdClients, [1, 2]); + assert.deepEqual(closedClients.sort(), [1, 2]); + assert.equal(firstPayload.runnerTrace.traceId, "trc_concurrent_1"); + assert.equal(secondPayload.runnerTrace.traceId, "trc_concurrent_2"); + assert.ok(firstPayload.runnerTrace.assistantStreams?.[0]?.text.includes("reply-from-client-1")); + assert.ok(secondPayload.runnerTrace.assistantStreams?.[0]?.text.includes("reply-from-client-2")); + assert.equal(firstPayload.runnerTrace.assistantStreams?.[0]?.text.includes("reply-from-client-2"), false); + assert.equal(secondPayload.runnerTrace.assistantStreams?.[0]?.text.includes("reply-from-client-1"), false); + } finally { + await rm(fakeCodex.root, { recursive: true, force: true }); + await rm(codexHome, { recursive: true, force: true }); + } +}); + test("Code Agent Keil gateway prompt is not blocked by M3 IO intent guard", async () => { const calls = []; let providerCalled = false; @@ -2040,6 +2112,8 @@ test("Codex app-server activity timeout fails only after no new events", async ( assert.ok(payload.error.idleMs >= 30); assert.match(payload.error.lastActivityAt, /^\d{4}-\d{2}-\d{2}T/u); assert.equal(payload.error.waitingFor, "turn/completed"); + assert.equal(payload.error.diagnosis.code, "codex_kernel_stalled"); + assert.equal(payload.diagnosis.code, "codex_kernel_stalled"); assert.ok(Date.now() - started >= 30); const timeoutEvent = payload.runnerTrace.events.find((event) => event.label === "timeout:no_activity"); assert.ok(timeoutEvent); @@ -2047,6 +2121,7 @@ test("Codex app-server activity timeout fails only after no new events", async ( assert.ok(timeoutEvent.idleMs >= 30); assert.match(timeoutEvent.lastActivityAt, /^\d{4}-\d{2}-\d{2}T/u); assert.equal(timeoutEvent.waitingFor, "turn/completed"); + assert.equal(timeoutEvent.diagnosisCode, "codex_kernel_stalled"); assert.equal(payload.runnerTrace.events.some((event) => event.label === "timeout:hard_cap"), false); } finally { await rm(fakeCodex.root, { recursive: true, force: true }); @@ -2054,6 +2129,72 @@ test("Codex app-server activity timeout fails only after no new events", async ( } }); +test("Codex app-server no-progress diagnosis identifies tool-result ack stalls", async () => { + const calls = []; + const fakeCodex = await createFakeCodexCommand(); + const codexHome = await prepareFakeCodexHome(); + const manager = createCodexStdioSessionManager({ + idFactory: () => "ses_stdio_tool_ack_stall", + createRpcClient: async () => createTimedFakeAppServerClient({ + calls, + eventEveryMs: 12, + eventCount: 0, + completeAfterMs: null, + commandExecution: { + id: "cmd_stdio_tool_ack_stall", + command: ["hwpod", "D601-F103-V2:workspace:/", "ls"], + exitCode: 0, + durationMs: 123, + aggregatedOutput: "workspace listing completed\n" + } + }) + }); + 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://127.0.0.1:49280/responses" + }; + + try { + const payload = await handleCodeAgentChat( + { + conversationId: "cnv_stdio_tool_ack_stall", + traceId: "trc_stdio_tool_ack_stall", + message: "工具结果已返回但 turn 不结束" + }, + { + now: () => "2026-05-31T16:00:00.000Z", + codexStdioManager: manager, + env, + timeoutMs: 35, + hardTimeoutMs: 500 + } + ); + + validateCodeAgentChatSchema(payload); + assert.equal(payload.status, "timeout"); + assert.equal(payload.error.code, "codex_stdio_idle_timeout"); + assert.equal(payload.error.diagnosis.code, "app_server_tool_result_ack_stalled"); + assert.equal(payload.error.diagnosis.layer, "codex-kernel"); + assert.equal(payload.error.diagnosis.lastRoutedEvent.label, "item/commandExecution:completed"); + assert.equal(payload.error.blocker.diagnosis.code, "app_server_tool_result_ack_stalled"); + const timeoutEvent = payload.runnerTrace.events.find((event) => event.label === "timeout:no_activity"); + assert.ok(timeoutEvent, JSON.stringify(payload.runnerTrace.events)); + assert.equal(timeoutEvent.diagnosisCode, "app_server_tool_result_ack_stalled"); + assert.equal(timeoutEvent.diagnosis.lastRoutedEvent.label, "item/commandExecution:completed"); + } finally { + await rm(fakeCodex.root, { recursive: true, force: true }); + await rm(codexHome, { recursive: true, force: true }); + } +}); + test("Codex app-server partial assistant output without turn completion is a timeout, not completed", async () => { const calls = []; const fakeCodex = await createFakeCodexCommand(); diff --git a/internal/cloud/code-agent-trace-store.ts b/internal/cloud/code-agent-trace-store.ts index 35e39d99..3451b2fc 100644 --- a/internal/cloud/code-agent-trace-store.ts +++ b/internal/cloud/code-agent-trace-store.ts @@ -150,7 +150,7 @@ export function createCodeAgentTraceStore(options = {}) { eventLabels: events.map((event) => event.label).filter(Boolean), lastEvent, elapsedMs: elapsedMs(trace.startedAt, trace.finishedAt ?? trace.updatedAt), - waitingFor: lastWaitingFor(events) ?? assistantStreams.at(-1)?.waitingFor ?? null, + waitingFor: snapshotWaitingFor(trace, events, assistantStreams, lastEvent), runnerKind: extra.runnerKind ?? trace.meta.runnerKind ?? null, workspace: extra.workspace ?? trace.meta.workspace ?? null, sandbox: extra.sandbox ?? trace.meta.sandbox ?? null, @@ -334,6 +334,8 @@ function normalizeTraceEvent(event, { traceId, seq, now, fallbackRunnerKind } = chunk: safeText(event.chunk, 400), message: safeText(event.message, 1200), errorCode: safeText(event.errorCode, 120), + diagnosisCode: safeText(event.diagnosisCode ?? event.diagnosis?.code, 160), + diagnosis: normalizeDiagnosis(event.diagnosis), waitingFor: safeText(event.waitingFor, 220), timeoutMs: typeof event.timeoutMs === "number" ? Math.trunc(event.timeoutMs) : undefined, hardTimeoutMs: typeof event.hardTimeoutMs === "number" ? Math.trunc(event.hardTimeoutMs) : undefined, @@ -397,7 +399,7 @@ function notify(trace, event) { eventLabels: trace.events.map((item) => item.label).filter(Boolean), lastEvent: event ?? trace.events.at(-1) ?? null, elapsedMs: elapsedMs(trace.startedAt, trace.finishedAt ?? trace.updatedAt), - waitingFor: lastWaitingFor(trace.events) ?? assistantStreams.at(-1)?.waitingFor ?? null, + waitingFor: snapshotWaitingFor(trace, trace.events, assistantStreams, event ?? trace.events.at(-1) ?? null), runnerKind: trace.meta.runnerKind, workspace: trace.meta.workspace, sandbox: trace.meta.sandbox, @@ -460,6 +462,14 @@ function lastWaitingFor(events) { return null; } +function snapshotWaitingFor(trace, events, assistantStreams, lastEvent) { + if (lastEvent?.terminal === true) { + return lastEvent.waitingFor ?? null; + } + if (trace?.finishedAt) return null; + return lastWaitingFor(events) ?? assistantStreams.at(-1)?.waitingFor ?? null; +} + function elapsedMs(start, end) { const startMs = Date.parse(start ?? ""); const endMs = Date.parse(end ?? ""); @@ -467,6 +477,36 @@ function elapsedMs(start, end) { return Math.max(0, endMs - startMs); } +function normalizeDiagnosis(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const lastRoutedEvent = value.lastRoutedEvent && typeof value.lastRoutedEvent === "object" && !Array.isArray(value.lastRoutedEvent) + ? dropUndefined({ + type: safeText(value.lastRoutedEvent.type, 80), + status: safeText(value.lastRoutedEvent.status, 80), + label: safeText(value.lastRoutedEvent.label, 180), + stage: safeText(value.lastRoutedEvent.stage, 80), + toolName: safeText(value.lastRoutedEvent.toolName, 120), + itemId: safeText(value.lastRoutedEvent.itemId, 180), + waitingFor: safeText(value.lastRoutedEvent.waitingFor, 220), + exitCode: Number.isInteger(value.lastRoutedEvent.exitCode) ? value.lastRoutedEvent.exitCode : undefined, + durationMs: typeof value.lastRoutedEvent.durationMs === "number" ? Math.max(0, Math.trunc(value.lastRoutedEvent.durationMs)) : undefined + }) + : undefined; + return dropUndefined({ + code: safeText(value.code, 160), + layer: safeText(value.layer, 120), + kind: safeText(value.kind, 120), + summary: safeText(value.summary, 500), + waitingFor: safeText(value.waitingFor, 220), + lastActivityAt: safeText(value.lastActivityAt, 80), + idleMs: typeof value.idleMs === "number" ? Math.max(0, Math.trunc(value.idleMs)) : undefined, + lastActivityLabel: safeText(value.lastActivityLabel, 180), + threadId: safeText(value.threadId, 180), + turnId: safeText(value.turnId, 180), + lastRoutedEvent + }); +} + function safeText(value, limit = TEXT_LIMIT) { if (value === undefined || value === null || value === "") return undefined; const redacted = String(value).replace(SECRET_PATTERN, "[redacted]").replace(/\s+/gu, " ").trim(); diff --git a/internal/cloud/codex-stdio-session-helpers.ts b/internal/cloud/codex-stdio-session-helpers.ts index fd582c26..196ce70d 100644 --- a/internal/cloud/codex-stdio-session-helpers.ts +++ b/internal/cloud/codex-stdio-session-helpers.ts @@ -42,10 +42,11 @@ import { const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../.."); -export function appServerActivityTimeoutError({ timeoutMs, idleMs, lastActivityAt, waitingFor, threadId, turnId, partialAssistant = false }) { +export function appServerActivityTimeoutError({ timeoutMs, idleMs, lastActivityAt, waitingFor, threadId, turnId, partialAssistant = false, diagnosis = null }) { + const diagnosisSuffix = diagnosis?.code ? ` diagnosis=${diagnosis.code}.` : ""; const error = new Error(partialAssistant - ? `Codex app-server idle timeout after partial assistant output: no turn/completed for ${idleMs}ms, threshold ${timeoutMs}ms; waitingFor=${waitingFor ?? "unknown"}.` - : `Codex app-server idle timeout: no new events for ${idleMs}ms, threshold ${timeoutMs}ms; waitingFor=${waitingFor ?? "unknown"}.`); + ? `Codex app-server idle timeout after partial assistant output: no turn/completed for ${idleMs}ms, threshold ${timeoutMs}ms; waitingFor=${waitingFor ?? "unknown"}.${diagnosisSuffix}` + : `Codex app-server idle timeout: no new events for ${idleMs}ms, threshold ${timeoutMs}ms; waitingFor=${waitingFor ?? "unknown"}.${diagnosisSuffix}`); Object.assign(error, { code: "codex_stdio_idle_timeout", timeoutMs, @@ -55,13 +56,16 @@ export function appServerActivityTimeoutError({ timeoutMs, idleMs, lastActivityA threadId, turnId, partialAssistant, + diagnosis, + diagnosisCode: diagnosis?.code ?? null, stage: "app-server-turn" }); return error; } -export function appServerHardTimeoutError({ hardTimeoutMs, timeoutMs, idleMs, lastActivityAt, waitingFor, threadId, turnId }) { - const error = new Error(`Codex app-server hard timeout reached after ${hardTimeoutMs}ms; last activity was ${idleMs}ms ago; waitingFor=${waitingFor ?? "unknown"}.`); +export function appServerHardTimeoutError({ hardTimeoutMs, timeoutMs, idleMs, lastActivityAt, waitingFor, threadId, turnId, diagnosis = null }) { + const diagnosisSuffix = diagnosis?.code ? ` diagnosis=${diagnosis.code}.` : ""; + const error = new Error(`Codex app-server hard timeout reached after ${hardTimeoutMs}ms; last activity was ${idleMs}ms ago; waitingFor=${waitingFor ?? "unknown"}.${diagnosisSuffix}`); Object.assign(error, { code: "codex_stdio_hard_timeout", timeoutMs, @@ -71,6 +75,8 @@ export function appServerHardTimeoutError({ hardTimeoutMs, timeoutMs, idleMs, la waitingFor, threadId, turnId, + diagnosis, + diagnosisCode: diagnosis?.code ?? null, stage: "app-server-turn" }); return error; @@ -88,7 +94,8 @@ export function normalizeAppServerTurnResult(rawResult, state, fallback) { lastActivityAt: raw.lastActivityAt ?? state.lastActivityAt ?? null, idleMs: raw.idleMs ?? state.idleMs ?? null, waitingFor: raw.waitingFor ?? state.waitingFor ?? null, - appServerExit: raw.appServerExit ?? null + appServerExit: raw.appServerExit ?? null, + diagnosis: raw.diagnosis ?? state.diagnosis ?? null }; } @@ -114,6 +121,7 @@ export function codexStdioProviderTrace({ availability, toolName, turnResult, se turnId: turnResult?.turnId ?? null, terminalStatus: turnResult?.terminalStatus ?? null, appServerExit: turnResult?.appServerExit ?? null, + diagnosis: turnResult?.diagnosis ?? null, valuesPrinted: false }; } @@ -316,7 +324,7 @@ export function publicSession(session, { conversationId = null, reused = null } }, { reused }); } -export function blockedAcquire({ code, summary, session, timestamp, traceId, currentTrace = null }) { +export function blockedAcquire({ code, summary, session, timestamp, traceId, currentTrace = null, diagnosis = null }) { const publicEvidence = session ? publicSession(session, { timestamp, reused: true }) : { @@ -338,7 +346,8 @@ export function blockedAcquire({ code, summary, session, timestamp, traceId, cur currentTraceId: currentTrace?.traceId ?? session?.currentTraceId ?? null, currentTraceStatus: currentTrace?.status ?? null, currentTraceLastEvent: currentTrace?.lastEvent ?? null, - currentTraceUpdatedAt: currentTrace?.updatedAt ?? null + currentTraceUpdatedAt: currentTrace?.updatedAt ?? null, + diagnosis } }; } diff --git a/internal/cloud/codex-stdio-session-turn-state.ts b/internal/cloud/codex-stdio-session-turn-state.ts index 00d7b301..7f256de6 100644 --- a/internal/cloud/codex-stdio-session-turn-state.ts +++ b/internal/cloud/codex-stdio-session-turn-state.ts @@ -1,5 +1,5 @@ -import { CODEX_STDIO_FIRST_TOKEN_PROGRESS_MS } from "./codex-stdio-session.ts"; +import { CODEX_STDIO_FIRST_TOKEN_PROGRESS_MS, CODEX_STDIO_NO_PROGRESS_DIAGNOSIS_MS } from "./codex-stdio-session.ts"; import { appServerActivityTimeoutError, appServerCommandStatus, @@ -30,6 +30,11 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { let lastActivityAt = Date.now(); let lastActivityLabel = "turn-state:created"; let lastWaitingFor = "app-server-turn"; + let lastRoutedEvent = null; + let lastClientRequest = null; + let lastDiagnosis = null; + let lastProgressDiagnosisAt = 0; + let lastProgressDiagnosisKey = null; function setThreadId(value) { threadId = optionalId(value) ?? threadId; @@ -57,6 +62,7 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { function appendTrace(event = {}) { const appended = traceRecorder?.append(event); + lastRoutedEvent = compactTraceEvent(appended ?? event); observeActivity({ label: appended?.label ?? event.label ?? event.type ?? "trace:event", waitingFor: appended?.waitingFor ?? event.waitingFor ?? null @@ -93,6 +99,34 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { }); } + function appendNoProgressDiagnosisTrace(referenceNow = Date.now(), diagnosticMs = CODEX_STDIO_NO_PROGRESS_DIAGNOSIS_MS) { + if (terminal) return null; + const activity = activitySnapshot(referenceNow); + if (activity.idleMs < diagnosticMs) return null; + const diagnosis = diagnoseNoProgress("running-no-progress", activity); + const key = `${diagnosis.code}:${activity.waitingFor ?? "unknown"}`; + if (lastProgressDiagnosisKey === key && referenceNow - lastProgressDiagnosisAt < diagnosticMs) return null; + lastProgressDiagnosisKey = key; + lastProgressDiagnosisAt = referenceNow; + return appendProgressTrace({ + type: "turn", + status: "running", + label: "turn:no_progress:diagnosed", + message: diagnosis.summary, + progressOnly: true, + diagnosis, + diagnosisCode: diagnosis.code, + idleMs: activity.idleMs, + lastActivityAt: activity.lastActivityAt, + sessionId: session?.sessionId, + sessionStatus: session?.status, + turn: session?.turn, + threadId, + turnId, + waitingFor: activity.waitingFor + }); + } + function finish(status, error = null, extra = {}) { if (terminal) return; terminal = { @@ -182,9 +216,18 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { if (method === "client/request/handled") { const decision = extractAppServerString(params, "decision") ?? "handled"; const handledMethod = extractAppServerString(params, "method") ?? "client/request"; + const pending = decision === "received"; + lastClientRequest = { + method: handledMethod, + decision, + status: pending ? "pending" : "handled", + itemId: optionalId(params?.itemId ?? params?.targetItemId), + approvalId: optionalId(params?.approvalId), + observedAt: new Date().toISOString() + }; appendTrace({ type: "client_request", - status: decision === "unsupported" || decision === "denied" ? "degraded" : "completed", + status: pending ? "started" : decision === "unsupported" || decision === "denied" ? "degraded" : "completed", label: `client/request:${decision}`, toolName: handledMethod, itemId: optionalId(params?.itemId ?? params?.targetItemId), @@ -194,7 +237,7 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { turn: session?.turn, threadId, turnId, - waitingFor: "turn/completed" + waitingFor: pending ? "client/request" : "turn/completed" }); return; } @@ -300,7 +343,67 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { }); } - async function wait(timeoutMs, closedPromise, { hardTimeoutMs = null } = {}) { + function diagnoseNoProgress(kind, activity = activitySnapshot()) { + const base = { + kind, + waitingFor: activity.waitingFor, + lastActivityAt: activity.lastActivityAt, + idleMs: activity.idleMs, + lastActivityLabel, + lastRoutedEvent, + threadId, + turnId + }; + if (lastClientRequest && lastClientRequest.status !== "handled") { + return { + ...base, + code: "app_server_client_request_stalled", + layer: "client-request", + summary: `Codex app-server client request ${lastClientRequest.method} did not reach a handled decision.`, + clientRequest: lastClientRequest + }; + } + if (activity.waitingFor === "commandExecution/completed") { + return { + ...base, + code: "app_server_command_execution_stalled", + layer: "tool-execution", + summary: "Codex app-server is waiting for commandExecution/completed; the tool command or tool transport did not return a terminal item." + }; + } + if (lastRoutedEvent?.label === "item/commandExecution:completed" && activity.waitingFor === "turn/completed") { + return { + ...base, + code: "app_server_tool_result_ack_stalled", + layer: "codex-kernel", + summary: "The commandExecution item completed and was routed to the turn, but Codex app-server did not emit the following turn/completed." + }; + } + if (hasAssistantOutput() && activity.waitingFor === "turn/completed") { + return { + ...base, + code: "app_server_assistant_finalization_stalled", + layer: "codex-kernel", + summary: "Assistant output was observed, but Codex app-server did not emit turn/completed." + }; + } + if (activity.waitingFor === "assistant-message" || activity.waitingFor === "turn/completed") { + return { + ...base, + code: "codex_kernel_stalled", + layer: "codex-kernel", + summary: "Codex app-server accepted the turn, but no further routed model/tool notification arrived before the no-activity timeout." + }; + } + return { + ...base, + code: "app_server_notification_stalled", + layer: "app-server-transport", + summary: `No routed app-server notification arrived while waiting for ${activity.waitingFor ?? "unknown"}.` + }; + } + + async function wait(timeoutMs, closedPromise, { hardTimeoutMs = null, diagnosticMs = CODEX_STDIO_NO_PROGRESS_DIAGNOSIS_MS } = {}) { let activityTimer; let hardTimer; let progressTimer; @@ -309,6 +412,8 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { const now = Date.now(); const activity = activitySnapshot(now); if (activity.idleMs >= timeoutMs) { + const diagnosis = diagnoseNoProgress("activity-timeout", activity); + lastDiagnosis = diagnosis; const error = appServerActivityTimeoutError({ timeoutMs, idleMs: activity.idleMs, @@ -316,13 +421,16 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { waitingFor: activity.waitingFor, threadId, turnId, - partialAssistant: hasAssistantOutput() + partialAssistant: hasAssistantOutput(), + diagnosis }); appendTrace({ type: "timeout", status: "failed", label: "timeout:no_activity", errorCode: error.code, + diagnosis, + diagnosisCode: diagnosis.code, message: error.message, timeoutMs, idleMs: activity.idleMs, @@ -345,6 +453,7 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { const progressPromise = new Promise(() => { const tick = () => { appendFirstTokenProgressTrace(); + appendNoProgressDiagnosisTrace(Date.now(), diagnosticMs); progressTimer = setTimeout(tick, CODEX_STDIO_FIRST_TOKEN_PROGRESS_MS); }; progressTimer = setTimeout(tick, CODEX_STDIO_FIRST_TOKEN_PROGRESS_MS); @@ -353,6 +462,8 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { ? new Promise((_, reject) => { hardTimer = setTimeout(() => { const activity = activitySnapshot(); + const diagnosis = diagnoseNoProgress("hard-timeout", activity); + lastDiagnosis = diagnosis; const error = appServerHardTimeoutError({ hardTimeoutMs, timeoutMs, @@ -360,13 +471,16 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { lastActivityAt: activity.lastActivityAt, waitingFor: activity.waitingFor, threadId, - turnId + turnId, + diagnosis }); appendTrace({ type: "timeout", status: "failed", label: "timeout:hard_cap", errorCode: error.code, + diagnosis, + diagnosisCode: diagnosis.code, message: error.message, timeoutMs, hardTimeoutMs, @@ -396,6 +510,7 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { threadId, turnId, ...activitySnapshot(), + diagnosis: diagnoseNoProgress("transport-closed", activitySnapshot()), appServerExit: exit }; })); @@ -420,7 +535,8 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { terminalError: terminal?.terminalError ?? null, lastActivityAt: terminal?.lastActivityAt ?? activity.lastActivityAt, idleMs: terminal?.idleMs ?? activity.idleMs, - waitingFor: terminal?.waitingFor ?? activity.waitingFor + waitingFor: terminal?.waitingFor ?? activity.waitingFor, + diagnosis: terminal?.diagnosis ?? lastDiagnosis }; } @@ -434,6 +550,21 @@ export function createAppServerTurnState({ traceRecorder, session } = {}) { }; } +function compactTraceEvent(event = null) { + if (!event || typeof event !== "object") return null; + return { + type: event.type ?? null, + status: event.status ?? null, + label: event.label ?? null, + stage: event.stage ?? null, + toolName: event.toolName ?? null, + itemId: event.itemId ?? null, + waitingFor: event.waitingFor ?? null, + exitCode: Number.isInteger(event.exitCode) ? event.exitCode : null, + durationMs: Number.isFinite(event.durationMs) ? event.durationMs : null + }; +} + function appServerWaitingForMethod(method) { if (method === "client/request/handled") return "turn/completed"; if (method === "thread/started") return "turn/start"; diff --git a/internal/cloud/codex-stdio-session.ts b/internal/cloud/codex-stdio-session.ts index c465d511..d34aed71 100644 --- a/internal/cloud/codex-stdio-session.ts +++ b/internal/cloud/codex-stdio-session.ts @@ -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);