From 79fa97be6e6ae974ccd31d9aeedc2da96a0baa73 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 01:57:33 -0700 Subject: [PATCH 01/16] chore(release): open dev at 2.90.0 before releasing 2.89.0 (#7371) Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> --- desktop/src-tauri/Cargo.lock | 2 +- desktop/src-tauri/Cargo.toml | 2 +- desktop/src-tauri/tauri.conf.json | 2 +- package.json | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index f7645cc7b7..5b8ff2963d 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -2645,7 +2645,7 @@ dependencies = [ [[package]] name = "opencodex-desktop" -version = "2.89.0" +version = "2.90.0" dependencies = [ "base64 0.22.1", "dbus", diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index 8c5cb958a8..0356f23c8a 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "opencodex-desktop" -version = "2.89.0" +version = "2.90.0" description = "OpenCodex desktop shell" authors = ["OpenCodex contributors"] license = "MIT" diff --git a/desktop/src-tauri/tauri.conf.json b/desktop/src-tauri/tauri.conf.json index 3a3751ddbe..c6b4aa73cc 100644 --- a/desktop/src-tauri/tauri.conf.json +++ b/desktop/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "OpenCodex", - "version": "2.89.0", + "version": "2.90.0", "identifier": "com.opencodex.desktop", "build": { "frontendDist": "../ui", diff --git a/package.json b/package.json index c1e5e08ab1..15ae9530fd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@bitkyc08/opencodex", - "version": "2.89.0", + "version": "2.90.0", "description": "Universal provider proxy for OpenAI Codex & Claude Code โ€” use any LLM with Codex CLI/App/SDK and Claude Code", "type": "module", "main": "./bin/package-main.mjs", From c120eead4126247a5f8dfaab409bbc03baa674bd Mon Sep 17 00:00:00 2001 From: JUN Date: Mon, 19 Oct 2026 14:01:02 -0700 Subject: [PATCH 02/16] fix(responses): recover cold opaque state and harden replay, archive and lifecycle paths (#7388) * fix(responses): recover explicitly rejected opaque state on cold history * fix(chat): check reasoning replay budgets before copying Preflight escaped-JSON size and reserve translator budget before materializing native Chat reasoning replay. Preserve existing budget errors, caller objects and retry/key-rotation ownership; retain synthetic regressions and update accounting contracts. * docs(chat): clarify replay size checks and budget reservations Distinguish the fixed serialized-request cap from the full final-size transient translator reservation. Document changed-replay ordering and finally release without changing production or test behavior. * fix(responses): cap aggregate reasoning archive storage * fix(responses): preserve archive scan errors during cleanup * fix(responses): normalize forward base URLs with a linear scan Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(empty-completion): separate a discarded attempt's reasoning from the retry's The guard holds pre-content events across an empty-completion retry and releases them with the retry's output. With nothing between the two attempts, an unsigned reasoning delta from the retry is appended to the discarded attempt's still-open reasoning item, so the client sees one reasoning item whose text runs both attempts together ("discardedretry"). Each retry point now queues an assistant_boundary after the held events (only when something is held), which the bridge and the direct encoders already treat as "close the open items". Co-Authored-By: Claude Code * fix(lifecycle): end a tracked stream when its turn controller is aborted A drain past its deadline calls abortAndReleaseAllTurns, which aborts every controller bound to an admitted turn. trackStreamLifetime only stopped on a consumer cancel, and several callers (the adapter bridge, run-turn execution, client-encoder delivery, policy refusals, local SSE) pass a fresh AbortController that nothing else listens to. For those turns the abort was a no-op: the stream kept relaying upstream output and its source stayed open. trackStreamLifetime now listens for the controller's abort and, once tracking has started, finishes the turn, cancels the source and errors the stream. The listener is attached after registerTurn so the lease's own abort handling still runs first, and a controller registerTurn aborted synchronously (a settled lease) keeps the previous behaviour. Co-Authored-By: Claude Code * fix(lifecycle): stop the tracked stream before onDone on abort A throwing onDone callback must not leave the source open. Co-Authored-By: Claude Code * fix(responses): sanitize reasoning archive refusals on the WebSocket transport The HTTP boundary already maps typed archive failures to a sanitized 503 with a stable code. The WebSocket pre-response catch sent them as a raw 502 proxy_error frame that carried the archive path. Apply the same mapping there. --------- Co-authored-by: Ingwannu Co-authored-by: Epinephrine Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Vadym O Co-authored-by: Claude Code --- .../docs/reference/configuration/server.md | 18 +- scripts/test-layout/layout.json | 6 +- src/adapters/openai-chat/passthrough.ts | 41 ++- src/adapters/openai-chat/reasoning-replay.ts | 111 ++++++- src/adapters/openai-responses/transport.ts | 14 +- src/responses/reasoning-retention.ts | 229 ++++++++++++- src/server/chat-native.ts | 7 +- src/server/index/serve-options.ts | 10 +- src/server/index/websocket-handler.ts | 10 + src/server/lifecycle.ts | 18 +- .../responses/combo-stream-preflight.ts | 9 +- src/server/responses/core-opaque-recovery.ts | 31 +- .../responses/empty-completion-guard.ts | 8 + .../responses/reasoning-archive-error.ts | 21 ++ structure/data-planes/inbound-compat.md | 8 + .../ADR-7369-cold-opaque-state-recovery.md | 15 + structure/providers-and-adapters.md | 2 +- structure/transports/byte-accounting.md | 8 +- .../transports/native-reasoning-retention.md | 18 + structure/transports/responses-failover.md | 2 +- structure/transports/responses.md | 2 +- tests/fixtures/test-layout-expected.json | 6 +- .../chat-native-reasoning-scope.test.ts | 97 +++++- .../responses/empty-completion-guard.test.ts | 23 +- .../responses/reasoning-archive-http.test.ts | 113 +++++++ tests/responses/reasoning-archive-ws.test.ts | 118 +++++++ .../reasoning-retention-quota.test.ts | 314 ++++++++++++++++++ tests/responses/reasoning-retention.test.ts | 14 +- ...sponses-azure-opaque-blob-recovery.test.ts | 6 +- .../responses-cold-opaque-recovery.test.ts | 114 +++++++ ...esponses-forward-transport-baseurl.test.ts | 46 +++ tests/service/shutdown-drain.test.ts | 19 ++ 32 files changed, 1392 insertions(+), 66 deletions(-) create mode 100644 src/server/responses/reasoning-archive-error.ts create mode 100644 structure/decisions/ADR-7369-cold-opaque-state-recovery.md create mode 100644 tests/responses/reasoning-archive-http.test.ts create mode 100644 tests/responses/reasoning-archive-ws.test.ts create mode 100644 tests/responses/reasoning-retention-quota.test.ts create mode 100644 tests/responses/responses-cold-opaque-recovery.test.ts create mode 100644 tests/responses/responses-forward-transport-baseurl.test.ts diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 08bc166c3b..ae044e3faa 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -836,7 +836,23 @@ Failed compaction returns no replacement history. The percentage/token limits bound reasoning returned in context, **not cumulative archive storage**. Each archive contains only readable reasoning admitted in that request; request admission is bounded by `maxInboundBodyBytes` (256 MiB by default, configurable up to 512 MiB), plus small block separators. -There is **no aggregate archive byte/file limit, automatic expiry, or background archive cleanup**. +The archive directory has a separate compiled-in ceiling of **512 MiB and 256 files**, counting +committed archives, provisional remnants, and other regular files in that directory. A bounded +streaming scan stops once refusal is certain; nonregular or uninspectable entries refuse closed. +A sibling SQLite transaction serializes the scan, exclusive creation, write, and close across +cooperating writers. The capacity check runs **before a new archive payload is written**; +acquiring the lock and recording local ownership can still write lock or metadata files. +This ceiling is not a physical free-space guarantee and does not enforce writes by older builds +or external tools. There is **no automatic expiry or background archive cleanup**. + +If the archive is full, routed compaction refuses the new payload with HTTP 503 +`reasoning_archive_quota_exceeded`: review and manually remove older archives before retrying +the unchanged request. HTTP 503 `reasoning_archive_busy` means another writer holds the lock; +wait for it to finish and retry. HTTP 503 `reasoning_archive_unavailable` requires checking +directory permissions and storage health. These responses do not expose filesystem paths or +raw causes and do not invent a Retry-After delay. A refused archive does not truncate reasoning +or delete earlier files, and failed compaction returns no replacement history. + Successful archives remain until you manually delete their files, including after disabling retention or resetting the setting. They are plaintext, contain potentially sensitive conversation reasoning, and share the resolved OpenCodex home of that proxy instance. Deleting an archive makes any existing diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 1c6e4ff4d1..ab3bed226d 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1333,7 +1333,7 @@ "openai-chat-video-part.test.ts": "adapters/openai", "openai-provider-option-e2e.test.ts": "adapters/openai", "openai-provider-option-migration.test.ts": "adapters/openai", "openai-provider-option-startup.test.ts": "adapters/openai", "openai-provider-option-tooling.test.ts": "adapters/openai", "openai-provider-option.test.ts": "adapters/openai", - "responses-forward-client-headers.test.ts": "responses", + "responses-forward-client-headers.test.ts": "responses", "responses-forward-transport-baseurl.test.ts": "responses", "openai-responses-passthrough.test.ts": "responses", "responses-native-item-extensions.test.ts": "responses", "responses-native-item-extensions-compact.test.ts": "responses", @@ -1466,7 +1466,7 @@ "reasoning-effort-summary-default.test.ts": "responses", "reasoning-effort.test.ts": "codex-integration", "reasoning-envelope.test.ts": "responses", - "reasoning-retention.test.ts": "responses", "reasoning-retention-routing.test.ts": "responses", "reasoning-retention-v2.test.ts": "responses", "reasoning-metadata.test.ts": "codex-integration", + "reasoning-retention.test.ts": "responses", "reasoning-retention-quota.test.ts": "responses", "reasoning-archive-http.test.ts": "responses", "reasoning-archive-ws.test.ts": "responses", "reasoning-retention-routing.test.ts": "responses", "reasoning-retention-v2.test.ts": "responses", "reasoning-metadata.test.ts": "codex-integration", "reasoning-replay-identity.test.ts": "adapters", "reasoning-replay-robustness.test.ts": "adapters", "reasoning-replay-scope-source.test.ts": "lib", @@ -1565,7 +1565,7 @@ "responses-muse-tool-name-alias.test.ts": "responses", "responses-muse-tool-choice.test.ts": "responses", "responses-native-main-refresh.test.ts": "responses", - "responses-opaque-blob-recovery.test.ts": "responses", + "responses-opaque-blob-recovery.test.ts": "responses", "responses-cold-opaque-recovery.test.ts": "responses", "responses-parser-agent-message.test.ts": "responses", "responses-parser-malformed-content.test.ts": "responses", "responses-parser.test.ts": "responses", diff --git a/src/adapters/openai-chat/passthrough.ts b/src/adapters/openai-chat/passthrough.ts index 9a06b77362..73823aaebc 100644 --- a/src/adapters/openai-chat/passthrough.ts +++ b/src/adapters/openai-chat/passthrough.ts @@ -14,8 +14,11 @@ import { modelRecordValue } from "../../reasoning-effort"; import { modelInList, type OcxProviderConfig } from "../../types"; import { chatParallelToolCallsWireValue } from "./parallel-tool-calls"; import { applyExplicitChatReasoningWirePolicy } from "./reasoning-wire"; -import { replayNativeChatReasoning } from "./reasoning-replay"; +import { planNativeChatReasoningReplay } from "./reasoning-replay"; import type { OcxReasoningReplayScopeRef } from "../../types"; +import { jsonUtf8Bytes } from "../../lib/json-byte-size"; +import { isTranslatorBudgetExceededError, TRANSLATOR_MAX_TURN_BYTES, TranslatorBudgetExceededError, + type TranslatorBudget } from "../../lib/translator-budget"; const CHAT_PASSTHROUGH_FIELDS = [ "audio", @@ -58,6 +61,8 @@ export function buildOpenAIChatPassthroughRequest( fastMode?: boolean, providerName?: string, replayScope?: OcxReasoningReplayScopeRef, + maxSerializedBytes = TRANSLATOR_MAX_TURN_BYTES, + translatorBudget?: TranslatorBudget, ): AdapterRequest { const { url, headers, hasCredential } = openAIChatTransport(provider); @@ -79,8 +84,8 @@ export function buildOpenAIChatPassthroughRequest( for (const field of CHAT_PASSTHROUGH_FIELDS) { if (rawBody[field] !== undefined) body[field] = rawBody[field]; } - body.messages = replayNativeChatReasoning(body.messages, provider, modelId, replayScope, - Array.isArray(rawBody.tools) && rawBody.tools.length > 0); + const replayPlan = planNativeChatReasoningReplay(body.messages, provider, modelId, replayScope, + Array.isArray(rawBody.tools) && rawBody.tools.length > 0, maxSerializedBytes); if (protectGlmSummaryBudget(body, provider.baseUrl, body.reasoning_effort)) body.reasoning_effort = "low"; const rawEfforts = modelRecordValue(provider.modelReasoningEfforts, modelId) ?? provider.reasoningEfforts; const reasoningDisabled = modelInList(provider.noReasoningModels, modelId) || rawEfforts?.length === 0; @@ -161,7 +166,35 @@ export function buildOpenAIChatPassthroughRequest( body.stream_options = rawBody.stream_options; } - const bodyJson = JSON.stringify(applyGithubCopilotContextTier(body, provider, modelId, providerName)); + const transformedBody = () => applyGithubCopilotContextTier(body, provider, modelId, providerName); + // The binding constraint is the final size base + net, so a shrinking replay may legitimately + // measure a base above the cap; only a base above max - net makes the result unserializable. + let baseBytes = 0; + if (replayPlan.hasChanges) { + try { + baseBytes = jsonUtf8Bytes(transformedBody(), maxSerializedBytes - replayPlan.serializedBytes); + } catch (error) { + if (isTranslatorBudgetExceededError(error)) { + throw new TranslatorBudgetExceededError("request_copies", maxSerializedBytes); + } + throw error; + } + } + if (replayPlan.serializedBytes > maxSerializedBytes - baseBytes) { + throw new TranslatorBudgetExceededError("request_copies", maxSerializedBytes); + } + const preSerialization = translatorBudget?.reserveTransient(baseBytes + replayPlan.serializedBytes, { + kind: "request_copies", + }); + let bodyJson: string; + try { + if (replayPlan.hasChanges) { + body.messages = replayPlan.apply(); + } + bodyJson = JSON.stringify(transformedBody()); + } finally { + preSerialization?.release(); + } if (isDebugEnabled()) { let host = "upstream"; diff --git a/src/adapters/openai-chat/reasoning-replay.ts b/src/adapters/openai-chat/reasoning-replay.ts index 1d6ca645c1..327ee4c153 100644 --- a/src/adapters/openai-chat/reasoning-replay.ts +++ b/src/adapters/openai-chat/reasoning-replay.ts @@ -1,10 +1,14 @@ import { modelInList, type OcxProviderConfig, type OcxReasoningReplayScopeRef } from "../../types"; import { peekReasoningForCall } from "../../responses/reasoning-replay-cache"; +import { jsonStringPartsUtf8Bytes, jsonUtf8Bytes } from "../../lib/json-byte-size"; +import { isTranslatorBudgetExceededError, TranslatorBudgetExceededError } from "../../lib/translator-budget"; type Rec = Record; export const isChatRecord = (value: unknown): value is Rec => value !== null && typeof value === "object" && !Array.isArray(value); +const REASONING_CONTENT_FIELD_BYTES = Buffer.byteLength('"reasoning_content":', "utf8"); + const isDeepSeekThinkingModel = (modelId: string): boolean => /(?:^|\/)deepseek-(?:v\d+(?:\.\d+)*-(?:flash|pro)|flash|pro)(?:$|[-_:])/i.test(modelId); @@ -25,28 +29,109 @@ export function chatReasoningReplayPolicy(provider: OcxProviderConfig, modelId: }; } -/** Repair missing native history without mutating caller bodies used by retries/combo siblings. */ -export function replayNativeChatReasoning( +interface ReplayMessagePlan { + message: Rec; + parts: string[]; +} + +export interface NativeChatReasoningReplayPlan { + /** Net byte change in the final escaped JSON object representation; may be zero or negative. */ + readonly serializedBytes: number; + /** Whether apply() rewrites any message, independent of the net byte change. */ + readonly hasChanges: boolean; + apply(): unknown[]; +} + +/** Preflight escaped replay growth before allocating the copied messages or serialized body. */ +export function planNativeChatReasoningReplay( messages: unknown, provider: OcxProviderConfig, modelId: string, scope?: OcxReasoningReplayScopeRef, hasTools = false, -): unknown { + maxSerializedBytes = Number.MAX_SAFE_INTEGER, +): NativeChatReasoningReplayPlan { const policy = chatReasoningReplayPolicy(provider, modelId); if (!policy.preserve || !Array.isArray(messages) - || modelInList(provider.reasoningDetailsModels, modelId)) return messages; - return messages.map(message => { + || modelInList(provider.reasoningDetailsModels, modelId)) { + return { serializedBytes: 0, hasChanges: false, apply: () => messages as unknown[] }; + } + const plans: ReplayMessagePlan[] = []; + let serializedBytes = 0; + for (const message of messages) { if (!isChatRecord(message) || message.role !== "assistant" - || (typeof message.reasoning_content === "string" && message.reasoning_content.length > 0)) return message; + || (typeof message.reasoning_content === "string" && message.reasoning_content.length > 0)) continue; const calls = Array.isArray(message.tool_calls) ? message.tool_calls : []; // DeepSeek's tools contract covers all prior assistant turns, even plain answers. // Those turns have no call id for original replay; only the configured fallback applies. - if (calls.length === 0 && !(hasTools && isDeepSeekThinkingModel(modelId))) return message; - const cached = calls.filter(isChatRecord) - .map(tool => typeof tool.id === "string" ? peekReasoningForCall(tool.id, scope) : undefined) - .filter((text): text is string => typeof text === "string" && text.length > 0); - const reasoning = cached.length > 0 ? [...new Set(cached)].join("\n") : policy.placeholder ? " " : undefined; - return reasoning === undefined ? message : { ...message, reasoning_content: reasoning }; - }); + if (calls.length === 0 && !(hasTools && isDeepSeekThinkingModel(modelId))) continue; + const cached = new Set(); + for (const tool of calls) { + if (!isChatRecord(tool) || typeof tool.id !== "string") continue; + const text = peekReasoningForCall(tool.id, scope); + if (text) cached.add(text); + } + const parts = cached.size > 0 + ? [...cached] + : policy.placeholder ? [" "] : []; + if (parts.length === 0) continue; + + // The envelope delta around the joined value is exact for the plain JSON the Chat + // ingress produces: an absent or skipped key adds the reasoning_content field bytes + // plus one separator when another field serializes, while an already-serialized key + // ("" included) keeps its position and commas, so only the value size changes. + const oldReasoning = message.reasoning_content; + try { + const envelopeDelta = oldReasoning === undefined + ? REASONING_CONTENT_FIELD_BYTES + + (Object.keys(message).some(key => message[key] !== undefined) ? 1 : 0) + : -jsonUtf8Bytes(oldReasoning, Number.MAX_SAFE_INTEGER); + const fragments: string[] = []; + for (let index = 0; index < parts.length; index++) { + if (index > 0) fragments.push("\n"); + fragments.push(parts[index]!); + } + const valueBytes = jsonStringPartsUtf8Bytes( + fragments, + maxSerializedBytes - serializedBytes - envelopeDelta, + ); + const added = valueBytes + envelopeDelta; + if (added > maxSerializedBytes - serializedBytes) { + throw new TranslatorBudgetExceededError("request_copies", maxSerializedBytes); + } + serializedBytes += added; + } catch (error) { + // Inner measurements cap at the remaining budget; report the caller's real limit. + if (isTranslatorBudgetExceededError(error)) { + throw new TranslatorBudgetExceededError("request_copies", maxSerializedBytes); + } + throw error; + } + plans.push({ message, parts }); + } + return { + serializedBytes, + hasChanges: plans.length > 0, + apply: () => { + if (plans.length === 0) return messages; + const byMessage = new Map(plans.map(plan => [plan.message, plan.parts])); + return (messages as unknown[]).map(message => { + if (!isChatRecord(message)) return message; + const parts = byMessage.get(message); + if (!parts) return message; + return { ...message, reasoning_content: parts.join("\n") }; + }); + }, + }; +} + +/** Repair missing native history without mutating caller bodies used by retries/combo siblings. */ +export function replayNativeChatReasoning( + messages: unknown, + provider: OcxProviderConfig, + modelId: string, + scope?: OcxReasoningReplayScopeRef, + hasTools = false, +): unknown { + return planNativeChatReasoningReplay(messages, provider, modelId, scope, hasTools).apply(); } diff --git a/src/adapters/openai-responses/transport.ts b/src/adapters/openai-responses/transport.ts index f183e78152..4250e167b0 100644 --- a/src/adapters/openai-responses/transport.ts +++ b/src/adapters/openai-responses/transport.ts @@ -23,6 +23,18 @@ export function copyResponsesClientMetadata(target: Headers, incoming: Headers, } } +/** + * Strip trailing slashes with a linear scan rather than a regex. + * + * `/\/+$/` retries candidate end positions when a long slash run is not at the end of the + * string; the scan walks each character once and produces the same result on every input. + */ +function stripTrailingSlashes(value: string): string { + let end = value.length; + while (end > 0 && value.charCodeAt(end - 1) === 47) end -= 1; + return end === value.length ? value : value.slice(0, end); +} + /** Shared Responses URL/auth/header construction; capabilities never grant auth authority. */ export function buildResponsesTransport(provider: OcxProviderConfig, incoming: Pick): { url: string; headers: Record; @@ -31,7 +43,7 @@ export function buildResponsesTransport(provider: OcxProviderConfig, incoming: P let url: string; if (provider.authMode === "forward") { const mayForwardCallerCredentials = isCanonicalOpenAiForwardProvider(provider); - const baseUrl = mayForwardCallerCredentials ? CODEX_FORWARD_BASE_URL : provider.baseUrl.replace(/\/+$/, ""); + const baseUrl = mayForwardCallerCredentials ? CODEX_FORWARD_BASE_URL : stripTrailingSlashes(provider.baseUrl); url = `${baseUrl}/responses`; if (provider.headers) Object.assign(headers, provider.headers); const runtimeProvider = provider as { diff --git a/src/responses/reasoning-retention.ts b/src/responses/reasoning-retention.ts index 4cdb581b95..09ea1c7a6e 100644 --- a/src/responses/reasoning-retention.ts +++ b/src/responses/reasoning-retention.ts @@ -4,14 +4,21 @@ * Provider ciphertext and signatures are never included in the portable retained items. */ +import { Database } from "bun:sqlite"; import { expandRetainedCompactionInput } from "./retained-compaction"; -import { closeSync, mkdirSync, openSync, rmSync, writeFileSync } from "node:fs"; -import { join } from "node:path"; +import { chmodSync, closeSync, lstatSync, mkdirSync, openSync, opendirSync, rmSync, writeFileSync } from "node:fs"; +import { basename, dirname, join } from "node:path"; +import { getConfigDir } from "../config/paths"; +import { recordOwnedConfigPath } from "../lib/config-ownership"; +import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; import { estimateTokens } from "../lib/token-estimate"; import { decodeReasoningEnvelope } from "./reasoning-envelope"; export const DEFAULT_REASONING_RETENTION_PERCENT = 20; export const DEFAULT_MAX_REASONING_TOKENS = 100_000; +/** Compiled-in aggregate ceiling for the on-disk reasoning archive directory. */ +export const DEFAULT_ARCHIVE_MAX_BYTES = 512 * 1024 * 1024; +export const DEFAULT_ARCHIVE_MAX_FILES = 256; export interface ReasoningRetentionOptions { /** Directory that receives archived reasoning files. */ @@ -24,6 +31,10 @@ export interface ReasoningRetentionOptions { maxTokens?: number; /** Clock injection for tests. */ now?: number; + /** Aggregate archive byte ceiling. Defaults to 512 MiB; smaller values are a test seam. */ + archiveMaxBytes?: number; + /** Aggregate archive file-count ceiling. Defaults to 256; smaller values are a test seam. */ + archiveMaxFiles?: number; } export interface ReasoningArchive { @@ -100,6 +111,183 @@ function retentionTokenCap(value: number | undefined): number { : DEFAULT_MAX_REASONING_TOKENS; } +/** Aggregate-capacity refusal: the archive directory is at or over its compiled-in ceiling. */ +export class ReasoningArchiveQuotaError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "ReasoningArchiveQuotaError"; + } +} + +/** The sibling archive lock is held by another cooperating writer. */ +export class ReasoningArchiveBusyError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "ReasoningArchiveBusyError"; + } +} + +/** Unsafe or uninspectable archive state; the attempt is refused without touching anything. */ +export class ReasoningArchiveError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "ReasoningArchiveError"; + } +} + +const ARCHIVE_SEPARATOR = "\n\n---\n\n"; +const ARCHIVE_SEPARATOR_BYTES = Buffer.byteLength(ARCHIVE_SEPARATOR, "utf8"); +const ARCHIVE_LOCK_SIDECARS = ["-journal", "-wal", "-shm"] as const; + +function archiveCap(value: number | undefined, fallback: number): number { + return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : fallback; +} + +/** lstat that treats a vanished path as absent and refuses anything it cannot classify. */ +function archiveLstat(path: string): ReturnType | undefined { + try { + return lstatSync(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw new ReasoningArchiveError(`reasoning archive path is not inspectable: ${path}`, { cause: error }); + } +} + +function assertRegularArchiveDirectory(archiveDir: string): void { + const entry = archiveLstat(archiveDir); + if (!entry?.isDirectory() || entry.isSymbolicLink()) { + throw new ReasoningArchiveError(`reasoning archive directory is not a regular directory: ${archiveDir}`); + } +} + +function assertRegularArchiveLockPath(lockPath: string): void { + const entry = archiveLstat(lockPath); + if (entry !== undefined && (!entry.isFile() || entry.isSymbolicLink())) { + throw new ReasoningArchiveError(`reasoning archive lock path is not a regular file: ${lockPath}`); + } +} + +/** Exact UTF-8 byte total of the would-be archive payload, computed before any joined string exists. */ +function archivePayloadBytes(texts: string[]): number { + let bytes = texts.length > 0 ? (texts.length - 1) * ARCHIVE_SEPARATOR_BYTES : 0; + for (const text of texts) { + const next = bytes + Buffer.byteLength(text, "utf8"); + if (!Number.isSafeInteger(next)) { + throw new ReasoningArchiveError("reasoning archive payload byte size is not representable"); + } + bytes = next; + } + return bytes; +} + +interface ArchiveUsage { + files: number; + bytes: number; +} + +/** + * Non-recursive accounting of every committed, remnant and foreign file. Anything that is not a + * plain regular file refuses closed; an entry vanishing mid-scan is already gone and skipped. + */ +function scanArchiveUsage(archiveDir: string, maxBytes: number, maxFiles: number): ArchiveUsage { + let directory: ReturnType | undefined; + let files = 0; + let bytes = 0; + let inspected = 0; + let failure: ReasoningArchiveError | undefined; + try { + // One entry per read: a legacy over-capacity directory must not first become an array + // or an unbounded read-ahead buffer on this synchronous request path. + directory = opendirSync(archiveDir, { bufferSize: 1 }); + let entry; + while ((entry = directory.readSync()) !== null) { + inspected += 1; + if (!entry.isFile()) { + throw new ReasoningArchiveError(`reasoning archive entry is not a regular file: ${entry.name}`); + } + const stat = archiveLstat(join(archiveDir, entry.name)); + if (stat === undefined) { + if (inspected >= maxFiles) throw new ReasoningArchiveError("reasoning archive scan changed beyond its inspection limit"); + continue; + } + if (!stat.isFile() || stat.isSymbolicLink()) { + throw new ReasoningArchiveError(`reasoning archive entry is not a regular file: ${entry.name}`); + } + files += 1; + bytes += Number(stat.size); + if (files >= maxFiles || bytes > maxBytes) break; // refusal is already certain + if (inspected >= maxFiles) throw new ReasoningArchiveError("reasoning archive scan changed beyond its inspection limit"); + } + } catch (error) { + failure = error instanceof ReasoningArchiveError ? error + : new ReasoningArchiveError("reasoning archive directory cannot be scanned", { cause: error }); + } finally { + try { directory?.closeSync(); } catch (error) { + failure ??= new ReasoningArchiveError("reasoning archive directory could not be closed", { cause: error }); + } + } + if (failure !== undefined) throw failure; + return { files, bytes }; +} + +function isSqliteBusy(error: unknown): boolean { + const code = error && typeof error === "object" && "code" in error + ? String((error as { code?: unknown }).code) + : ""; + const message = error instanceof Error ? error.message : ""; + return code === "SQLITE_BUSY" || code === "SQLITE_LOCKED" || /database (?:is|table is) locked/i.test(message); +} + +/** + * Serialize usage scan, exclusive create, write and close for one physical archive directory + * across cooperating processes. The sibling SQLite write transaction is the lock: the OS + * releases it on process exit, so no stale-owner reclamation is needed. busy_timeout=0 keeps + * the synchronous request path fail-fast instead of freezing the event loop. + */ +function withArchiveQuotaLockSync(archiveDir: string, fn: () => T): T { + const normalized = archiveDir.replace(/[/\\]+$/, ""); + const base = basename(normalized); + if (!base || base === "." || base === "..") { + throw new ReasoningArchiveError(`reasoning archive directory is not a usable path: ${archiveDir}`); + } + const parent = dirname(normalized); + assertNotRealHomeUnderTest(parent); + const lockPath = `${normalized}.lock`; + // The lock lives beside the archive inside the config home; foreign parents need no manifest entry. + if (parent === getConfigDir()) { + recordOwnedConfigPath(parent, lockPath); + for (const suffix of ARCHIVE_LOCK_SIDECARS) recordOwnedConfigPath(parent, `${lockPath}${suffix}`); + } + assertRegularArchiveLockPath(lockPath); + let database: Database; + try { + database = new Database(lockPath, { create: true }); + } catch (error) { + throw new ReasoningArchiveError(`reasoning archive lock could not be opened: ${lockPath}`, { cause: error }); + } + try { + try { chmodSync(lockPath, 0o600); } catch { /* platform may ignore chmod */ } + assertRegularArchiveLockPath(lockPath); + database.exec("PRAGMA locking_mode = NORMAL; PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + } catch (error) { + try { database.close(); } catch { /* acquisition already failed */ } + if (error instanceof ReasoningArchiveError) throw error; + if (isSqliteBusy(error)) { + throw new ReasoningArchiveBusyError( + `reasoning archive lock is held by another writer: ${lockPath}`, + { cause: error }, + ); + } + throw new ReasoningArchiveError(`reasoning archive lock could not be acquired: ${lockPath}`, { cause: error }); + } + try { + return fn(); + } finally { + try { database.exec("ROLLBACK"); } catch { /* close below still releases the OS lock */ } + try { database.close(); } catch { /* the OS lock is released with the handle */ } + } +} + /** Separate reasoning locally before dispatch; unknown windows enforce the absolute cap. */ export function applyReasoningRetention( input: unknown, @@ -130,17 +318,36 @@ export function applyReasoningRetention( content: [{ type: "input_text", text }] })), }; - mkdirSync(options.archiveDir, { recursive: true, mode: 0o700 }); + const archiveBytes = archiveCap(options.archiveMaxBytes, DEFAULT_ARCHIVE_MAX_BYTES); + const archiveFiles = archiveCap(options.archiveMaxFiles, DEFAULT_ARCHIVE_MAX_FILES); + const incomingBytes = archivePayloadBytes(texts); const stamp = new Date(options.now ?? Date.now()).toISOString().replace(/[:.]/g, "-"); const path = join(options.archiveDir, "reasoning-" + stamp + "-" + Math.random().toString(36).slice(2, 8) + ".md"); - // Only a successful exclusive open grants ownership; an occupied path is never removed. - const fd = openSync(path, "wx", 0o600); - let failure: unknown; - try { writeFileSync(fd, texts.join("\n\n---\n\n"), "utf-8"); } catch (error) { failure = error; } - try { closeSync(fd); } catch (error) { failure ??= error; } - if (failure !== undefined) { - try { rmSync(path); } catch { /* preserve the original I/O error; cleanup is best-effort */ } - throw failure; + try { + mkdirSync(options.archiveDir, { recursive: true, mode: 0o700 }); + assertRegularArchiveDirectory(options.archiveDir); + withArchiveQuotaLockSync(options.archiveDir, () => { + assertRegularArchiveDirectory(options.archiveDir); + const usage = scanArchiveUsage(options.archiveDir, archiveBytes, archiveFiles); + if (usage.files >= archiveFiles || incomingBytes > archiveBytes - usage.bytes) { + throw new ReasoningArchiveQuotaError( + `reasoning archive is at capacity: ${usage.files} file(s), ${usage.bytes} byte(s) under ${options.archiveDir}; delete older archives to archive more`, + ); + } + // Only a successful exclusive open grants ownership; an occupied path is never removed. + const fd = openSync(path, "wx", 0o600); + let failure: unknown; + try { writeFileSync(fd, texts.join(ARCHIVE_SEPARATOR), "utf-8"); } catch (error) { failure = error; } + try { closeSync(fd); } catch (error) { failure ??= error; } + if (failure !== undefined) { + try { rmSync(path); } catch { /* preserve the original I/O error; cleanup is best-effort */ } + throw failure; + } + }); + } catch (error) { + if (error instanceof ReasoningArchiveQuotaError || error instanceof ReasoningArchiveBusyError + || error instanceof ReasoningArchiveError) throw error; + throw new ReasoningArchiveError("reasoning archive could not be written", { cause: error }); } const limitDescription = windowKnown diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index ec4a05ee75..ed52d20a85 100644 --- a/src/server/chat-native.ts +++ b/src/server/chat-native.ts @@ -45,10 +45,7 @@ import { UPSTREAM_RESET_REPLAY_REFUSED_CODE, wrapWithZeroOutputRefetch, } from "../lib/upstream-retry"; -import { - isTranslatorBudgetExceededError, - type TranslatorBudget, -} from "../lib/translator-budget"; +import { isTranslatorBudgetExceededError, type TranslatorBudget } from "../lib/translator-budget"; import { authorizeResendForRecovery } from "../lib/request-resend-gate"; import { ambiguousResendAllowanceFor, selfContainedChatBody } from "./responses/reset-replay"; import { @@ -365,6 +362,8 @@ export async function runNativeChatAttempt( config.fastMode, route.providerName, replayScope, + undefined, + translatorBudget, ); observeAdvisorNativeReasoning(req.headers, logCtx, request.body); return request; diff --git a/src/server/index/serve-options.ts b/src/server/index/serve-options.ts index 56557344e9..728c1fffe1 100644 --- a/src/server/index/serve-options.ts +++ b/src/server/index/serve-options.ts @@ -206,6 +206,7 @@ import type { WorkflowRefusalLog } from "../workflow-refusal"; import { readyProtocolMetadata } from "../../remote/protocol"; import { modelCapabilityFields } from "../models-capabilities"; import { createWebsocketHandler } from "./websocket-handler"; +import { reasoningArchiveErrorResponse } from "../responses/reasoning-archive-error"; import { withGrokSessionIdentity } from "../../grok/session-identity"; import { withCallerSessionIdentity } from "../caller-session-identity"; @@ -1347,8 +1348,9 @@ export function createServeOptions(ctx: ServeOptionsContext) { response = await handleResponsesCompact(req, config, logCtx, turnAdmissionLease, admission, { onRequestBodyRead: () => disableResponsesRequestTimeout(req, requestServer), }); - } catch { - response = formatErrorResponse(500, "server_error", "Unexpected compact request failure"); + } catch (error) { + response = reasoningArchiveErrorResponse(error) + ?? formatErrorResponse(500, "server_error", "Unexpected compact request failure"); } addFinalRequestLog(requestId, start, logCtx, response.status, response.status === 499 ? { closeReason: "client_cancel" } : undefined); @@ -1519,6 +1521,10 @@ export function createServeOptions(ctx: ServeOptionsContext) { }, }); } catch (error) { + const archiveError = reasoningArchiveErrorResponse(error); + if (archiveError) return withRequestLogId( + withCors(responseWithDeferredRequestLog(archiveError, requestId, start, logCtx), req, policy), requestId, + ); // A bounded upstream body can fail before handleResponses returns a client response. // Finalize before rethrow so accounting observes the physical send while the caller // retains the existing reset/rejection instead of receiving a synthesized response. diff --git a/src/server/index/websocket-handler.ts b/src/server/index/websocket-handler.ts index e39f2e7140..981aec3029 100644 --- a/src/server/index/websocket-handler.ts +++ b/src/server/index/websocket-handler.ts @@ -2,6 +2,7 @@ import { nativeSteeringUnavailableReason, nativeResponseControlMode, type Native import { NativeInjectionChannel } from "../responses/native-injection"; import { NativeSteeringChannel, NativeSteeringError } from "../responses/native-steering"; import { createNativeSteeringLogObserver } from "../responses/native-steering-log"; +import { reasoningArchiveErrorResponse } from "../responses/reasoning-archive-error"; import type { Server, ServerWebSocket } from "bun"; import { LIVE_SIDEBAND_UPSTREAM_OPEN_TIMEOUT_MS, @@ -396,6 +397,15 @@ export function createWebsocketHandler( })); return; } + // A local archive refusal happens before any send. Carry the same sanitized 503 + // body and stable code as HTTP so no filesystem path reaches the client frame. + const archiveError = reasoningArchiveErrorResponse(err); + if (archiveError) { + finalizeLog(archiveError.status); + const body = await archiveError.json() as { error: Record }; + sendJsonFrame(ws, buildWsErrorFrame(archiveError.status, body.error)); + return; + } finalizeLog(502); sendJsonFrame(ws, buildWsErrorFrame(502, { type: "proxy_error", diff --git a/src/server/lifecycle.ts b/src/server/lifecycle.ts index c9e6670ee1..57421fb29e 100644 --- a/src/server/lifecycle.ts +++ b/src/server/lifecycle.ts @@ -476,13 +476,25 @@ export function trackStreamLifetime( registerTurn(ac, lease); const reader = body.getReader(); let closed = false; + let streamController: ReadableStreamDefaultController | undefined; + // A drain aborts the turn's controller; many callers pass a fresh one nothing else listens to, + // so the tracked stream itself has to stop. + const onAbort = () => { + if (closed) return; + // Stop the stream before the caller's onDone runs, so a throwing callback cannot leave it open. + reader.cancel(ac.signal.reason).catch(() => {}); + try { streamController?.error(ac.signal.reason); } catch { /* already closed */ } + finish(); + }; const finish = () => { if (closed) return; closed = true; + ac.signal.removeEventListener("abort", onAbort); unregisterTurn(ac); onDone?.(); }; - return new ReadableStream({ + const tracked = new ReadableStream({ + start(controller) { streamController = controller; }, async pull(controller) { try { const { done, value } = await reader.read(); @@ -499,6 +511,10 @@ export function trackStreamLifetime( reader.cancel(reason).catch(() => {}); }, }); + // After registerTurn, so the lease's own abort listener (release the turn) runs first. A + // controller registerTurn already aborted (a settled lease) keeps its previous behaviour. + ac.signal.addEventListener("abort", onAbort, { once: true }); + return tracked; } export async function drainAndShutdown( diff --git a/src/server/responses/combo-stream-preflight.ts b/src/server/responses/combo-stream-preflight.ts index 39554a2bd8..1a7b64bb26 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -224,8 +224,13 @@ function failedTerminalResponse( : topLevelError && typeof topLevelError === "object" && !Array.isArray(topLevelError) ? topLevelError as Record : { - type: "upstream_error", - code: "upstream_server_error", + // A caller-validated flat error can carry its identity beside type:error. + // Retain it when that caller established a 400, rather than discarding the + // rejection code before the single-shot opaque-state recovery sees it. + type: terminalPayload.type === "error" && logCtx.terminalHttpStatus === 400 + ? "invalid_request_error" : "upstream_error", + code: terminalPayload.type === "error" && logCtx.terminalHttpStatus === 400 + ? (typeof terminalPayload.code === "string" ? terminalPayload.code : null) : "upstream_server_error", message: typeof terminalPayload.message === "string" ? terminalPayload.message : logCtx.upstreamError ?? "Provider stream failed before producing output", diff --git a/src/server/responses/core-opaque-recovery.ts b/src/server/responses/core-opaque-recovery.ts index 016afa5e1d..fab2957ba9 100644 --- a/src/server/responses/core-opaque-recovery.ts +++ b/src/server/responses/core-opaque-recovery.ts @@ -259,8 +259,15 @@ export function isSelfIdentifiedOpaqueBlobRejection(bodyText: string): boolean { } } - // The flat stream-error envelope carries type/message at the top level rather than under - // an error object; the same anchored identity applies there. + // Flat SSE error events carry their code/message beside type:error, without a nested + // error object. Match only the same dedicated code or exact verification identity. + if ((record.error === undefined || record.error === null) + && (record.type === "error" || record.type === "invalid_request_error") + && (record.code === "invalid_encrypted_content" || ((record.code === null || record.code === undefined) + && typeof record.message === "string" && record.message.startsWith("The encrypted content ") + && record.message.endsWith(" could not be verified. Reason: Encrypted content could not be decrypted or parsed.")))) return true; + + // The caller-mismatch identity also arrives as a flat typed error. if ( record.type === "invalid_request_error" && typeof record.message === "string" @@ -277,14 +284,14 @@ export function isSelfIdentifiedOpaqueBlobRejection(bodyText: string): boolean { /** - * Opted-in native reasoning retention forwards ciphertext another model or account minted. ChatGPT - * can refuse it inside a 200 stream -- a Codex WebSocket `error` frame or `response.failed` -- - * rather than as an HTTP 4xx, so the passthrough inspects the zero-output prelude of exactly these - * requests. Ordinary turns keep the function-output-only inspection. + * Opaque state can be refused inside a 200 stream rather than as an HTTP 4xx. The passthrough + * inspects its bounded zero-output prelude even when a restored thread has no serving record; + * successful native ciphertext remains unchanged and blobless turns do not gain inspection. */ export function retainedReasoningStreamCandidate(parsed: OcxParsedRequest, outboundBody: string | undefined): boolean { - return parsed._nativeReasoningRetainedAcrossRoute === true - && parsed._stripReasoningEncryptedContent !== true + // A restored thread has no process-local serving record. Inspect its actual outbound + // opaque state too; only an explicit zero-output rejection can authorize the rebuild. + return parsed._stripReasoningEncryptedContent !== true && outboundResponsesBodyCarriesOpaqueBlob(outboundBody); } @@ -310,7 +317,7 @@ export function retainedReasoningStreamRejection( /** - * Whether an upstream Responses 4xx authoritatively rejected opaque replay state. + * Whether an upstream Responses 4xx or explicitly identified 502 rejected opaque replay state. * * The outbound-body check is intentional: the inbound transcript may contain a proxy envelope or * compaction blob that the adapter already lowered, in which case a replay would be byte-identical. @@ -329,8 +336,10 @@ export function shouldAttemptOpaqueBlobRecovery(args: { const acceptedStatus = (args.status >= 400 && args.status < 500) || ( args.status === 502 - && outboundResponsesBodyCarriesEncryptedFunctionOutput(args.outboundBody) - && isEncryptedFunctionOutputRejection(args.errorBody) + // Gateways can project a reasoning-only rejection as 502. The existing strict + // identity classifier below still excludes generic gateway/transport failures. + && (!isEncryptedFunctionOutputRejection(args.errorBody) + || outboundResponsesBodyCarriesEncryptedFunctionOutput(args.outboundBody)) ); return acceptedStatus && adapterSpeaksResponsesWire(args.adapterName) diff --git a/src/server/responses/empty-completion-guard.ts b/src/server/responses/empty-completion-guard.ts index 969f2161fe..0b8728897c 100644 --- a/src/server/responses/empty-completion-guard.ts +++ b/src/server/responses/empty-completion-guard.ts @@ -243,6 +243,11 @@ export async function* guardEmptyCompletionEventStream( heldBytes = 0; return released; }; + // Held events from a discarded attempt are released with the retry's output, so a boundary + // keeps the retry's reasoning from being appended to the discarded attempt's reasoning item. + const markRetryBoundary = () => { + if (held.length > 0 && held.at(-1)?.type !== "assistant_boundary") held.push({ type: "assistant_boundary" }); + }; while (true) { let terminalSeen = false; @@ -279,6 +284,7 @@ export async function* guardEmptyCompletionEventStream( if (retries < maxRetries) { // Suppress the terminal: the client must never see a completed event // for a turn that produced nothing. Retry the identical turn. + markRetryBoundary(); retries += 1; try { source = await options.continuation(); @@ -300,6 +306,7 @@ export async function* guardEmptyCompletionEventStream( // malformed function call). Nothing actionable reached the client, so this is the same // replay-safe shape as an empty completion: retry the identical turn once. usage = mergeUsage(usage, event.usage); + markRetryBoundary(); retries += 1; try { source = await options.continuation(); @@ -351,6 +358,7 @@ export async function* guardEmptyCompletionEventStream( // actionable reached the client. Retry once, then surface a stated error // instead of letting the bridge reduce the turn to adapter_eof. if (!sawContent && !passthrough && retries < maxRetries) { + markRetryBoundary(); retries += 1; try { source = await options.continuation(); diff --git a/src/server/responses/reasoning-archive-error.ts b/src/server/responses/reasoning-archive-error.ts new file mode 100644 index 0000000000..a1e90a45d8 --- /dev/null +++ b/src/server/responses/reasoning-archive-error.ts @@ -0,0 +1,21 @@ +import { formatErrorResponse } from "../../bridge"; +import { ReasoningArchiveBusyError, ReasoningArchiveError, + ReasoningArchiveQuotaError } from "../../responses/reasoning-retention"; + +/** Local storage refusal before any send. Never expose filesystem messages or causes. */ +export function reasoningArchiveErrorResponse(error: unknown): Response | undefined { + let code: string; + let message: string; + if (error instanceof ReasoningArchiveQuotaError) { + code = "reasoning_archive_quota_exceeded"; + message = "Local reasoning archive is full. Review and manually remove older reasoning archives, then retry the unchanged request."; + } else if (error instanceof ReasoningArchiveBusyError) { + code = "reasoning_archive_busy"; + message = "Another local reasoning archive writer is active. Wait for it to finish, then retry the unchanged request."; + } else if (error instanceof ReasoningArchiveError) { + code = "reasoning_archive_unavailable"; + message = "Local reasoning archive is unavailable. Check archive directory permissions and local storage health before retrying the unchanged request."; + } else return undefined; + // Match the local spend-ledger storage refusal's 503 convention without guessing a delay. + return formatErrorResponse(503, "server_error", message, { code, classified: true }); +} diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index 39feeb31e8..9ce77e1db9 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -542,6 +542,14 @@ Explicit preserve lists (including `[]`) remain authoritative, and an explicit p can disable fabricated placeholders while allowing original cached reasoning. For tool-enabled DeepSeek requests the configured placeholder also repairs earlier assistant answers without tool calls. Those turns have no call-id replay entry; supplied reasoning wins. +Before copying replay into the request, native passthrough measures the exact escaped-JSON net +change and checks the final request size (`baseBytes + serializedBytes`) against the fixed +`maxSerializedBytes` cap. For a changed replay, it then reserves that full size as transient +`request_copies` against the translator budget remaining after other retained copies, before +applying replay and serializing the body. The transient reservation is released in `finally`. +Exceeding either limit returns the standard 413 `translation_buffer_limit` response; history is +never silently truncated. Key rotation and retry rebuild and charge the request after releasing +the previous body, preserving the same scope and normal replay behavior within budget. `src/server/chat-native-reasoning.ts` records upstream JSON and fragmented SSE reasoning under all parallel call ids, separately for each choice. Replay requires a sanitized `thread-id`, diff --git a/structure/decisions/ADR-7369-cold-opaque-state-recovery.md b/structure/decisions/ADR-7369-cold-opaque-state-recovery.md new file mode 100644 index 0000000000..b490e0e61d --- /dev/null +++ b/structure/decisions/ADR-7369-cold-opaque-state-recovery.md @@ -0,0 +1,15 @@ +# ADR-7369: decision recorded under "Cold opaque-state rejection recovery" + +- Purpose: recover an explicitly rejected native reasoning blob in a restored thread (#7369). +- Constraint: process-local route/retention history can be absent after restart, while real outbound + ciphertext still exists. HTTP 200 SSE can carry a failure; a projected HTTP 502 can identify the + same rejection. Neither status alone is proof of acceptance or rejection identity. +- Alternatives: strip every cold blob in advance, duplicate the recovery serializer, or reuse the + existing bounded preflight, strict rejection classifier and guarded outbound-only rebuild. +- Decision: preserve the same dedicated code/exact wording across flat and nested envelopes, + select preflight from actual outbound opaque state, and admit an explicitly identified + 502 under the existing classifier. Preserve the special function-output evidence requirement. +- Consequences: native success retains ciphertext; one sanitized rebuild follows only a proven + pre-output rejection, subject to the existing send/ambiguity guards. Generic transient retries + remain separately counted. No on-disk history change, plaintext invention or native decryption is + introduced. Stream preflight may retain its existing bounded prelude on more cold opaque turns. diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 58965fc812..afb8992417 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -200,7 +200,7 @@ rewrite rules and the routed-id settlement. | `src/combos/failover.ts` | Recognizes the exact bounded ChatGPT plan/model HTTP400 refusal in a bare message or a single `detail`/`error.message` envelope. Competing envelopes stop before code-based hops; finite evidence survives display truncation and SSE error reconstruction without retaining raw bodies. This permits another declared target before output commitment, never credential revocation or provider-wide quota evidence. | | `src/combos/request.ts` | Clones each selected combo target request and applies the existing target capability ladder: adaptive unknown targets and explicit empty ladders receive no unsupported reasoning/thinking controls, while known ladders retain per-target resolution. | | `src/adapters/openai-responses.ts`, `src/adapters/openai-responses/` | Native OpenAI/ChatGPT Responses passthrough. Compaction parsing normalizes known incomplete aliases under the [incomplete reason contract](transports/responses-wire-shapes.md#adapter-incomplete-reasons). The canonical ChatGPT adapter forces its upstream-only `stream: true` requirement without changing caller `store`; downstream JSON negotiation remains owned by the [Responses HTTP/SSE contract](transports/responses.md#responses-httpsse). | -| `src/adapters/openai-responses/transport.ts` | Shared Responses URL, configured auth and caller-header construction for ordinary requests and explicitly enabled custom native v1 compact. Its compact-handoff metadata copier uses the existing client-header allowlist, never credentials, through internal request construction and Codex auth materialization. Protocol capability never grants canonical identity; v2 trigger and native blob replay remain independently declared. | +| `src/adapters/openai-responses/transport.ts` | Shared Responses URL, configured auth and caller-header construction for ordinary requests and explicitly enabled custom native v1 compact. Its compact-handoff metadata copier uses the existing client-header allowlist, never credentials, through internal request construction and Codex auth materialization. Protocol capability never grants canonical identity; v2 trigger and native blob replay remain independently declared. On the noncanonical forward path, configured `baseUrl` trailing-slash stripping is a linear scan that preserves interior slash runs and leaves the canonical forward base and non-forward auth modes untouched; `tests/responses/responses-forward-transport-baseurl.test.ts` pins it. | | `src/responses/muse-tool-name-alias.ts` | Host-gated Meta Muse 64-char tool-name alias/restore used by the Responses passthrough. | | `src/adapters/openai-chat.ts`, `src/adapters/openai-chat/` | OpenAI-compatible Chat Completions bridge, split into leaves (`wire.ts`, `messages.ts`, `response-events.ts`, `passthrough.ts`, `parallel-tool-calls.ts`, `reasoning-wire.ts`, `serialized-tool-call-content.ts`, `tool-call-validation.ts`, `tool-call-id-remint.ts`, `tool-schema.ts`, `errors.ts`). `parallel-tool-calls.ts` owns the `parallel_tool_calls` wire value for both the translated and native builders, so the three provider states โ€” configured opt-out, configured opt-in, and the unset default that forwards only a caller's explicit `false` โ€” cannot drift between them. `reasoning-wire.ts` applies explicit gateway-object and tool-bearing effort-omission declarations to both builders; absent declarations leave native raw forwarding unchanged. Its client delivery shapes in `src/chat/outbound.ts` and `src/server/chat-native-sse.ts` relay the upstream `service_tier` echo on non-stream, folded-stream, and synthesized-SSE bodies, never inventing the key when the upstream omits it. | | `src/adapters/anthropic.ts` | Anthropic Messages bridge. A `refusal` or `content_filter` stop reason yields an explicit `incomplete` event with `retryable: false` rather than `done` with that stopReason (#4312); `max_tokens` remains `done`. For a Responses client the bridge delivers that event, and the same one from the Kiro adapter, as a terminal `response.failed` ([Adapter-to-Responses bridge](transports/responses.md#adapter-to-responses-bridge)). It is the wire that defines `tools[*].strict` and `tools[*].allowed_callers`, so a rebuilt declaration carries both: an explicit `strict: true` and any `allowed_callers` the caller declared. An absent `strict` stays absent, because the Messages inbound records it as `false` and a `false` on the wire would read as an opt-out nobody asked for. Anthropic Fast uses the native `anthropic-speed` FastWire: a set decision sends `speed: "fast"` with `fast-mode-2026-02-01` in one case-insensitively merged, deduplicated `anthropic-beta` header that preserves OAuth betas. Stream and buffered `usage.speed` echoes confirm fast or downgrade to standard; no echo leaves the request assumed. `tests/adapters/anthropic/anthropic-fast-speed.test.ts` pins the wire and echoes. Anthropic Fast is opt-in: the registry marks both Anthropic entries `fastOptIn`, and `src/providers/fast-opt-in.ts` (`providerFastSwitchOff`) keeps Fast off until `providers..fastEnabled` is `true`. An off switch is provider capability `false`, applied in the FastPolicy authority (`service-tier.ts`), `resolveModelPolicy`, and router registry enrichment, so no model-level Fast toggle, `--fast` row, or proxy-generated `speed` field is produced. Native Claude Messages passthrough still forwards a `speed` field the caller sends itself, outside the proxy Fast policy. `tests/adapters/anthropic/anthropic-fast-opt-in.test.ts` pins the default, the switch, and the management PATCH/GET. For `api.anthropic.com` only, the Claude client's typed web tools (`parsed._anthropicServerTools`) follow the function tools unchanged, minus the caller's `cache_control`; a function tool of the same name wins, so the web-search sidecar's `web_search` keeps priority and `web_fetch` is still carried. No beta header is added: both tools are GA. Each streamed `web_search_tool_result` or `web_fetch_tool_result` becomes a `web_search_call_begin`/`web_search_call_end` pair (the query of its `server_tool_use` in the same response, hit URLs and titles, or `failed` for an error result) carrying `anthropic`: that `server_tool_use` with its complete streamed input and the result block, verbatim. An orphan result, a result with a different call id, or a result whose call input cannot be parsed is dropped with no synthetic call; the surrounding text is preserved. Streamed server-tool input retains at most 4,096 characters, slicing each incoming fragment to the remaining capacity; an oversized input that cannot be recovered drops its exchange under the same rule. The bridge and the direct encoder keep both on the `web_search_call` item as `anthropic_blocks` (a fetch is an `open_page` action), and the Claude outbound renders them back unchanged, with the original `srvtoolu_` id, `encrypted_content` and `page_age`; a fetch is not counted in `web_search_requests`. A search Anthropic defers beside a client `tool_use` has no result in that response and surfaces nothing. On the next turn, `messagesToAnthropicFormat` re-inserts the client's replayed pairs (`parsed._anthropicServerReplay`) ahead of the `tool_use`, text or thinking block that followed them, for `api.anthropic.com` only. An anchor must occur exactly once across assistant history, counting string and block content alike; repeated anchors drop their exchanges and preserve the other turns. Replay requires one-to-one call/result ids and matching tool types within the same assistant turn, with the call preceding its result in the original sequence. A single forward pass tracks pending calls and rejects reversed pairs, duplicate ids and orphan results. Calls require a plain input object with a nonempty `web_search.query` or `web_fetch.url` string; malformed input, a pair without `encrypted_content`, or a pair with no following anchor stays dropped. Stored `anthropic_blocks` restore only when call id, result id, tool/result types and input object agree; malformed metadata falls back to the rebuilt pair. Citations, a deferred search and `pause_turn` are not resumed. `tests/claude-integration/claude-translated-server-tools.test.ts` pins the wire, the sidecar priority, the bridge's client stream and JSON, and the direct encoder's stream. Prompt-cache TTL: an omitted `cacheRetention` resolves through `src/adapters/anthropic/cache-retention.ts` to 1h (`ttl: "1h"`) for first-party `api.anthropic.com` (API key and OAuth pools) and to 5m for Anthropic-compatible endpoints, which do not document the 1h TTL; an explicit value wins everywhere. The native Messages passthrough upgrades caller breakpoints without a TTL to 1h under the same rule unless the caller pinned any breakpoint to another TTL. Usage carries `cache_creation.ephemeral_1h_input_tokens` as `cacheCreation1hInputTokens`, which `src/usage/cost.ts` prices at 1.6x the 5m write rate (2x base input). `tests/adapters/adapter-usage.test.ts`, `tests/adapters/anthropic/anthropic-messages-passthrough.test.ts` and `tests/claude-integration/claude-translated-cache-ttl.test.ts` pin the defaults. | diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 2c11779f71..06cba46863 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -292,7 +292,13 @@ Native Chat's optional reasoning capture (`src/server/chat-native-reasoning.ts`) reasoning and tool ids before concatenation and reserves the previous string during replacement. Its separate 256 KiB/64-call bound drops the fallback capture without interrupting the wire relay. Terminal, error and cancellation paths release collector retention; the shared replay cache owns -its own bounded text after insertion. The [native continuation contract](../data-planes/inbound-compat.md#native-chat-thinking-continuation) defines replay scope and cache-miss behavior. +its own bounded text after insertion. For a changed replay, native Chat measures the exact +escaped-JSON net change and checks the final request size (`baseBytes + serializedBytes`) +against the fixed `maxSerializedBytes` cap before expansion. It then reserves that full size as +transient `request_copies` against the remaining translator budget before applying replay and +serializing the request, releasing the reservation in `finally`. Exceeding either limit returns +413 without truncating history. The post-build request charge remains owned and released by +the existing retry/key-rotation lifecycle. The [native continuation contract](../data-planes/inbound-compat.md#native-chat-thinking-continuation) defines replay scope and cache-miss behavior. The Opus 5.5 progress normalizer in `src/adapters/anthropic-progress.ts` reserves its one-block thinking/signature buffer before concatenation, replaces signatures rather than appending, and diff --git a/structure/transports/native-reasoning-retention.md b/structure/transports/native-reasoning-retention.md index 58444b25b0..38851fb44f 100644 --- a/structure/transports/native-reasoning-retention.md +++ b/structure/transports/native-reasoning-retention.md @@ -26,3 +26,21 @@ A combo picks a target per request, but a tool loop spans several requests that When the pinned target is unavailable or fails, normal selection and failover run, and each strict serializer accepts foreign history instead of refusing it. `sanitizeReasoningInputContent` in `src/adapters/openai-responses/reasoning.ts` gives every assistant tool-call turn with no reasoning item a `reasoning_text: " "` item at its head on destinations with `requiresPlaintextReasoningReplay`. That is the Responses twin of the Chat adapter's `reasoning_content: " "` placeholder (#5421): a fixed single space, never invented reasoning. Gemini on AI Studio receives the validator sentinel for an unsigned first call, as Antigravity and Vertex already did ([Google provider](../providers/google.md#google-tool-call-thought-signature-replay)). Anthropic receives the call as an ordinary `tool_use`/`tool_result` pair with no thinking block. The opaque-only eligibility rule in `core-replay.ts` is unchanged. Coverage: `tests/server/server-combo-tool-loop-continuation.test.ts`, `tests/providers/deepseek-bare-tool-call-replay.test.ts` and `tests/adapters/google/google-ai-studio-foreign-call-signature.test.ts`. + +## Cold opaque-state rejection recovery + +`src/server/responses/core-opaque-recovery.ts` selects bounded zero-output stream preflight from +opaque state in the actual outbound body, even without a process-local serving record after restart. +The flat SSE error projector preserves a caller-validated 400 rejection identity, including its +dedicated code or exact verification wording. A successful native replay remains untouched. +Only the existing explicit rejection identities can +trigger the existing one-shot sanitized rebuild; generic 502s do not. An identified reasoning-only +502 follows that contract too. The encrypted-function-output-specific message still requires actual +outbound encrypted function output. Existing transport retries retain their separate send accounting. +Text/tool commitment, ambiguous-resend ownership, non-replayable responses, configured send limits +and the one-shot guard still prohibit unsafe replay. The outbound rebuild preserves store:false and +paired tool calls/results; this path never edits the on-disk conversation. It does not decrypt native +ciphertext or add a new plaintext reconstruction rule. `tests/responses/responses-cold-opaque-recovery.test.ts` +and `tests/responses/responses-azure-opaque-blob-recovery.test.ts` cover cold recovery and the negatives. + +> Decision record: [ADR-7369](../decisions/ADR-7369-cold-opaque-state-recovery.md) diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index 49fd6df56b..51d2a00c12 100644 --- a/structure/transports/responses-failover.md +++ b/structure/transports/responses-failover.md @@ -304,7 +304,7 @@ build both honor for canonical ChatGPT and explicitly native-capable custom dest The portable Responses summarizer in `src/adapters/openai-responses/passthrough.ts` uses `src/adapters/openai-responses/compaction-search-history.ts` to render remaining top-level `web_search_call` items, after bridge restoration, as assistant reference notes labeled untrusted historical metadata (not instructions or fetched page content). The allowlist keeps string status and action fields (type, query/queries, URL, pattern, source URL/title/type); opaque state, IDs, unknown fields and malformed values are omitted. Strings are cut at 2048 code units, lists at 20 entries, and one request carries at most 64 KiB of notes; later hosted cells collapse into a single omission note reserved inside that budget. Tool declarations stay absent, bridge-restored call/result pairs and existing messages/citations stay intact, and this portable path changes only compact summary input, never stored history or native compaction. Ordinary Codex continuation policy is separate: see [hosted-search-continuation](../providers-and-adapters.md#hosted-search-continuation-binding). A failed summary publishes no replacement history. Lite requests retain `parallel_tool_calls=false`, which the upstream requires even without tool declarations. `tests/responses/responses-compaction-override.test.ts` covers source filtering, trigger selection, config -validation, native and routed handlers, credential retention, portable summaries and replay, and combo failover. Only explicit `reasoningRetention.enabled: true` (default off, including legacy limit-only blocks) makes routed v1 `/responses/compact` and direct routed v2 `compaction_trigger` use `src/responses/reasoning-retention.ts` to withhold readable reasoning from its summarizer; exact text including its historical-context frame within the smaller configured percentage/token cap is restored as framed historical context after successful compaction, including routed v1 Combo failover to an opaque native result. Repeated compact removes that frame locally again; ordinary requests replay it as a user context message, not a fresh reasoning item. Overflow is provisionally archived once per routed turn and kept only when successful replacement history names its path, in the readable summary or a separate historical notice alongside an opaque native v1 result; failed, cancelled, or invalid outcomes remove that attempt's file without deleting earlier successful archives. The percentage/token cap bounds retained context only; each overflow file is request-admission bounded under the resolved OpenCodex homeโ€™s `reasoning-archive/`, with no aggregate quota or automatic expiry. Successful archives require manual deletion; switching off or resetting does not delete them. Crash or filesystem cleanup failure can leave provisional files. Failure yields no replacement history. Opaque compaction preserves its unchanged item and ciphertext. Routed v1 failover appends any locally withheld retained messages or an archive notice; direct native compact keeps existing behavior. Direct routed v2 remains a single compaction output item: `src/responses/retained-compaction.ts` encodes summary and framed reasoning separately as versioned `ocx2:` JSON/Base64 (encoding, not encryption). Replay expands them into separate historical messages; repeated v1/v2 compaction withholds the reasoning again. Existing `ocx1:` summaries remain readable. `src/codex/ocx-compaction-history.ts` also lowers `ocx2:` into separate summary and reasoning messages for explicit native-history recovery, so later proxy compaction still withholds reasoning. `src/server/responses/compaction-retention.ts` owns provisional archives and request-budget retention; failure or body cancellation releases the attempt. Emergency recovery preserves the structured envelope for JSON and SSE instead of flattening reasoning into the summary. `tests/responses/reasoning-retention-v2.test.ts` and `tests/responses/responses-compaction-recovery.test.ts` cover v2 replay and recovery; `tests/responses/reasoning-retention-routing.test.ts` covers these public contracts; `tests/responses/reasoning-retention.test.ts` covers envelopes, limits and archive failures. +validation, native and routed handlers, credential retention, portable summaries and replay, and combo failover. Only explicit `reasoningRetention.enabled: true` (default off, including legacy limit-only blocks) makes routed v1 `/responses/compact` and direct routed v2 `compaction_trigger` use `src/responses/reasoning-retention.ts` to withhold readable reasoning from its summarizer; exact text including its historical-context frame within the smaller configured percentage/token cap is restored as framed historical context after successful compaction, including routed v1 Combo failover to an opaque native result. Repeated compact removes that frame locally again; ordinary requests replay it as a user context message, not a fresh reasoning item. Overflow is provisionally archived once per routed turn and kept only when successful replacement history names its path, in the readable summary or a separate historical notice alongside an opaque native v1 result; failed, cancelled, or invalid outcomes remove that attempt's file without deleting earlier successful archives. The percentage/token cap bounds retained context only; each overflow file is request-admission bounded under the resolved OpenCodex homeโ€™s `reasoning-archive/`, and also shares a compiled-in aggregate ceiling under that directory: at most 512 MiB and 256 files, counted across every committed, remnant, or foreign file. The ceiling is a conservative policy bound, not a retention-duration or free-space guarantee. `src/responses/reasoning-retention.ts` enforces it inside a dedicated sibling SQLite write transaction (`reasoning-archive.lock`, `BEGIN IMMEDIATE`, `busy_timeout = 0`) that serializes each non-recursive scan, exclusive create, write, and close across cooperating writers; older builds and external OS actors stay outside that guarantee. An archive directory or lock path that is not a plain regular object, a non-regular or uninspectable directory entry, and an at- or over-capacity directory all refuse the new archive closed without deleting or mutating anything already there, surfacing distinct typed capacity, busy, and I/O errors that keep the original input intact for a client retry with no upstream send. Successful archives require manual deletion; switching off or resetting does not delete them. Crash or filesystem cleanup failure can leave provisional files, which then count toward the ceiling. Failure yields no replacement history. Opaque compaction preserves its unchanged item and ciphertext. Routed v1 failover appends any locally withheld retained messages or an archive notice; direct native compact keeps existing behavior. Direct routed v2 remains a single compaction output item: `src/responses/retained-compaction.ts` encodes summary and framed reasoning separately as versioned `ocx2:` JSON/Base64 (encoding, not encryption). Replay expands them into separate historical messages; repeated v1/v2 compaction withholds the reasoning again. Existing `ocx1:` summaries remain readable. `src/codex/ocx-compaction-history.ts` also lowers `ocx2:` into separate summary and reasoning messages for explicit native-history recovery, so later proxy compaction still withholds reasoning. `src/server/responses/compaction-retention.ts` owns provisional archives and request-budget retention; failure or body cancellation releases the attempt. Emergency recovery preserves the structured envelope for JSON and SSE instead of flattening reasoning into the summary. `tests/responses/reasoning-retention-v2.test.ts` and `tests/responses/responses-compaction-recovery.test.ts` cover v2 replay and recovery; `tests/responses/reasoning-retention-routing.test.ts` covers these public contracts; `tests/responses/reasoning-retention.test.ts` covers envelopes, limits and archive failures; `tests/responses/reasoning-retention-quota.test.ts` covers the aggregate ceiling and the archive lock. Archive accounting uses a streaming one-entry directory buffer, stops actual reads when refusal is certain, and closes on every outcome; vanished entries consume the inspection allowance and excessive churn refuses closed. At the final v1/v2 HTTP boundary in `src/server/index/serve-options.ts`, after internal recovery unwinds, `src/server/responses/reasoning-archive-error.ts` maps only typed archive failures to local-storage HTTP 503: `reasoning_archive_quota_exceeded` requests manual review/removal of older archives, `reasoning_archive_busy` requests retry after the other writer finishes, and `reasoning_archive_unavailable` requests permission/storage checks. Responses expose no raw filesystem message, path, cause or guessed Retry-After. `tests/responses/reasoning-archive-http.test.ts` exercises the production router's quota, lock, unsafe-directory and partial-write refusals with zero sends, including enabled recovery and streaming. The WebSocket transport's pre-response catch in `src/server/index/websocket-handler.ts` applies the same mapping as a 503 error frame with the same stable code; `tests/responses/reasoning-archive-ws.test.ts` covers its quota, lock and unsafe-directory refusals with zero sends. ## Ambiguous connection-reset replay boundary Three failures look alike from the outside โ€” the turn may have executed and we cannot diff --git a/structure/transports/responses.md b/structure/transports/responses.md index c39dd69a53..c07f0b9e4b 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -335,7 +335,7 @@ A replayed compaction item carries an `encrypted_content` blob only its minting native blob is relayed only when there is no known serving-identity mismatch and the destination is known to decode native blobs โ€” the canonical ChatGPT forward surface, the official OpenAI API, or a provider with the explicit `decodesNativeCompactionBlobs` capability. The destination gate alone is insufficient because more than one backend, including OpenAI and xAI, mints native blobs: a destination can decode its own blob without being able to decode the previous backend's. The same serving-identity mismatch signal therefore strips reasoning `encrypted_content` unless the narrowly scoped [native reasoning retention](native-reasoning-retention.md#runtime-boundary) policy permits it; native compaction blobs still degrade through the existing opaque-note path independently of that policy. Claude-native reasoning batches rejected blob tags into one scrub pass: only the -matching durable endpoint and generation-bearing credential retain raw and parsed carriers. Without recorded thread identity, the destination gate remains. Noncanonical forward providers receive no caller credentials and may point at any backend. On any +matching durable endpoint and generation-bearing credential retain raw and parsed carriers. Without recorded thread identity, the destination gate remains. Cold opaque preflight and identified rejection recovery follow the [cold-history contract](native-reasoning-retention.md#cold-opaque-state-rejection-recovery). Noncanonical forward providers receive no caller credentials and may point at any backend. On any other routed destination the blob also degrades to the same opaque note the bridged parser uses, because forwarding it there fails the turn and the item outlives the failure in the client transcript, repeating on every later turn including the compaction turn the proxy itself drives. With `store: false`, request sanitization strips ids from every input item, including compact-wire items, matching codex-rs (`core/src/client.rs:918-925`). Compact-wire items remain exempt from response-side field backfill. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b7c535715a..8ce291f463 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -949,7 +949,7 @@ "openai-provider-option-migration.test.ts": "adapters/openai", "openai-provider-option-startup.test.ts": "adapters/openai", "openai-provider-option-tooling.test.ts": "adapters/openai", "openai-provider-option.test.ts": "adapters/openai", - "responses-forward-client-headers.test.ts": "responses", "openai-responses-passthrough.test.ts": "responses", + "responses-forward-client-headers.test.ts": "responses", "responses-forward-transport-baseurl.test.ts": "responses", "openai-responses-passthrough.test.ts": "responses", "responses-native-item-extensions.test.ts": "responses", "responses-native-item-extensions-compact.test.ts": "responses", "responses-main-quota-observation.test.ts": "responses", @@ -1023,7 +1023,7 @@ "rate-limit-retry.test.ts": "providers", "raycast-client.test.ts": "clients", "raycast-detect.test.ts": "clients", "reasoning-effort-summary-default.test.ts": "responses", "reasoning-effort.test.ts": "codex-integration", "reasoning-envelope.test.ts": "responses", - "reasoning-retention.test.ts": "responses", "reasoning-retention-routing.test.ts": "responses", "reasoning-retention-v2.test.ts": "responses", + "reasoning-retention.test.ts": "responses", "reasoning-retention-quota.test.ts": "responses", "reasoning-archive-http.test.ts": "responses", "reasoning-archive-ws.test.ts": "responses", "reasoning-retention-routing.test.ts": "responses", "reasoning-retention-v2.test.ts": "responses", "reasoning-metadata.test.ts": "codex-integration", "reasoning-replay-identity.test.ts": "adapters", "reasoning-replay-robustness.test.ts": "adapters", "reasoning-replay-scope-source.test.ts": "lib", "redact.test.ts": "lib", "relay-eager.test.ts": "server", @@ -1093,7 +1093,7 @@ "responses-inbound-store-default.test.ts": "responses", "responses-item-id-repair.test.ts": "responses", "responses-json-events.test.ts": "responses", "responses-legacy-dotted-tool-name-repair.test.ts": "responses", "responses-muse-tool-name-alias.test.ts": "responses", "responses-muse-tool-choice.test.ts": "responses", - "responses-native-main-refresh.test.ts": "responses", "responses-opaque-blob-recovery.test.ts": "responses", + "responses-native-main-refresh.test.ts": "responses", "responses-opaque-blob-recovery.test.ts": "responses", "responses-cold-opaque-recovery.test.ts": "responses", "responses-parser-agent-message.test.ts": "responses", "responses-parser-malformed-content.test.ts": "responses", "responses-parser.test.ts": "responses", "responses-passthrough-transient-policy.test.ts": "responses", "responses-pool-401-refresh.test.ts": "responses", "responses-pool-refresh-attribution.test.ts": "responses", diff --git a/tests/responses/chat-native-reasoning-scope.test.ts b/tests/responses/chat-native-reasoning-scope.test.ts index 8d37200499..c838d1a3a0 100644 --- a/tests/responses/chat-native-reasoning-scope.test.ts +++ b/tests/responses/chat-native-reasoning-scope.test.ts @@ -1,6 +1,6 @@ import { afterEach, describe, expect, test } from "bun:test"; import { buildOpenAIChatPassthroughRequest, createOpenAIChatAdapter } from "../../src/adapters/openai-chat"; -import { chatReasoningReplayPolicy } from "../../src/adapters/openai-chat/reasoning-replay"; +import { chatReasoningReplayPolicy, planNativeChatReasoningReplay } from "../../src/adapters/openai-chat/reasoning-replay"; import { nativeChatReplayScope, bindNativeChatReplayScope, nativeChatReasoningCapture } from "../../src/server/chat-native-reasoning"; import { nativeChatSse } from "../../src/server/chat-native-sse"; import { createTranslatorBudget } from "../../src/lib/translator-budget"; @@ -80,6 +80,101 @@ describe("native Chat reasoning policy and physical replay ownership", () => { expect(reasoning(scope(), raw, "custom")).toBeUndefined(); expect(reasoning(scope(), PROVIDER, "plain-model")).toBeUndefined(); }); + test("bounds repeated escaped replay before building the expanded request", () => { + const owner = scope(); + const budget = createTranslatorBudget(); + const capture = nativeChatReasoningCapture(PROVIDER, MODEL, owner, budget); + budget.chargeRetained(60, { kind: "request_copies" }); + const secretLikeReasoning = 'quoted "reasoning" \\ and unicode ๐Ÿง '; + const secondReasoning = "second\nline"; + capture.json({ choices: [ + { message: { reasoning_content: secretLikeReasoning, tool_calls: [tool("call_a")] } }, + { message: { reasoning_content: secondReasoning, tool_calls: [tool("call_b")] } }, + ] }); + const retainedBeforeBuild = budget.snapshot().currentBytes; + const messages = Array.from({ length: 40 }, () => ({ + role: "assistant", content: null, tool_calls: [tool("call_a"), tool("call_b")], + })); + const raw = { messages }; + const replayed = `${secretLikeReasoning}\n${secondReasoning}`; + const expandedBytes = Buffer.byteLength(JSON.stringify({ + messages: messages.map(message => ({ ...message, reasoning_content: replayed })), + })); + expect(expandedBytes).toBeGreaterThan(512); + expect(() => buildOpenAIChatPassthroughRequest( + PROVIDER, raw, MODEL, false, undefined, undefined, "fixture", owner, 512, + )).toThrow(expect.objectContaining({ code: "translation_buffer_limit" })); + expect(raw.messages).toHaveLength(40); + expect(raw.messages[0]).not.toHaveProperty("reasoning_content"); + const single = { messages: [{ role: "assistant", content: null, tool_calls: [tool("call_a"), tool("call_b")] }] }; + const full = buildOpenAIChatPassthroughRequest(PROVIDER, single, MODEL, false, undefined, undefined, "fixture", owner); + const exactBytes = Buffer.byteLength(full.body); + expect(JSON.parse(buildOpenAIChatPassthroughRequest( + PROVIDER, single, MODEL, false, undefined, undefined, "fixture", owner, exactBytes, budget, + ).body).messages[0].reasoning_content).toBe(replayed); + expect(() => buildOpenAIChatPassthroughRequest( + PROVIDER, single, MODEL, false, undefined, undefined, "fixture", owner, exactBytes - 1, + )).toThrow(expect.objectContaining({ code: "translation_buffer_limit" })); + const preflightBudget = createTranslatorBudget({ maxTurnBytes: exactBytes - 1 }); + preflightBudget.chargeRetained(60, { kind: "request_copies" }); + const preflightBaseline = preflightBudget.snapshot().currentBytes; + expect(() => buildOpenAIChatPassthroughRequest( + PROVIDER, single, MODEL, false, undefined, undefined, "fixture", owner, undefined, preflightBudget, + )).toThrow(expect.objectContaining({ code: "translation_buffer_limit" })); + expect(preflightBudget.snapshot().currentBytes).toBe(preflightBaseline); + expect(budget.snapshot().currentBytes).toBe(retainedBeforeBuild); + preflightBudget.dispose(); + budget.dispose(); + }); + test("exact value-delta accounting replaces a prior reasoning_content of any JSON shape", () => { + const owner = scope(); + const budget = createTranslatorBudget(); + nativeChatReasoningCapture(PROVIDER, MODEL, owner, budget).json({ choices: [ + { message: { reasoning_content: "cached-text", tool_calls: [tool("call_a")] } }, + ] }); + // "", null, object and number priors already serialize inside the message: the plan credits + // their exact bytes, so the boundary is the real final size in every shape. + for (const prior of ["", null, { nested: ["object"] }, 42] as const) { + const raw = { messages: [{ role: "assistant", content: null, tool_calls: [tool("call_a")], reasoning_content: prior }] }; + const exactBytes = Buffer.byteLength(buildOpenAIChatPassthroughRequest( + PROVIDER, raw, MODEL, false, undefined, undefined, "fixture", owner).body); + expect(JSON.parse(buildOpenAIChatPassthroughRequest( + PROVIDER, raw, MODEL, false, undefined, undefined, "fixture", owner, exactBytes, + ).body).messages[0].reasoning_content).toBe("cached-text"); + expect(() => buildOpenAIChatPassthroughRequest( + PROVIDER, raw, MODEL, false, undefined, undefined, "fixture", owner, exactBytes - 1, + )).toThrow(expect.objectContaining({ code: "translation_buffer_limit" })); + } + budget.dispose(); + }); + test("plan charges the net serialized delta, still applies at zero, and reports the caller limit", () => { + const owner = scope(); + const budget = createTranslatorBudget(); + nativeChatReasoningCapture(PROVIDER, MODEL, owner, budget).json({ choices: [ + { message: { reasoning_content: "aa", tool_calls: [tool("call_a")] } }, + ] }); + // Replacing null (4 serialized bytes) with "aa" (4) nets zero but must still rewrite the wire. + const replacingNull = [{ role: "assistant", content: null, tool_calls: [tool("call_a")], reasoning_content: null }]; + const nullPlan = planNativeChatReasoningReplay(replacingNull, PROVIDER, MODEL, owner, false, 0); + expect(nullPlan.serializedBytes).toBe(0); + expect(nullPlan.hasChanges).toBe(true); + const netZero = { messages: replacingNull }; + expect(JSON.parse(buildOpenAIChatPassthroughRequest( + PROVIDER, netZero, MODEL, false, undefined, undefined, "fixture", owner, + ).body).messages[0].reasoning_content).toBe("aa"); + // "" credits its 2 serialized bytes: only the escaped value difference is charged. + const replacingEmpty = [{ role: "assistant", content: null, tool_calls: [tool("call_a")], reasoning_content: "" }]; + expect(planNativeChatReasoningReplay(replacingEmpty, PROVIDER, MODEL, owner, false, 2).serializedBytes).toBe(2); + // An absent field pays the envelope: "reasoning_content": is 20 bytes, the comma 1, "aa" 4. + const absent = [{ role: "assistant", content: null, tool_calls: [tool("call_a")] }]; + expect(planNativeChatReasoningReplay(absent, PROVIDER, MODEL, owner, false, 25).serializedBytes).toBe(25); + // Inner measurements cap at what is left; the thrown error still reports the caller's limit. + for (const max of [1, 10, 24]) { + expect(() => planNativeChatReasoningReplay(absent, PROVIDER, MODEL, owner, false, max)) + .toThrow(expect.objectContaining({ code: "translation_buffer_limit", limitBytes: max })); + } + budget.dispose(); + }); test("plain-answer placeholders require DeepSeek, tools and both policy opt-ins", () => { const messages = [{ role: "assistant", content: "Earlier answer" }]; const tools = [{ type: "function", function: { name: "inspect", parameters: { type: "object", properties: {} } } }]; diff --git a/tests/responses/empty-completion-guard.test.ts b/tests/responses/empty-completion-guard.test.ts index 227b5c4936..ca77326ce7 100644 --- a/tests/responses/empty-completion-guard.test.ts +++ b/tests/responses/empty-completion-guard.test.ts @@ -8,6 +8,7 @@ import { isContentEvent, observeEmptyCompletion, } from "../../src/server/responses/empty-completion-guard"; +import { bridgeToResponsesSSE } from "../../src/bridge/sse"; import type { AdapterEvent } from "../../src/types"; function collect(source: AsyncIterable): Promise { @@ -115,12 +116,31 @@ describe("empty-completion guard retry", () => { expect(withoutHeartbeats(events)).toEqual([ { type: "thinking_delta", thinking: "first attempt" }, { type: "reasoning_raw_delta", text: "raw" }, + { type: "assistant_boundary" }, { type: "thinking_delta", thinking: "second attempt" }, { type: "text_delta", text: "finally an answer" }, { type: "done", usage: { inputTokens: 30, outputTokens: 5, totalTokens: 35 } }, ]); }); + test("the discarded attempt's reasoning stays out of the retry's reasoning item", async () => { + const guarded = guardEmptyCompletionEventStream({ + firstEvents: eventsOf({ type: "thinking_delta", thinking: "discarded" }, { type: "done" }), + continuation: () => eventsOf( + { type: "thinking_delta", thinking: "retry" }, + { type: "text_delta", text: "answer" }, + { type: "done" }, + ), + }); + const sse = await new Response(bridgeToResponsesSSE(guarded, "fixture-model")).text(); + const reasoning = sse.split("\n") + .filter(line => line.startsWith("data: {")) + .map(line => JSON.parse(line.slice(6)) as { type: string; item?: { type: string; summary?: { text: string }[] } }) + .filter(event => event.type === "response.output_item.done" && event.item?.type === "reasoning") + .map(event => event.item!.summary!.map(part => part.text).join("")); + expect(reasoning).toEqual(["discarded", "retry"]); + }); + test("usage is merged across attempts", async () => { const events = await collect(guardEmptyCompletionEventStream({ firstEvents: eventsOf( @@ -321,7 +341,7 @@ describe("empty-completion guard retry", () => { expect(continuations).toBe(1); expect(events.map(event => event.type)).toEqual([ - "heartbeat", "heartbeat", "thinking_delta", "text_delta", "done", + "heartbeat", "heartbeat", "thinking_delta", "assistant_boundary", "text_delta", "done", ]); }); @@ -341,6 +361,7 @@ describe("empty-completion guard retry", () => { expect(continuations).toBe(1); expect(withoutHeartbeats(events)).toEqual([ { type: "thinking_delta", thinking: "..." }, + { type: "assistant_boundary" }, { type: "text_delta", text: "recovered" }, { type: "done" }, ]); diff --git a/tests/responses/reasoning-archive-http.test.ts b/tests/responses/reasoning-archive-http.test.ts new file mode 100644 index 0000000000..a979b84e9d --- /dev/null +++ b/tests/responses/reasoning-archive-http.test.ts @@ -0,0 +1,113 @@ +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { Database } from "bun:sqlite"; +import * as fs from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, readdirSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createServeOptions } from "../../src/server/index/serve-options"; +import { reasoningArchiveErrorResponse } from "../../src/server/responses/reasoning-archive-error"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { drainCompactionResponseState, installCompactionRoutingAclFixture, + keyProviderConfig, removeCompactionFixture } from "../helpers/compaction-routing-fixtures"; + +installCompactionRoutingAclFixture(); +const previousHome = process.env.OPENCODEX_HOME; +const previousCodexHome = process.env.CODEX_HOME; +const originalFetch = globalThis.fetch; +let home: string; +let releaseSpendHome: () => void; +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "ocx-archive-http-")); + process.env.OPENCODEX_HOME = home; + process.env.CODEX_HOME = home; + releaseSpendHome = acquireOwnedSpendHome(); +}); +afterEach(async () => { + globalThis.fetch = originalFetch; + try { await drainCompactionResponseState(); } finally { + releaseSpendHome(); + await removeCompactionFixture(home); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; + } +}); + +// Exercise the production HTTP router and handlers without opening a listener. Only ingress, +// package integrity and outer turn admission are fixture seams; retention and its SQLite lock run. +function publicFetch(config: ReturnType) { + const options = createServeOptions({ + config, deps: {}, managementApiDeps: {}, + ingressForServer: () => "public", + packageTreeIntegrity: { status: () => ({ ok: true }) }, + runAdmittedHttpTurn: (_req: Request, _policy: unknown, work: () => Promise) => work(), + } as unknown as Parameters[0]); + const server = { timeout() {} } as unknown as Parameters[1]; + return (req: Request) => options.fetch(req, server); +} + +describe("reasoning archive public HTTP refusal", () => { + test("unrelated exceptions retain their existing handling", () => { + expect(reasoningArchiveErrorResponse(new Error("unrelated failure"))).toBeUndefined(); + }); + for (const [v1, stream] of [[true, false], [false, false], [false, true]]) { + test.each(["quota", "busy", "io", "write"] as const)(`v${v1 ? 1 : 2} stream=${stream} %s is safe, actionable and sends nothing`, async kind => { + const dir = join(home, "reasoning-archive"); + mkdirSync(dir); + writeFileSync(join(dir, "prior.md"), "preserved old archive"); + let lock: Database | undefined; + if (kind === "quota") { + for (let i = 1; i < 256; i++) writeFileSync(join(dir, `prior-${i}.md`), ""); + } else if (kind === "busy") { + lock = new Database(`${dir}.lock`, { create: true }); + lock.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + } else if (kind === "io") { + mkdirSync(join(dir, "foreign-directory")); + } + const before = readdirSync(dir).sort(); + const config = keyProviderConfig(); + config.reasoningRetention = { enabled: true, maxTokens: 1 }; + config.compactionRecovery = { enabled: true, model: "backup/other" }; + config.providers.backup = { ...config.providers.gw! }; + let sends = 0; + const mock = (async () => { sends++; throw new Error("unexpected upstream send"); }) as typeof fetch; + globalThis.fetch = mock; + for (const provider of Object.values(config.providers)) Object.assign(provider, { fetch: mock, upstreamWebsocket: false }); + const body = { model: "gw/some-model", stream, input: [ + { type: "message", role: "user", content: "continue" }, + { type: "reasoning", summary: [{ type: "summary_text", text: "private retained reasoning" }], + encrypted_content: "private-provider-signature" }, + ...(!v1 ? [{ type: "compaction_trigger" }] : []), + ] }; + const original = structuredClone(body); + const originalWrite = fs.writeFileSync; + const write = kind === "write" ? spyOn(fs, "writeFileSync").mockImplementation((path, ...args) => { + if (typeof path === "number") { + originalWrite(path, "partial"); + throw new Error(`fixture ENOSPC ${dir} private-provider-signature`); + } + return originalWrite(path, ...args); + }) : undefined; + try { + const response = await publicFetch(config)(new Request(`http://localhost/v1/responses${v1 ? "/compact" : ""}`, { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), + })); + expect(response.status).toBe(503); + expect(response.headers.get("retry-after")).toBeNull(); + const text = await response.text(); + const json = JSON.parse(text); + const code = kind === "quota" ? "reasoning_archive_quota_exceeded" + : kind === "busy" ? "reasoning_archive_busy" : "reasoning_archive_unavailable"; + expect(json.error).toMatchObject({ type: "server_error", code }); + expect(json.error.message).toContain(kind === "quota" ? "manually" : kind === "busy" ? "retry" : "permissions"); + for (const secret of [home, dir, "prior.md", "foreign-directory", "private retained reasoning", "private-provider-signature", "test-key"]) + expect(text).not.toContain(secret); + expect(sends).toBe(0); + expect(body).toEqual(original); + expect(readdirSync(dir).sort()).toEqual(before); + expect(readFileSync(join(dir, "prior.md"), "utf8")).toBe("preserved old archive"); + } finally { write?.mockRestore(); lock?.close(); } + }); + } +}); diff --git a/tests/responses/reasoning-archive-ws.test.ts b/tests/responses/reasoning-archive-ws.test.ts new file mode 100644 index 0000000000..bdb8539b21 --- /dev/null +++ b/tests/responses/reasoning-archive-ws.test.ts @@ -0,0 +1,118 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { Database } from "bun:sqlite"; +import { mkdirSync, mkdtempSync, readFileSync, readdirSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { saveConfig } from "../../src/config"; +import { startServer } from "../../src/server"; +import type { OcxConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { drainCompactionResponseState, installCompactionRoutingAclFixture, + removeCompactionFixture } from "../helpers/compaction-routing-fixtures"; +import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; + +// The WebSocket transport has its own catch for errors thrown before a response exists. A local +// archive refusal must reach the client with the same sanitized 503 body and stable code as HTTP +// (reasoning-archive-http.test.ts), never as a raw 502 proxy_error carrying the archive path. +installCompactionRoutingAclFixture(); +const previousHome = process.env.OPENCODEX_HOME; +const originalFetch = globalThis.fetch; +let home: string; +let codex: IsolatedCodexHome; +let releaseSpendHome: () => void; +let proxy: ReturnType | undefined; +let socket: WebSocket | undefined; +let sends = 0; + +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "ocx-archive-ws-")); + process.env.OPENCODEX_HOME = home; + codex = installIsolatedCodexHome("ocx-archive-ws-codex-"); + releaseSpendHome = acquireOwnedSpendHome(); + sends = 0; + globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(input instanceof Request ? input.url : String(input)); + if (url.hostname === "127.0.0.1" || url.hostname === "localhost") return originalFetch(input, init); + sends++; + throw new Error("unexpected upstream send"); + }) as typeof fetch; +}); + +afterEach(async () => { + try { + socket?.close(); + await proxy?.stop(true); + await drainCompactionResponseState(); + } finally { + socket = undefined; + proxy = undefined; + globalThis.fetch = originalFetch; + releaseSpendHome(); + codex.restore(); + await removeCompactionFixture(home); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + } +}); + +async function until(condition: () => boolean, description: string, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!condition()) { + if (Date.now() >= deadline) throw new Error(`Timed out: ${description}`); + await Bun.sleep(2); + } +} + +describe("reasoning archive WebSocket refusal", () => { + test.each(["quota", "busy", "io"] as const)("%s is a safe 503 frame with a stable code and sends nothing", async kind => { + const dir = join(home, "reasoning-archive"); + mkdirSync(dir); + writeFileSync(join(dir, "prior.md"), "preserved old archive"); + let lock: Database | undefined; + if (kind === "quota") { + for (let i = 1; i < 256; i++) writeFileSync(join(dir, `prior-${i}.md`), ""); + } else if (kind === "busy") { + lock = new Database(`${dir}.lock`, { create: true }); + lock.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + } else { + mkdirSync(join(dir, "foreign-directory")); + } + const before = readdirSync(dir).sort(); + const provider = { adapter: "openai-responses", baseUrl: "https://gateway.example/v1", authMode: "key", + apiKey: "test-key", upstreamWebsocket: false }; + saveConfig({ + port: 0, hostname: "127.0.0.1", codexAutoStart: false, websockets: true, defaultProvider: "gw", + providers: { gw: { ...provider }, backup: { ...provider } }, + reasoningRetention: { enabled: true, maxTokens: 1 }, + compactionRecovery: { enabled: true, model: "backup/other" }, + } as unknown as OcxConfig); + try { + proxy = startServer(0); + const url = new URL("/v1/responses", proxy.url); + url.protocol = "ws:"; + const frames: Record[] = []; + const raw: string[] = []; + socket = new WebSocket(url); + socket.addEventListener("message", event => { raw.push(String(event.data)); frames.push(JSON.parse(String(event.data))); }); + await until(() => socket!.readyState === WebSocket.OPEN, "socket handshake"); + socket.send(JSON.stringify({ type: "response.create", model: "gw/some-model", input: [ + { type: "message", role: "user", content: "continue" }, + { type: "reasoning", summary: [{ type: "summary_text", text: "private retained reasoning" }], + encrypted_content: "private-provider-signature" }, + { type: "compaction_trigger" }, + ] })); + await until(() => frames.some(frame => frame.type === "error"), "error frame"); + const frame = frames.find(item => item.type === "error")!; + const code = kind === "quota" ? "reasoning_archive_quota_exceeded" + : kind === "busy" ? "reasoning_archive_busy" : "reasoning_archive_unavailable"; + expect(frame.status).toBe(503); + expect(frame.error).toMatchObject({ type: "server_error", code }); + const text = raw.join("\n"); + for (const secret of [home, dir, "prior.md", "foreign-directory", "private retained reasoning", "private-provider-signature", "test-key"]) + expect(text).not.toContain(secret); + expect(sends).toBe(0); + expect(readdirSync(dir).sort()).toEqual(before); + expect(readFileSync(join(dir, "prior.md"), "utf8")).toBe("preserved old archive"); + } finally { lock?.close(); } + }); +}); diff --git a/tests/responses/reasoning-retention-quota.test.ts b/tests/responses/reasoning-retention-quota.test.ts new file mode 100644 index 0000000000..1bef066921 --- /dev/null +++ b/tests/responses/reasoning-retention-quota.test.ts @@ -0,0 +1,314 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import * as fs from "node:fs"; +import { createHash } from "node:crypto"; +import { + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + readdirSync, + rmSync, + statSync, + symlinkSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + applyReasoningRetention, + ReasoningArchiveBusyError, + ReasoningArchiveError, + ReasoningArchiveQuotaError, +} from "../../src/responses/reasoning-retention"; + +const reasoningItem = (text: string) => ({ + type: "reasoning", + summary: [{ type: "summary_text", text }], +}); + +const sha256 = (path: string) => createHash("sha256").update(readFileSync(path)).digest("hex"); + +const LOCK_SIDECARS = ["-journal", "-wal", "-shm"] as const; + +describe("applyReasoningRetention archive quota", () => { + const dirs: string[] = []; + const extraPaths: string[] = []; + afterEach(() => { + for (const dir of dirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + rmSync(dir + ".lock", { recursive: true, force: true }); + for (const suffix of LOCK_SIDECARS) rmSync(dir + ".lock" + suffix, { force: true }); + } + for (const path of extraPaths.splice(0)) rmSync(path, { recursive: true, force: true }); + }); + + const setup = () => { + const dir = mkdtempSync(join(tmpdir(), "reasoning-quota-")); + dirs.push(dir); + return dir; + }; + + test("legacy over-capacity scanning stops actual directory reads at the cap and closes", () => { + const archiveDir = setup(); + for (let i = 0; i < 12; i++) writeFileSync(join(archiveDir, `${i}.md`), ""); + const openDirectory = fs.opendirSync.bind(fs); + let reads = 0; + let closes = 0; + const open = spyOn(fs, "opendirSync").mockImplementation(((path: fs.PathLike) => { + const dir = openDirectory(path, { bufferSize: 1 }); + const read = dir.readSync.bind(dir); + const close = dir.closeSync.bind(dir); + dir.readSync = () => { reads++; return read(); }; + dir.closeSync = () => { closes++; close(); }; + return dir; + }) as typeof fs.opendirSync); + try { + expect(() => applyReasoningRetention([reasoningItem("new")], { + archiveDir, maxTokens: 1, archiveMaxFiles: 2, + })).toThrow(ReasoningArchiveQuotaError); + expect(open).toHaveBeenCalledTimes(1); + expect(reads).toBe(2); + expect(closes).toBe(1); + } finally { open.mockRestore(); } + expect(readdirSync(archiveDir)).toHaveLength(12); + }); + + test.each(["bytes", "unsafe", "read-error", "vanished", "close-error"] as const)( + "streaming scan closes on %s refusal without reading beyond its budget", kind => { + const archiveDir = setup(); + for (let i = 0; i < 12; i++) writeFileSync(join(archiveDir, `${i}.md`), "data"); + const openDirectory = fs.opendirSync.bind(fs); + const originalStat = fs.lstatSync.bind(fs); + let reads = 0; + let closes = 0; + const stat = kind === "vanished" ? spyOn(fs, "lstatSync").mockImplementation(((path: fs.PathLike) => { + if (String(path).endsWith(".md")) throw Object.assign(new Error("vanished"), { code: "ENOENT" }); + return originalStat(path); + }) as typeof fs.lstatSync) : undefined; + const open = spyOn(fs, "opendirSync").mockImplementation(((path: fs.PathLike) => { + const dir = openDirectory(path, { bufferSize: 1 }); + const read = dir.readSync.bind(dir); + const close = dir.closeSync.bind(dir); + dir.readSync = () => { + reads++; + if (kind === "read-error") throw new Error("fixture EIO"); + const entry = read(); + if (kind === "unsafe" && entry) entry.isFile = () => false; + return entry; + }; + dir.closeSync = () => { closes++; close(); if (kind === "close-error") throw new Error("fixture EIO"); }; + return dir; + }) as typeof fs.opendirSync); + try { + expect(() => applyReasoningRetention([reasoningItem("new")], { + archiveDir, maxTokens: 1, archiveMaxFiles: 2, archiveMaxBytes: 1, + })).toThrow(kind === "bytes" ? ReasoningArchiveQuotaError : ReasoningArchiveError); + expect(reads).toBe(kind === "vanished" ? 2 : 1); + expect(closes).toBe(1); + } finally { open.mockRestore(); stat?.mockRestore(); } + expect(readdirSync(archiveDir)).toHaveLength(12); + }, + ); + + test.each(["read", "stat", "undefined", "success"] as const)("scan and close failure priority (%s)", kind => { + const archiveDir = setup(); + const prior = join(archiveDir, "prior.md"); + writeFileSync(prior, "preserve"); + const primary = kind === "undefined" ? undefined : new Error(`fixture primary ${kind}`); + const secondary = new Error("fixture close failure"); + const openDirectory = fs.opendirSync.bind(fs); + const originalStat = fs.lstatSync.bind(fs); + let closes = 0; + const stat = kind === "stat" ? spyOn(fs, "lstatSync").mockImplementation(((path: fs.PathLike, ...args: unknown[]) => { + if (String(path) === prior) throw primary; + return (originalStat as (...args: unknown[]) => unknown)(path, ...args); + }) as typeof fs.lstatSync) : undefined; + const open = spyOn(fs, "opendirSync").mockImplementation(((path: fs.PathLike) => { + const dir = openDirectory(path, { bufferSize: 1 }); + const read = dir.readSync.bind(dir); + const close = dir.closeSync.bind(dir); + dir.readSync = () => { + if (kind === "read" || kind === "undefined") throw primary; + return read(); + }; + dir.closeSync = () => { closes++; close(); throw secondary; }; + return dir; + }) as typeof fs.opendirSync); + const input = [reasoningItem("original reasoning")]; + const before = structuredClone(input); + let rejected = false; + let failure: unknown; + try { + try { applyReasoningRetention(input, { archiveDir, maxTokens: 1, archiveMaxFiles: 1 }); } + catch (error) { rejected = true; failure = error; } + expect(rejected).toBe(true); + expect(failure).toBeInstanceOf(ReasoningArchiveError); + expect(Object.hasOwn(failure as object, "cause")).toBe(true); + expect((failure as Error).cause).toBe(kind === "success" ? secondary : primary); + expect((failure as Error).message).toContain(kind === "success" ? "could not be closed" + : kind === "stat" ? "not inspectable" : "cannot be scanned"); + expect(closes).toBe(1); + expect(input).toEqual(before); + expect(readdirSync(archiveDir)).toEqual(["prior.md"]); + expect(readFileSync(prior, "utf8")).toBe("preserve"); + } finally { open.mockRestore(); stat?.mockRestore(); } + }); + + test("byte cap admits an exactly fitting archive and refuses one byte over", () => { + const archiveDir = setup(); + const input = [reasoningItem("x".repeat(100))]; + const allowed = applyReasoningRetention(input, { archiveDir, maxTokens: 1, archiveMaxBytes: 100 }); + expect(allowed.archived).toBeDefined(); + expect(statSync(allowed.archived!.path).size).toBe(100); + + const denied = setup(); + expect(() => applyReasoningRetention(input, { archiveDir: denied, maxTokens: 1, archiveMaxBytes: 99 })) + .toThrow(ReasoningArchiveQuotaError); + expect(readdirSync(denied)).toEqual([]); + }); + + test("UTF-8 text and separator bytes are charged exactly before the join exists", () => { + const archiveDir = setup(); + // "ํ•œ"(3) + separator "\n\n---\n\n"(7) + "๊ฐ€"(3) = 13 bytes on disk. + const input = [reasoningItem("ํ•œ"), reasoningItem("๊ฐ€")]; + expect(() => applyReasoningRetention(input, { archiveDir, maxTokens: 1, archiveMaxBytes: 12 })) + .toThrow(ReasoningArchiveQuotaError); + expect(readdirSync(archiveDir)).toEqual([]); + + const allowed = applyReasoningRetention(input, { archiveDir, maxTokens: 1, archiveMaxBytes: 13 }); + expect(allowed.archived).toBeDefined(); + expect(statSync(allowed.archived!.path).size).toBe(13); + }); + + test("file-count cap refuses a new archive without touching existing files", () => { + const archiveDir = setup(); + const prior = join(archiveDir, "reasoning-old.md"); + writeFileSync(prior, "keep me"); + const before = sha256(prior); + + expect(() => applyReasoningRetention([reasoningItem("new")], { archiveDir, maxTokens: 1, archiveMaxFiles: 1 })) + .toThrow(ReasoningArchiveQuotaError); + expect(sha256(prior)).toBe(before); + expect(readdirSync(archiveDir)).toEqual(["reasoning-old.md"]); + }); + + test("boundary: the file landing exactly at the count cap is archived, the next is refused", () => { + const archiveDir = setup(); + const first = applyReasoningRetention([reasoningItem("first")], { archiveDir, maxTokens: 1, archiveMaxFiles: 1 }); + expect(first.archived).toBeDefined(); + expect(readdirSync(archiveDir)).toHaveLength(1); + expect(() => applyReasoningRetention([reasoningItem("second")], { archiveDir, maxTokens: 1, archiveMaxFiles: 1 })) + .toThrow(ReasoningArchiveQuotaError); + expect(readdirSync(archiveDir)).toHaveLength(1); + }); + + test("legacy over-capacity contents are preserved and only refuse new archives", () => { + const archiveDir = setup(); + const heavy = join(archiveDir, "reasoning-heavy.md"); + const other = join(archiveDir, "unrelated.bin"); + writeFileSync(heavy, "h".repeat(50)); + writeFileSync(other, "o".repeat(30)); + const beforeHeavy = sha256(heavy); + const beforeOther = sha256(other); + + const input = [reasoningItem("small")]; + const snapshot = JSON.stringify(input); + expect(() => applyReasoningRetention(input, { archiveDir, maxTokens: 1, archiveMaxBytes: 10 })) + .toThrow(ReasoningArchiveQuotaError); + expect(JSON.stringify(input)).toBe(snapshot); + expect(sha256(heavy)).toBe(beforeHeavy); + expect(sha256(other)).toBe(beforeOther); + }); + + test("a non-regular archive entry refuses closed instead of being ignored", () => { + const archiveDir = setup(); + mkdirSync(join(archiveDir, "foreign-subdir")); + expect(() => applyReasoningRetention([reasoningItem("x")], { archiveDir, maxTokens: 1 })) + .toThrow(ReasoningArchiveError); + expect(readdirSync(archiveDir)).toEqual(["foreign-subdir"]); + }); + + test("a non-inspectable archive entry refuses closed", () => { + const archiveDir = setup(); + const blocked = join(archiveDir, "blocked.md"); + writeFileSync(blocked, "data"); + const original = fs.lstatSync.bind(fs); + const stat = spyOn(fs, "lstatSync").mockImplementation(((path: string, options?: unknown) => { + if (String(path) === blocked) { + const denied = new Error("denied") as NodeJS.ErrnoException; + denied.code = "EACCES"; + throw denied; + } + return (original as (...args: unknown[]) => unknown)(path, options); + }) as typeof fs.lstatSync); + try { + expect(() => applyReasoningRetention([reasoningItem("x")], { archiveDir, maxTokens: 1 })) + .toThrow(ReasoningArchiveError); + } finally { + stat.mockRestore(); + } + }); + + test("a symlinked archive directory refuses closed", () => { + const archiveDir = setup(); + const real = mkdtempSync(join(tmpdir(), "reasoning-quota-real-")); + extraPaths.push(real); + const link = join(tmpdir(), "reasoning-quota-link-" + Math.random().toString(36).slice(2, 8)); + extraPaths.push(link); + try { + symlinkSync(real, link, "junction"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "EPERM" || (error as NodeJS.ErrnoException).code === "ENOTSUP") return; + throw error; + } + expect(() => applyReasoningRetention([reasoningItem("x")], { archiveDir: link, maxTokens: 1 })) + .toThrow(ReasoningArchiveError); + expect(readdirSync(real)).toEqual([]); + expect(archiveDir && existsSync(archiveDir)).toBe(true); + }); + + test("a lock path that is not a regular file refuses closed", () => { + const archiveDir = setup(); + mkdirSync(archiveDir + ".lock"); + expect(() => applyReasoningRetention([reasoningItem("x")], { archiveDir, maxTokens: 1 })) + .toThrow(ReasoningArchiveError); + expect(readdirSync(archiveDir)).toEqual([]); + }); + + test("a lock held by another process fails busy, writes nothing, and the retry archives", async () => { + const archiveDir = setup(); + const lockPath = archiveDir + ".lock"; + const marker = archiveDir + ".held"; + extraPaths.push(marker); + const child = Bun.spawn([ + process.execPath, + "-e", + "import { Database } from \"bun:sqlite\";" + + "import { writeFileSync } from \"node:fs\";" + + "const db = new Database(" + JSON.stringify(lockPath) + ", { create: true });" + + "db.exec(\"PRAGMA busy_timeout = 0; BEGIN IMMEDIATE\");" + + "writeFileSync(" + JSON.stringify(marker) + ", \"held\");" + + "await Bun.sleep(30000);", + ], { stdout: "ignore", stderr: "ignore" }); + try { + const deadline = Date.now() + 10_000; + while (!existsSync(marker)) { + if (Date.now() > deadline) throw new Error("child did not acquire the archive lock"); + await Bun.sleep(20); + } + const input = [reasoningItem("queued reasoning")]; + const snapshot = JSON.stringify(input); + expect(() => applyReasoningRetention(input, { archiveDir, maxTokens: 1 })) + .toThrow(ReasoningArchiveBusyError); + expect(JSON.stringify(input)).toBe(snapshot); + expect(readdirSync(archiveDir)).toEqual([]); + } finally { + child.kill(); + await child.exited; + } + // The OS released the killed child's transaction; a same-request retry archives normally. + const result = applyReasoningRetention([reasoningItem("queued reasoning")], { archiveDir, maxTokens: 1 }); + expect(result.archived).toBeDefined(); + expect(readdirSync(archiveDir)).toHaveLength(1); + }, 15_000); +}); diff --git a/tests/responses/reasoning-retention.test.ts b/tests/responses/reasoning-retention.test.ts index dfeb4cb9a3..c7cee8f0d9 100644 --- a/tests/responses/reasoning-retention.test.ts +++ b/tests/responses/reasoning-retention.test.ts @@ -7,6 +7,7 @@ import { appendRetentionNotice, applyReasoningRetention, collectReasoningTexts, + ReasoningArchiveError, } from "../../src/responses/reasoning-retention"; import { encodeReasoningEnvelope } from "../../src/responses/reasoning-envelope"; import { estimateTokens } from "../../src/lib/token-estimate"; @@ -35,7 +36,11 @@ describe("collectReasoningTexts", () => { describe("applyReasoningRetention", () => { let dir = ""; afterEach(() => { - if (dir) rmSync(dir, { recursive: true, force: true }); + if (dir) { + rmSync(dir, { recursive: true, force: true }); + rmSync(dir + ".lock", { force: true }); + for (const suffix of ["-journal", "-wal", "-shm"]) rmSync(dir + ".lock" + suffix, { force: true }); + } dir = ""; }); @@ -132,7 +137,12 @@ describe("applyReasoningRetention", () => { throw new Error("fixture ENOSPC"); }); try { - expect(() => applyReasoningRetention([reasoningItem("full reasoning")], { archiveDir, maxTokens: 1 })).toThrow("fixture ENOSPC"); + let failure: unknown; + try { applyReasoningRetention([reasoningItem("full reasoning")], { archiveDir, maxTokens: 1 }); } + catch (error) { failure = error; } + expect(failure).toBeInstanceOf(ReasoningArchiveError); + expect((failure as Error).cause).toBeInstanceOf(Error); + expect(((failure as Error).cause as Error).message).toBe("fixture ENOSPC"); expect(readdirSync(archiveDir)).toEqual([]); } finally { write.mockRestore(); } }); diff --git a/tests/responses/responses-azure-opaque-blob-recovery.test.ts b/tests/responses/responses-azure-opaque-blob-recovery.test.ts index b06fb5dd8a..01219bab40 100644 --- a/tests/responses/responses-azure-opaque-blob-recovery.test.ts +++ b/tests/responses/responses-azure-opaque-blob-recovery.test.ts @@ -191,12 +191,14 @@ describe("the recovery gate follows the adapter's declared Responses contract", expect(adapterSpeaksResponsesWire("not-a-registered-adapter")).toBe(false); }); - test("azure-openai does not recover an ordinary 400, a 429, or a 5xx", () => { + test("azure-openai recovers an identified 502 but not unrelated failures", () => { const azure = { ...base, adapterName: "azure-openai" }; expect(shouldAttemptOpaqueBlobRecovery(azure)).toBe(true); expect(shouldAttemptOpaqueBlobRecovery({ ...azure, errorBody: UNRELATED_400 })).toBe(false); expect(shouldAttemptOpaqueBlobRecovery({ ...azure, status: 429, errorBody: RATE_LIMITED })).toBe(false); - for (const status of [500, 502, 503]) { + expect(shouldAttemptOpaqueBlobRecovery({ ...azure, status: 502 })).toBe(true); + expect(shouldAttemptOpaqueBlobRecovery({ ...azure, status: 502, errorBody: UNRELATED_400 })).toBe(false); + for (const status of [500, 503]) { expect({ status, recovers: shouldAttemptOpaqueBlobRecovery({ ...azure, status }) }) .toEqual({ status, recovers: false }); } diff --git a/tests/responses/responses-cold-opaque-recovery.test.ts b/tests/responses/responses-cold-opaque-recovery.test.ts new file mode 100644 index 0000000000..d081e68f1b --- /dev/null +++ b/tests/responses/responses-cold-opaque-recovery.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { isSelfIdentifiedOpaqueBlobRejection } from "../../src/server/responses/core-opaque-recovery"; +import { handleResponses } from "../../src/server/responses"; +import { clearReasoningReplayCacheForTests } from "../../src/responses/reasoning-replay-cache"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import type { OcxConfig } from "../../src/types"; + +const originalFetch = globalThis.fetch; +let previousHome: string | undefined; +let home: string; +let release: () => void; +const blob = "fixture-native-ciphertext"; +const error = { type: "invalid_request_error", code: "invalid_encrypted_content", message: "The encrypted content could not be verified." }; +const cfg = { defaultProvider: "relay", providers: { relay: { adapter: "openai-responses", baseUrl: "https://relay.example.test/v1", authMode: "key", apiKey: "fixture-key", decodesNativeCompactionBlobs: true } } } as OcxConfig; +const input = [ + { type: "message", role: "user", content: [{ type: "input_text", text: "remember fixture" }] }, + { type: "reasoning", summary: [], encrypted_content: blob }, + { type: "function_call", call_id: "fixture-call", name: "lookup", arguments: "{}" }, + { type: "function_call_output", call_id: "fixture-call", output: "fixture-result" }, + { type: "message", role: "user", content: [{ type: "input_text", text: "continue" }] }, +]; +const frame = (type: string, data: object) => `event: ${type}\ndata: ${JSON.stringify({ type, ...data })}\n\n`; +const success = () => new Response(frame("response.completed", { response: { id: "fixture-ok", status: "completed", output: [] } }), { headers: { "content-type": "text/event-stream" } }); +const rejected = () => new Response(frame("response.failed", { response: { id: "fixture-error", status: "failed", error } }), { headers: { "content-type": "text/event-stream" } }); + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + home = mkdtempSync(join(tmpdir(), "ocx-cold-opaque-")); + process.env.OPENCODEX_HOME = home; + release = acquireOwnedSpendHome(); + clearReasoningReplayCacheForTests(); +}); +afterEach(() => { + globalThis.fetch = originalFetch; + clearReasoningReplayCacheForTests(); + release(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + removeTreeWithRetry(home); +}); + +async function send() { + const body = JSON.stringify({ model: "relay/model-a", stream: true, store: false, input }); + const req = new Request("http://localhost/v1/responses", { method: "POST", headers: { "content-type": "application/json" }, body }); + const response = await handleResponses(req, cfg, { model: "", provider: "" }); + const text = await response.text(); + expect(JSON.stringify(input)).toContain(blob); + return text; +} + +for (const shape of ["sse", "flat-code", "flat-wording", "502"] as const) test(`cold reasoning rejection ${shape} rebuilds once and preserves tool pairs (#7369)`, async () => { + const sent: Record[] = []; + globalThis.fetch = (async (_url, init) => { + sent.push(JSON.parse(String(init?.body))); + if (sent.length <= (shape === "502" ? 3 : 1)) { + if (shape === "502") return Response.json({ error }, { status: 502 }); + if (shape === "sse") return rejected(); + const flat = shape === "flat-code" ? { code: error.code, message: error.message } : { + message: "The encrypted content fixture could not be verified. Reason: Encrypted content could not be decrypted or parsed.", + }; + return new Response(frame("error", flat), { headers: { "content-type": "text/event-stream" } }); + } + return success(); + }) as typeof fetch; + const text = await send(); + expect(text).toContain("response.completed"); + expect(sent).toHaveLength(shape === "502" ? 4 : 2); + expect(sent.every(body => body.store === false)).toBe(true); + expect(JSON.stringify(sent[0])).toContain(blob); + expect(JSON.stringify(sent.at(-1))).not.toContain(blob); + const items = sent.at(-1)!.input as Array>; + expect(items.filter(x => x.call_id === "fixture-call").map(x => x.type)).toEqual(["function_call", "function_call_output"]); + expect(items.find(x => x.type === "function_call_output")?.output).toBe("fixture-result"); +}); + +test("cold valid native replay is preserved without another send", async () => { + let sends = 0; + globalThis.fetch = (async (_url, init) => { sends++; expect(String(init?.body)).toContain(blob); return success(); }) as typeof fetch; + expect(await send()).toContain("response.completed"); + expect(sends).toBe(1); +}); + +test("repeated explicit rejection is terminal after one rebuild", async () => { + let sends = 0; + globalThis.fetch = (async () => { sends++; return rejected(); }) as typeof fetch; + expect(await send()).not.toContain("response.completed"); + expect(sends).toBe(2); +}); + +test("generic 502 does not authorize opaque recovery", async () => { + let sends = 0; + globalThis.fetch = (async () => { sends++; return Response.json({ error: { type: "server_error", message: "gateway unavailable" } }, { status: 502 }); }) as typeof fetch; + await send(); + expect(sends).toBe(3); // Existing generic pre-header transient retries remain unchanged. +}); + +test.each(["text", "tool"])("a rejection after committed %s cannot resend the turn", async kind => { + let sends = 0; + globalThis.fetch = (async () => { + sends++; + return new Response((kind === "text" ? frame("response.output_text.delta", { delta: "visible text", item_id: "fixture-text", output_index: 0, content_index: 0 }) : frame("response.output_item.added", { output_index: 0, item: { type: "function_call", id: "fixture-tool", call_id: "fixture-tool-call", name: "lookup", arguments: "{}" } })) + frame("response.failed", { response: { status: "failed", error } }), { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch; + await send(); + expect(sends).toBe(1); +}); + +test("a competing nested failure cannot borrow a flat ciphertext code", () => { + expect(isSelfIdentifiedOpaqueBlobRejection(JSON.stringify({ type: "error", code: "invalid_encrypted_content", + message: "The encrypted content could not be verified.", error: { type: "rate_limit_error", code: "rate_limit_exceeded", message: "quota" } }))).toBe(false); +}); diff --git a/tests/responses/responses-forward-transport-baseurl.test.ts b/tests/responses/responses-forward-transport-baseurl.test.ts new file mode 100644 index 0000000000..cefe51ebde --- /dev/null +++ b/tests/responses/responses-forward-transport-baseurl.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, test } from "bun:test"; +import { buildResponsesTransport } from "../../src/adapters/openai-responses/transport"; +import type { OcxProviderConfig } from "../../src/types"; + +// Regression coverage for the forward-mode baseUrl normalization: the previous implementation +// used `replace(/\/+$/, "")`, which retries candidate end positions when a long slash run is not +// at the end of the string. The replacement is a linear scan, pinned here on the shapes that +// distinguish the two. Provider validation requires an absolute http(s) baseUrl, so empty and +// slash-only values are outside this call's contract and are not exercised here. +const forwardProvider = (baseUrl: string): OcxProviderConfig => ({ + adapter: "openai-responses", + authMode: "forward", + baseUrl, +}); + +const incoming = { headers: new Headers() }; + +describe("buildResponsesTransport baseUrl normalization", () => { + test("noncanonical forward baseUrl drops every trailing slash", () => { + const { url } = buildResponsesTransport(forwardProvider("https://relay.example.test/v1///"), incoming); + expect(url).toBe("https://relay.example.test/v1/responses"); + }); + + test("a single trailing slash is dropped", () => { + const { url } = buildResponsesTransport(forwardProvider("https://relay.example.test/"), incoming); + expect(url).toBe("https://relay.example.test/responses"); + }); + + test("a baseUrl without a trailing slash is unchanged", () => { + const { url } = buildResponsesTransport(forwardProvider("https://relay.example.test/v1"), incoming); + expect(url).toBe("https://relay.example.test/v1/responses"); + }); + + test("interior slash runs are preserved", () => { + const { url } = buildResponsesTransport(forwardProvider("https://relay.example.test//v1//"), incoming); + expect(url).toBe("https://relay.example.test//v1/responses"); + }); + + test("a mid-path slash run does not stall normalization", () => { + // The input shape that made `/\/+$/` retry: a long slash run that is not at the end. + // The linear scan treats it like any other URL and preserves it. + const baseUrl = `https://relay.example.test${"/".repeat(2_000)}v1`; + const { url } = buildResponsesTransport(forwardProvider(baseUrl), incoming); + expect(url).toBe(`${baseUrl}/responses`); + }); +}); diff --git a/tests/service/shutdown-drain.test.ts b/tests/service/shutdown-drain.test.ts index dd3d1e31cd..329f761b79 100644 --- a/tests/service/shutdown-drain.test.ts +++ b/tests/service/shutdown-drain.test.ts @@ -14,6 +14,7 @@ import { markRecyclingForExit, } from "../../src/server"; import { + abortAndReleaseAllTurns, acquireTemporaryDrain, acquireNativeMainProfileDrain, activeRegistryMetrics, @@ -375,6 +376,24 @@ describe("trackStreamLifetime", () => { }); }); +describe("trackStreamLifetime abort", () => { + // Callers such as the bridge and client-encoder delivery pass a fresh controller that nothing + // else listens to, so the tracked stream is the only thing a drain abort can stop. + test("a drain abort ends the stream and cancels its source", async () => { + let sourceCancelled = false; + let done = 0; + const source = new ReadableStream({ pull() {}, cancel() { sourceCancelled = true; } }); + const lease = tryAdmitTurn(); + expect(lease).not.toBeNull(); + const reader = trackStreamLifetime(source, new AbortController(), () => { done += 1; }, lease!).getReader(); + const pending = reader.read().then(() => "read", () => "errored"); + abortAndReleaseAllTurns(new Error("server shutdown")); + expect(await Promise.race([pending, Bun.sleep(200).then(() => "still open")])).toBe("errored"); + expect(sourceCancelled).toBe(true); + expect(done).toBe(1); + }); +}); + describe("recycling exit flag (#563)", () => { test("markRecyclingForExit flips the recycle sentinel for syncCleanup", () => { expect(isRecyclingForExit()).toBe(false); From bddd61f74bc89f45e6740845823b47f7a0a2f629 Mon Sep 17 00:00:00 2001 From: JUN Date: Mon, 19 Oct 2026 14:10:18 -0700 Subject: [PATCH 03/16] fix(responses): carry WS dispatch override, pooled key rotation, usage accounting and Kiro refusal hop (#7389) * fix(responses): run the dispatch override on the Codex WS branch providerFetch applied `dispatchOverride` only on the HTTP path. The WebSocket branch called codexWsUpstreamFetch directly, so a key-auth WebSocket send (api.openai.com with upstreamWebsocket) never reached commitKeyAttemptSend or rebindPhysicalSend: its attempt row reported sendCount 0 for a frame that left. The WS send now runs through the same override; the override hands a marked WS executor its send directly instead of HTTP connection policy, and the HTTP fallback inside it skips the override so one logical send is counted once. Co-Authored-By: Claude Code * fix(responses): send a dispatch-time reselection over HTTP, not the WS WS eligibility (egress, pacing, URL) is checked against the provider the WS executor was built for. The dispatch override can replace route.provider before it calls the executor; that replacement was never checked, so it could ride a WebSocket its own config rules out (e.g. an explicit proxy). The WS executor now records the provider it was admitted for and its HTTP fallback; when the override serves a different provider it sends through that fallback under the replacement's own egress route. Co-Authored-By: Claude Code * fix(responses): rotate a pooled API key on a native passthrough 429 The native Responses passthrough returns before the adapter recovery loop that rotates pooled keys, so a key-pool provider's 429 was returned to the client while the same refusal rotates on the adapter and chat-native wires. Rotate to the next key before output, inside the shared send budget. Distributed round-robin selection keeps its cool-without-replay behaviour. Co-Authored-By: Claude Code * fix(usage): seal a credential-rotation attempt with the status it answered When an account rotation splits the attempt row, the sealed attempt with no recorded status fell back to 401 for key-401, 429 for any *-429 kind, and 502 for everything else. An `oauth-account-403` rotation was therefore logged as a 502 server error, and `oauth-401` / `oauth-account-402` likewise, although the recovery kind names the status the account actually returned. Map those kinds to 401, 403 and 402 too. A status already recorded on the attempt still wins. Co-Authored-By: Claude Code * fix(usage): a leg refused before any send does not make the row unreported A multi-attempt row takes its usage status from aggregateAttemptUsage, which let any usage-less attempt mark the whole row "unreported". A combo leg the gateway refused locally (input admission, a local cooldown) never reached upstream and carries no usage, so a combo that then served the turn on its next leg with fully reported usage was still logged as unreported. The status now ignores usage-less attempts that never sent or were answered locally -- the same "spent nothing" rule summary.ts already prices by. An attempt that was sent and reported nothing still leaves the row unreported. The existing "folds partial and unsupported" test built its unreported 503 without a send; it now notes the send, which is the case it describes. Co-Authored-By: Claude Code * fix(usage): sum cache reads across attempts that report different fields aggregateAttemptUsage summed `cachedInputTokens` and `cacheReadInputTokens` independently. Anthropic usage sets both; an OpenAI-shaped attempt sets only `cachedInputTokens`. Row-level readers take `cacheReadInputTokens ?? cachedInputTokens` (request-log, management/shared.ts), so a row aggregating an Anthropic leg (60 read) and an OpenAI-shaped leg (40 cached) reported 60 cache reads instead of 100. Cost is priced per attempt and is not affected. Once any attempt reports `cacheReadInputTokens`, the aggregate's read field now counts each attempt's `cacheReadInputTokens ?? cachedInputTokens`. Rows whose attempts never report the read field are unchanged. Co-Authored-By: Claude Code * fix(messages-native): answer a spent Anthropic pool as a usage limit with its reset #6991 made the bridge lane answer a 429 as "All Anthropic OAuth accounts have reached their usage limit; the first one resets at