From 2ad05a604969a618cfd68aa104b66c4b647591b0 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:55:07 +0900 Subject: [PATCH] fix(server): preserve cyber-policy request logs --- src/server/chat-completions.ts | 4 +- src/server/claude-messages.ts | 4 +- src/server/index.ts | 3 +- src/server/request-log.ts | 42 ++++++++++++-- src/server/responses/policy-fallback.ts | 1 + structure/04_transports-and-sidecars.md | 4 ++ tests/chat-completions-endpoint.test.ts | 74 +++++++++++++++++++++++++ tests/claude-messages-endpoint.test.ts | 66 +++++++++++++++++++++- tests/request-log.test.ts | 38 +++++++++++++ tests/routing-policy-fallback.test.ts | 4 ++ tests/server-auth.test.ts | 72 ++++++++++++++++++++++++ 11 files changed, 301 insertions(+), 11 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 765cebc73e..34864ad834 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -25,7 +25,7 @@ import type { OcxConfig } from "../types"; import { readJsonRequestBody } from "./request-decompress"; import { addFinalRequestLog, - httpStatusForTerminalStatus, + httpStatusForRequestLogTerminal, recordFirstOutput, type RequestLogContext, type RequestLogEntry, @@ -216,7 +216,7 @@ async function handleChatCompletionsWithBudget( inboundWire: "chat", translatorBudget, ...(logIds ? { onFirstOutput: () => recordFirstOutput(logCtx, logIds.start) } : {}), - onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForTerminalStatus(status), { terminalStatus: status, closeReason: "terminal" }), + onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForRequestLogTerminal(status, logCtx), { terminalStatus: status, closeReason: "terminal" }), onNativePassthroughCancel: () => finalizeNativeLog(499, { closeReason: "client_cancel" }), }); diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index ee25632a26..766ca6a4d6 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -31,7 +31,7 @@ import { evidenceFromBody } from "../routing/request-evidence"; import { resolveWireProtocolOverride } from "./adapter-resolve"; import type { OcxConfig } from "../types"; import { readJsonRequestBody } from "./request-decompress"; -import { addFinalRequestLog, httpStatusForTerminalStatus, recordFirstOutput, type RequestLogContext, type RequestLogEntry } from "./request-log"; +import { addFinalRequestLog, httpStatusForRequestLogTerminal, recordFirstOutput, type RequestLogContext, type RequestLogEntry } from "./request-log"; import { conversationIdFromClaudeMetadata } from "./request-log-conversation"; import { responseWithDeferredRequestLog } from "./relay"; import { handleResponses } from "./responses"; @@ -738,7 +738,7 @@ async function handleClaudeMessagesWithBudget( stripClaudeMainAuthForNoncanonicalForward: true, translatorBudget, ...(logIds ? { onFirstOutput: () => recordFirstOutput(logCtx, logIds.start) } : {}), - onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForTerminalStatus(status), { terminalStatus: status, closeReason: "terminal" }), + onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForRequestLogTerminal(status, logCtx), { terminalStatus: status, closeReason: "terminal" }), onNativePassthroughCancel: () => finalizeNativeLog(499, { closeReason: "client_cancel" }), }); const response = logIds ? responseWithDeferredRequestLog(upstream, logIds.requestId, logIds.start, logCtx) : upstream; diff --git a/src/server/index.ts b/src/server/index.ts index 5d63ff4bca..8a5ffd0103 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -100,7 +100,6 @@ import { addFinalRequestLog, hydrateRequestLogsFromDisk, httpStatusForRequestLogTerminal, - httpStatusForTerminalStatus, inspectResponseLogSsePayload, nextRequestLogId, recordFirstOutput, @@ -1185,7 +1184,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server recordFirstOutput(logCtx, start), onNativePassthroughTerminal: status => { - finalizeNativePassthroughLog(httpStatusForTerminalStatus(status), { + finalizeNativePassthroughLog(httpStatusForRequestLogTerminal(status, logCtx), { terminalStatus: status, closeReason: "terminal", }); diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 2a1117a4ea..8c701f7391 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -2,8 +2,10 @@ import { existsSync, readFileSync } from "node:fs"; import type { ResponsesTerminalStatus } from "../bridge"; import { classifyError, + CYBER_POLICY_ERROR_CODE, httpStatusFromTerminalError as httpStatusFromClassifiedTerminalError, isClientClosedMessage, + isCyberPolicyCode, } from "../lib/errors"; import { CODEX_CONFIG_PATH, readRootTomlString } from "../codex/paths"; import { readCodexCatalogPath } from "../codex/catalog"; @@ -94,6 +96,8 @@ export interface RequestLogContext { upstreamError?: string; /** HTTP status derived from a terminal `response.failed` SSE payload (429/401/503/etc.). */ terminalHttpStatus?: number; + /** Recognized structured terminal code whose exact identity must survive status mapping. */ + terminalErrorCode?: typeof CYBER_POLICY_ERROR_CODE; /** Structured reason from `response.incomplete`; internal-only input to log classification. */ terminalIncompleteReason?: string; affinity?: "reused" | "new_bind" | "rebound" | "cleared"; @@ -446,12 +450,26 @@ export function recordAdapterReasoning( } } -export function requestLogErrorCode(status: number, upstreamError?: string): string | undefined { +export function requestLogErrorCode( + status: number, + upstreamError?: string, + terminalErrorCode?: string, +): string | undefined { if (status >= 200 && status < 400) return undefined; + // A structured terminal code is authoritative even when the provider message is localized, + // generic, or absent. Only preserve the one narrowly recognized policy code here: broadly + // forwarding arbitrary upstream codes would change unrelated request-log taxonomy. + if (isCyberPolicyCode(terminalErrorCode)) return CYBER_POLICY_ERROR_CODE; + const classifiedCode = upstreamError?.trim() + ? classifyError(status, "upstream_error", upstreamError).code + : undefined; // Defense in depth: mid-stream web-search aborts used to land as 502 with this message. - if (status === 499 || (upstreamError?.trim() && classifyError(status, "upstream_error", upstreamError).code === "client_closed_request")) { + if (status === 499 || classifiedCode === "client_closed_request") { return "client_closed_request"; } + // Keep the high-confidence message fallback for runtimes/providers that stripped the + // structured code before emitting response.failed. + if (classifiedCode === CYBER_POLICY_ERROR_CODE) return CYBER_POLICY_ERROR_CODE; if (status === 400 || status === 409) return "invalid_request_error"; if (status === 401) return "invalid_api_key"; if (status === 403) { @@ -719,9 +737,17 @@ function captureTerminalHttpStatus( if (json.type !== "response.failed") return; const error = json.response?.error; if (!error || typeof error !== "object") return; + const terminalCode = error.code === null || typeof error.code === "string" + ? error.code + : undefined; + if (isCyberPolicyCode(terminalCode)) { + logCtx.terminalErrorCode = CYBER_POLICY_ERROR_CODE; + } else { + delete logCtx.terminalErrorCode; + } logCtx.terminalHttpStatus = httpStatusFromTerminalError({ type: typeof error.type === "string" ? error.type : undefined, - code: error.code === null || typeof error.code === "string" ? error.code : undefined, + code: terminalCode, message: typeof error.message === "string" ? error.message : undefined, }); } @@ -777,7 +803,11 @@ export function addFinalRequestLog( const effectiveStatus = status >= 500 && logCtx.upstreamError && isClientClosedMessage(logCtx.upstreamError) ? 499 : status; - const errorCode = requestLogErrorCode(effectiveStatus, logCtx.upstreamError); + const errorCode = requestLogErrorCode( + effectiveStatus, + logCtx.upstreamError, + logCtx.terminalErrorCode, + ); // A response.failed whose classified status is 499 is still a client cancel, not an upstream // terminal failure — keep /api/logs closeReason aligned with that. const closeReason = effectiveStatus === 499 @@ -790,6 +820,10 @@ export function addFinalRequestLog( Date.now() - (logCtx.activeAttemptStartedAt ?? start), logCtx.usage, ); + // The final row and its active physical attempt describe the same terminal. Preserve the + // semantic code on both so detailed attempt telemetry cannot regress to a generic status code. + if (errorCode) logCtx.activeAttempt.errorCode = errorCode; + else delete logCtx.activeAttempt.errorCode; } const existing = finalizedUsage( logCtx.providerAdapter ?? logCtx.provider, diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index fd63b40c31..7e9cbd3c74 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -96,6 +96,7 @@ function finishFailedPolicyAttempt(logCtx: RequestLogContext, status: number): v delete logCtx.usageFromBridge; delete logCtx.upstreamError; delete logCtx.terminalHttpStatus; + delete logCtx.terminalErrorCode; delete logCtx.terminalIncompleteReason; } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 7cd9407e76..3a26f1179e 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -88,6 +88,10 @@ Translated response request-log tracking and the heartbeat relay also reuse `createSseInspector`. This keeps every client-facing SSE observation path on the same byte-bounded, discard-and-resynchronize frame policy and ensures the request-log, first-output, and terminal observers share one payload parse. +The inspector records a structured `response.failed` status before invoking the +terminal observer. Native Responses, Chat Completions, Claude Messages, and WebSocket +request logs must therefore finalize through the context-aware terminal mapper; recognized +`cyber_policy` terminals stay `400 / cyber_policy` rather than collapsing to a generic 502. ## Standalone Search and exact account selectors diff --git a/tests/chat-completions-endpoint.test.ts b/tests/chat-completions-endpoint.test.ts index d0ad8b0120..eea0719eb3 100644 --- a/tests/chat-completions-endpoint.test.ts +++ b/tests/chat-completions-endpoint.test.ts @@ -887,6 +887,80 @@ test("POST /v1/chat/completions finalizes native passthrough request logs", asyn } }); +test("POST /v1/chat/completions logs native cyber terminals as 400 cyber_policy", async () => { + const { clearRequestLogsForTests, getRequestLogEntries } = await import("../src/server/request-log"); + clearRequestLogsForTests(); + const upstream = Bun.serve({ + port: 0, + fetch() { + return new Response([ + "event: response.failed", + `data: ${JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { type: "invalid_request_error", code: "cyber_policy", message: "blocked" }, + }, + })}`, + "", + "", + ].join("\n"), { headers: { "Content-Type": "text/event-stream" } }); + }, + }); + const originalFetch = globalThis.fetch; + globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const requestUrl = typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + const url = new URL(requestUrl); + const prefix = "/backend-api/codex"; + if (url.hostname === "chatgpt.com" && url.pathname.startsWith(prefix)) { + return originalFetch(new URL(`${url.pathname.slice(prefix.length)}${url.search}`, upstream.url), init); + } + return originalFetch(input, init); + }) as typeof fetch; + saveConfig({ + port: 0, + defaultProvider: "openai", + providers: { + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + }, + } as OcxConfig); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { + "content-type": "application/json", + authorization: ["Bear" + "er", "caller-direct-token"].join(" "), + }, + body: JSON.stringify({ + model: "gpt-test", + stream: true, + messages: [{ role: "user", content: "hi" }], + }), + }); + expect(response.status).toBe(200); + expect(await response.text()).toContain("blocked"); + const entry = getRequestLogEntries().findLast(e => e.inboundProtocol === "chat"); + expect(entry).toMatchObject({ + status: 400, + errorCode: "cyber_policy", + terminalStatus: "failed", + closeReason: "terminal", + upstreamError: "blocked", + }); + } finally { + await server.stop(true); + await upstream.stop(true); + globalThis.fetch = originalFetch; + clearRequestLogsForTests(); + } +}); + test("responsesSseToChatCompletionsSse reconciles done-frame final arguments (last-write-wins)", async () => { const { responsesSseToChatCompletionsSse, collectChatCompletion } = budgetedChatOutbound(await import("../src/chat/outbound")); diff --git a/tests/claude-messages-endpoint.test.ts b/tests/claude-messages-endpoint.test.ts index 2d751c26d2..6e5bd89f09 100644 --- a/tests/claude-messages-endpoint.test.ts +++ b/tests/claude-messages-endpoint.test.ts @@ -7,7 +7,11 @@ import { join } from "node:path"; import { saveConfig } from "../src/config"; import { createAnthropicAdapter } from "../src/adapters/anthropic"; import { clearableDeadline } from "../src/lib/abort"; -import type { RequestLogContext } from "../src/server/request-log"; +import { + clearRequestLogsForTests, + getRequestLogEntries, + type RequestLogContext, +} from "../src/server/request-log"; import { startServer } from "../src/server"; import { ownedServiceHomeInspection } from "./helpers/owned-service-home-inspection"; import { @@ -687,6 +691,66 @@ test("native openai-responses route carries prompt_cache_key + synthesized sessi } }); +test("native openai-responses Claude route logs cyber terminals as 400 cyber_policy", async () => { + clearRequestLogsForTests(); + const upstream = Bun.serve({ + port: 0, + fetch() { + return new Response([ + "event: response.failed", + `data: ${JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { type: "invalid_request_error", code: "cyber_policy", message: "blocked" }, + }, + })}`, + "", + "", + ].join("\n"), { headers: { "Content-Type": "text/event-stream" } }); + }, + }); + saveConfig({ + port: 0, + defaultProvider: "native", + providers: { + native: { + adapter: "openai-responses", + baseUrl: `${upstream.url.toString().replace(/\/$/, "")}/v1`, + authMode: "forward", + allowPrivateNetwork: true, + }, + }, + } as OcxConfig); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/messages", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "native/gpt-test", + max_tokens: 128, + stream: true, + messages: [{ role: "user", content: "hi" }], + }), + }); + expect(response.status).toBe(200); + expect(await response.text()).toContain("blocked"); + const entry = getRequestLogEntries().findLast(e => e.surface === "claude"); + expect(entry).toMatchObject({ + status: 400, + errorCode: "cyber_policy", + terminalStatus: "failed", + closeReason: "terminal", + upstreamError: "blocked", + }); + } finally { + await server.stop(true); + await upstream.stop(true); + clearRequestLogsForTests(); + } +}); + test("custom forward openai-responses route never receives the main ChatGPT credential", async () => { writeFileSync(join(isolatedCodexHome!.path, "auth.json"), JSON.stringify({ tokens: { access_token: "main-secret-must-not-leave", account_id: "main-account-must-not-leave" }, diff --git a/tests/request-log.test.ts b/tests/request-log.test.ts index 347724dcc4..faf0027b0f 100644 --- a/tests/request-log.test.ts +++ b/tests/request-log.test.ts @@ -550,6 +550,11 @@ describe("request log metadata", () => { expect(requestLogErrorCode(429)).toBe("rate_limit_exceeded"); expect(requestLogErrorCode(499)).toBe("client_closed_request"); expect(requestLogErrorCode(502, "client closed request during web-search")).toBe("client_closed_request"); + expect(requestLogErrorCode(400, "blocked", "cyber_policy")).toBe("cyber_policy"); + expect(requestLogErrorCode( + 502, + "This content was flagged for possible cybersecurity risk. To get authorized for security work, join the Trusted Access for Cyber program.", + )).toBe("cyber_policy"); expect(requestLogErrorCode(503)).toBe("server_is_overloaded"); expect(requestLogErrorCode(502)).toBe("upstream_server_error"); expect(requestLogErrorCode(404)).toBe("http_404"); @@ -849,6 +854,39 @@ describe("request log metadata", () => { }); }); + test("deferred SSE logging preserves structured cyber_policy status and code", async () => { + const entries: RequestLogEntry[] = []; + const failedPayload = JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { type: "invalid_request_error", code: "cyber_policy", message: "blocked" }, + }, + }); + const response = responseWithDeferredRequestLog( + new Response(new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(`data: ${failedPayload}\n\n`)); + controller.close(); + }, + }), { status: 200, headers: { "content-type": "text/event-stream" } }), + "ocx-test-cyber-policy", + Date.now(), + { model: "gpt-5.6-sol", provider: "openai" }, + entry => entries.push(entry), + ); + + await response.text(); + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ + terminalStatus: "failed", + upstreamError: "blocked", + status: 400, + errorCode: "cyber_policy", + closeReason: "terminal", + }); + }); + test("deferred SSE logging maps web-search client closes to 499 client_cancel", async () => { const entries: RequestLogEntry[] = []; const message = "client closed request during web-search"; diff --git a/tests/routing-policy-fallback.test.ts b/tests/routing-policy-fallback.test.ts index 26503e5eb7..8ba1193e1b 100644 --- a/tests/routing-policy-fallback.test.ts +++ b/tests/routing-policy-fallback.test.ts @@ -57,16 +57,19 @@ describe("policy candidate fallback", () => { const trace = policyTrace(); const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext; const seenModels: string[] = []; + const seenTerminalCodes: Array = []; const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, { runCore: async (req, _config, childLog) => { const body = await req.json() as { model: string }; seenModels.push(body.model); + seenTerminalCodes.push(childLog.terminalErrorCode); const first = seenModels.length === 1; seedAttempt(childLog, first ? "provider-a" : "provider-b", first ? "model-a" : "model-b"); if (first) { childLog.requestedModel = "policy/daily"; childLog.routeDecision = trace; + childLog.terminalErrorCode = "cyber_policy"; return new Response(JSON.stringify({ error: { message: "rate limited", type: "rate_limit_error" } }), { status: 429, headers: { "content-type": "application/json" }, @@ -80,6 +83,7 @@ describe("policy candidate fallback", () => { expect(response.status).toBe(200); expect(seenModels).toEqual(["policy/daily", "provider-b/model-b"]); + expect(seenTerminalCodes).toEqual([undefined, undefined]); expect(logCtx.requestedModel).toBe("policy/daily"); expect(logCtx.routeDecision).toBe(trace); expect(logCtx.attempts).toHaveLength(2); diff --git a/tests/server-auth.test.ts b/tests/server-auth.test.ts index 4ff03dd6fc..888fbeb4b1 100644 --- a/tests/server-auth.test.ts +++ b/tests/server-auth.test.ts @@ -3004,6 +3004,78 @@ describe("server local API auth", () => { } }); + test("passthrough SSE cyber terminal is logged as 400 cyber_policy", async () => { + if (existsSync(TEST_DIR)) rmSync(TEST_DIR, { recursive: true }); + mkdirSync(TEST_DIR, { recursive: true }); + process.env.OPENCODEX_HOME = TEST_DIR; + clearRequestLogsForTests(); + + const message = "This content was flagged for possible cybersecurity risk. To get authorized for security work, join the Trusted Access for Cyber program."; + const upstream = Bun.serve({ + port: 0, + fetch() { + return new Response([ + "event: response.failed", + `data: ${JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { type: "invalid_request_error", code: "cyber_policy", message }, + }, + })}`, + "", + "", + ].join("\n"), { headers: { "content-type": "text/event-stream" } }); + }, + }); + redirectCanonicalCodexTo(upstream.url.toString()); + saveConfig({ + port: 0, + defaultProvider: "openai", + openaiProviderTierVersion: 2, + providers: poolProviders(), + codexAccounts: [ + { id: "main", email: "main@example.test", isMain: true }, + { id: "pool-a", email: "pool@example.test", isMain: false, chatgptAccountId: "acct-pool-a" }, + ], + activeCodexAccountId: "pool-a", + } as OcxConfig); + saveCodexAccountCredential("pool-a", { + accessToken: "pool-access-token", + refreshToken: "pool-refresh-token", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "acct-pool-a", + }); + + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer inbound-main-token", + }, + body: JSON.stringify({ model: "gpt-test", input: "hello", stream: true }), + }); + + expect(response.status).toBe(200); + expect(await response.text()).toContain("response.failed"); + const logs = logsFromApiBody(await fetch(new URL("/api/logs?tail=1", server.url), { headers: managementHeaders() }).then(r => r.json())); + expect(logs.at(-1)).toMatchObject({ + status: 400, + errorCode: "cyber_policy", + terminalStatus: "failed", + closeReason: "terminal", + upstreamError: message, + attempts: [expect.objectContaining({ status: 400, errorCode: "cyber_policy" })], + }); + } finally { + await server.stop(true); + await upstream.stop(true); + clearRequestLogsForTests(); + } + }); + test("native passthrough SSE records completed usage without pool terminal tracking", async () => { if (existsSync(TEST_DIR)) rmSync(TEST_DIR, { recursive: true }); mkdirSync(TEST_DIR, { recursive: true });