From 521cf5c42060a02e6123e684f5875c8ca41847ed Mon Sep 17 00:00:00 2001 From: nahuelb Date: Thu, 10 Sep 2026 01:50:35 -0300 Subject: [PATCH 1/7] feat: preserve Astra cache prefixes across reasoning effort changes --- .../docs/reference/configuration/server.md | 48 ++++ scripts/test-layout/layout.json | 1 + src/adapters/astra-effort-cache.ts | 170 ++++++++++++++ src/adapters/openai-responses.ts | 14 +- tests/fixtures/test-layout-expected.json | 1 + tests/responses/astra-effort-cache.test.ts | 208 ++++++++++++++++++ 6 files changed, 441 insertions(+), 1 deletion(-) create mode 100644 src/adapters/astra-effort-cache.ts create mode 100644 tests/responses/astra-effort-cache.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 6a3533f10c..b8095b8cd6 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -480,3 +480,51 @@ A hub that is reachable from a browser needs `hub.managementPublicOrigin` and at in `remoteGui.allowedTailscaleUsers`. Setting the origin without the user list produces a hub that advertises itself correctly and then refuses every session; setting the user list without the origin produces sessions pointed at whichever origin the request happened to use. + +## Experimental Astra effort cache preservation + +Set `OCX_ASTRA_EFFORT_CACHE=1` in the proxy process environment to opt in. The default is disabled. +This applies only to `gpt-6-astra` on the canonical ChatGPT Codex forward destination in standard, +single-agent mode. It does not enable the feature for Luna, Pro, public API destinations, or custom gateways. + +For a known conversation prefix, OpenCodex keeps the original request-level `reasoning.effort` +and inserts a `configuration_update` before the next user message when the requested effort changes. +It replays earlier updates in their original positions. Repeating an effort or retrying the same +request does not add another update. Switching back appends another update. +This preserves the earlier prefix structure; cache reuse still depends on backend caching and is not guaranteed. + +The caller must supply a distinct conversation identity through `thread-id` or +`client_metadata.thread_id`. A parent task ID, session ID, or shared prompt-cache key alone is +insufficient: side chats can share those values. Clients without a distinct identity continue with +their requested effort unchanged. Confirm an `updated` diagnostic before treating a Desktop client +as supported by this opt-in path. + +State lives under `$OPENCODEX_HOME/astra-effort-cache/` (normally `~/.opencodex/astra-effort-cache/`). +Files contain hashed prefixes and envelope identities, effort values, and item positions. They contain +no conversation text, credentials, or raw account/task identifiers. State survives restart; a fork or +missing baseline starts a new baseline using the requested effort. Changed instructions or tools also +start a new baseline. Each conversation/account is limited to 256 request snapshots and 2 MiB of state; +requests exceeding those limits use their requested effort unchanged. Conflicting retries and missing +user boundaries reset history. A busy, corrupt, or unavailable state file causes unchanged fallback. + +Automatic context management, automatic truncation, multi-agent history, and compaction input disable +automatic rewriting. This includes `compaction_trigger` requests and histories containing compaction +items. OpenCodex does not change compaction settings to obtain cache hits. Explicit client-supplied +configuration updates remain client-managed and pass through unchanged. +The standalone `/responses/compact` path receives the client's history, without proxy-injected updates. +If clients supply updates themselves, that endpoint rejects them. OpenAI documents `compaction_trigger` +as an alternative, with a fresh update after compaction; automatic post-compaction rewriting is not +implemented by this opt-in path. + +Diagnostics tagged `[ocx:astra-effort-cache]` report a fixed status code, baseline, and effective effort. +Request and usage logs preserve requested effort and record effective effort separately from the +request-level wire value. The upstream response's `reasoning.effort` still reports the baseline, as +specified by OpenAI. `baseline_reset`, `missing_thread_identity`, `compaction`, and `unavailable_state` +indicate that the optimization was not applied to that request. + +Unset `OCX_ASTRA_EFFORT_CACHE` or set it to `0` in the proxy process environment to disable rewriting. +After a coordinated restart, ordinary request-level effort behavior resumes. Retained state files may +be removed while the proxy is stopped. Do not share one thread identity across independent conversations. + +See OpenAI's [reasoning update compatibility](https://developers.openai.com/api/docs/guides/reasoning#change-reasoning-mid-conversation) +and [prompt caching guidance](https://developers.openai.com/api/docs/guides/prompt-caching#change-reasoning-effort-without-rewriting-the-prefix). diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index b5ea45c4a3..6e7722941e 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1057,6 +1057,7 @@ "responses-forward-dangling-call.test.ts": "responses", "responses-forward-incomplete-quota.test.ts": "responses", "responses-forward-posit-continuation.test.ts": "responses", + "astra-effort-cache.test.ts": "responses", "responses-forward-prompt-envelope.test.ts": "responses", "responses-function-tool-repair.test.ts": "responses", "responses-image-gen-repair.test.ts": "responses", diff --git a/src/adapters/astra-effort-cache.ts b/src/adapters/astra-effort-cache.ts new file mode 100644 index 0000000000..d65d3fc67a --- /dev/null +++ b/src/adapters/astra-effort-cache.ts @@ -0,0 +1,170 @@ +import { createHash } from "node:crypto"; +import { mkdirSync, readFileSync, rmdirSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { getConfigDir } from "../config/paths"; +import { atomicWriteFile } from "../config/atomic-write"; +import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; + +const EFFORTS = new Set(["low", "medium", "high", "xhigh", "max"]); +const MAX_SNAPSHOTS = 256; +const MAX_ITEMS = 20_000; +const MAX_STATE_BYTES = 2 * 1024 * 1024; + +type RecordValue = Record; +type Update = { position: number; effort: string }; +type Snapshot = { prefix: string; length: number; envelope: string; baseline: string; effective: string; updates: Update[] }; +type State = { version: 1; snapshots: Snapshot[] }; + +export interface AstraEffortResult { + body: unknown; + status: string; + baseline?: string; + effective?: string; +} + +function record(value: unknown): value is RecordValue { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function effort(value: unknown): value is string { + return typeof value === "string" && EFFORTS.has(value); +} + +function digest(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +function validState(value: unknown): value is State { + if (!record(value) || value.version !== 1 || !Array.isArray(value.snapshots) || value.snapshots.length > MAX_SNAPSHOTS) return false; + return value.snapshots.every(s => record(s) && typeof s.prefix === "string" && /^[a-f0-9]{64}$/.test(s.prefix) + && typeof s.envelope === "string" && /^[a-f0-9]{64}$/.test(s.envelope) + && Number.isSafeInteger(s.length) && Number(s.length) > 0 && Number(s.length) <= MAX_ITEMS + && effort(s.baseline) && effort(s.effective) && Array.isArray(s.updates) && s.updates.length <= Number(s.length) + && s.updates.every((u, i, all) => record(u) && Number.isSafeInteger(u.position) && Number(u.position) >= 0 + && Number(u.position) < Number(s.length) && effort(u.effort) + && (i === 0 || Number(u.position) > Number(all[i - 1].position))) + && (s.updates.at(-1)?.effort ?? s.baseline) === s.effective); +} + +function unsupported(body: RecordValue, original: unknown, headers: Headers): string | undefined { + if (body.model !== "gpt-6-astra") return "unsupported_model"; + if (!record(body.reasoning) || !effort(body.reasoning.effort)) return "unsupported_effort"; + if (body.reasoning.mode !== undefined && body.reasoning.mode !== "standard") return "unsupported_mode"; + if (headers.has("x-openai-subagent") || body.multi_agent !== undefined || body.agents !== undefined) return "multi_agent"; + if (record(original) && (original.truncation === "auto" || original.context_management !== undefined)) return "automatic_context_management"; + if (!Array.isArray(body.input) || body.input.length === 0 || body.input.length > MAX_ITEMS) return "unsupported_input"; + if (body.input.some(i => !record(i))) return "unsupported_input"; + if (body.input.some(i => ["compaction", "context_compaction", "compaction_trigger"].includes(i.type))) return "compaction"; + if (body.input.some(i => ["agent_message", "multi_agent_call", "multi_agent_call_output"].includes(i.type))) return "multi_agent"; + if (Array.isArray(body.tools) && body.tools.some(t => record(t) && t.type === "multi_agent")) return "multi_agent"; + return undefined; +} + +function transform(body: RecordValue, state: State): AstraEffortResult & { snapshot?: Snapshot } { + const input = body.input as RecordValue[]; + const requested = (body.reasoning as RecordValue).effort as string; + const { input: _input, reasoning, stream: _stream, client_metadata: _client, metadata: _metadata, ...envelope } = body; + const { effort: _effort, ...reasoningRest } = reasoning as RecordValue; + const envelopeHash = digest(JSON.stringify({ ...envelope, reasoning: reasoningRest })); + const prefixes = [digest("")]; + for (const item of input) prefixes.push(digest(prefixes.at(-1)! + JSON.stringify(item))); + const prefix = prefixes.at(-1)!; + const candidates = state.snapshots.filter(s => s.envelope === envelopeHash && s.length <= input.length && prefixes[s.length] === s.prefix); + const longest = Math.max(0, ...candidates.map(s => s.length)); + const matches = candidates.filter(s => s.length === longest); + const histories = new Set(matches.map(s => JSON.stringify([s.baseline, s.effective, s.updates]))); + if (histories.size > 1) return { body, status: "ambiguous_history", effective: requested, baseline: requested }; + const prior = matches[0]; + if (prior?.length === input.length && prior.effective !== requested) { + return { body, status: "conflicting_retry", effective: requested, baseline: requested }; + } + if (prior?.updates.some(u => input[u.position]?.role !== "user" + || (input[u.position]?.type !== undefined && input[u.position]?.type !== "message"))) { + return { body, status: "invalid_state", effective: requested, baseline: requested }; + } + const baseline = prior?.baseline ?? requested; + const updates = prior ? prior.updates.map(u => ({ ...u })) : []; + let status = prior ? "replay" : "baseline_reset"; + if (prior && prior.effective !== requested) { + const nextUser = input.findLastIndex((item, i) => i >= prior.length && item.role === "user" && (item.type === undefined || item.type === "message")); + if (nextUser < 0 || input.slice(nextUser + 1).some(item => + !["user", "developer", "system"].includes(String(item.role)) || (item.type !== undefined && item.type !== "message"))) { + return { body, status: "missing_user_boundary", effective: requested, baseline: requested }; + } + updates.push({ position: nextUser, effort: requested }); + status = "updated"; + } + const output: unknown[] = []; + let updateIndex = 0; + for (let i = 0; i < input.length; i++) { + const update = updates[updateIndex]; + if (update?.position === i) { + output.push({ type: "configuration_update", reasoning: { effort: update.effort } }); + updateIndex++; + } + output.push(input[i]); + } + const snapshot: Snapshot = { prefix, length: input.length, envelope: envelopeHash, baseline, effective: requested, updates }; + return { body: updates.length ? { ...body, reasoning: { ...(reasoning as RecordValue), effort: baseline }, input: output } : body, + status, baseline, effective: requested, snapshot }; +} + +export function applyAstraEffortCache( + body: unknown, + original: unknown, + headers: Headers, + servingHeaders: Headers, + directory = join(getConfigDir(), "astra-effort-cache"), +): AstraEffortResult { + if (!record(body)) return { body, status: "unsupported_input" }; + const requested = record(body.reasoning) && effort(body.reasoning.effort) ? body.reasoning.effort : undefined; + const fallback = (status: string): AstraEffortResult => ({ body, status, baseline: requested, effective: requested }); + if (Array.isArray(body.input) && body.input.some(i => record(i) && i.type === "configuration_update")) { + const last = body.input.findLast(i => record(i) && i.type === "configuration_update"); + return { ...fallback("client_managed"), effective: record(last?.reasoning) && effort(last.reasoning.effort) ? last.reasoning.effort : undefined }; + } + const reason = unsupported(body, original, headers); + if (reason) return fallback(reason); + const metadata = record(body.client_metadata) ? body.client_metadata : {}; + const thread = headers.get("thread-id")?.trim() || (typeof metadata.thread_id === "string" ? metadata.thread_id.trim() : ""); + if (!thread || thread.length > 256) return fallback("missing_thread_identity"); + const account = servingHeaders.get("chatgpt-account-id"); + if (!account) return fallback("missing_serving_identity"); + const scope = digest(JSON.stringify([thread, account])); + const path = join(directory, `${scope}.json`); + const lock = join(directory, `${scope}.lock`); + let locked = false; + try { + assertNotRealHomeUnderTest(directory); + mkdirSync(directory, { recursive: true, mode: 0o700 }); + mkdirSync(lock, { mode: 0o700 }); + locked = true; + let state: State = { version: 1, snapshots: [] }; + try { + if (statSync(path).size > MAX_STATE_BYTES) return fallback("state_limit"); + const loaded: unknown = JSON.parse(readFileSync(path, "utf8")); + if (!validState(loaded)) return fallback("invalid_state"); + state = loaded; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") return fallback("unavailable_state"); + } + const result = transform(body, state); + if (!result.snapshot) { + atomicWriteFile(path, JSON.stringify({ version: 1, snapshots: [] })); + return result; + } + const snapshot = result.snapshot; + if (!state.snapshots.some(s => JSON.stringify(s) === JSON.stringify(snapshot))) { + if (state.snapshots.length >= MAX_SNAPSHOTS) return fallback("state_limit"); + state.snapshots.push(snapshot); + const serialized = JSON.stringify(state); + if (serialized.length > MAX_STATE_BYTES) return fallback("state_limit"); + atomicWriteFile(path, serialized); + } + return { body: result.body, status: result.status, baseline: result.baseline, effective: result.effective }; + } catch { + return fallback("unavailable_state"); + } finally { + if (locked) { try { rmdirSync(lock); } catch {} } + } +} diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index c4aa523ee6..cdaf0ac931 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -1,3 +1,4 @@ +import { applyAstraEffortCache } from "./astra-effort-cache"; import { normalizeRoutedAgentMessages } from "./routed-agent-messages"; import { normalizeOpenCodeGoAdditionalTools } from "./opencode-go-additional-tools"; import { isXaiResponsesDestination } from "../providers/xai-transport"; @@ -2505,7 +2506,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): ), isXaiSchemaTarget(provider), ); - const finalBody = stripDisabledVerbosity( + let finalBody = stripDisabledVerbosity( stripDisabledReasoningSummaries( normalizeConfiguredReasoningSummaryDelivery(sanitizedBody, provider, parsed.modelId), provider, @@ -2514,6 +2515,16 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): provider, parsed.modelId, ); + let astraReasoningLog: { effectiveEffort: string; wireField: "reasoning.effort"; wireValue: string } | undefined; + if (isCanonicalOpenAiForwardProvider(provider) && process.env["OCX_ASTRA_EFFORT_CACHE"] === "1") { + const effortResult = applyAstraEffortCache(finalBody, parsed._rawBody, incoming.headers, new Headers(headers)); + finalBody = effortResult.body; + if (effortResult.baseline && effortResult.effective) { + astraReasoningLog = { effectiveEffort: effortResult.effective, wireField: "reasoning.effort", wireValue: effortResult.baseline }; + } + console.info("[ocx:astra-effort-cache]", JSON.stringify({ status: effortResult.status, + baseline: effortResult.baseline, effective: effortResult.effective })); + } if (isCanonicalOpenAiForwardProvider(provider)) { // Spark closes Responses Lite streams before a terminal completion. Select compatibility // from the final wire model so aliases cannot leave the caller or a static header enabled. @@ -2557,6 +2568,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): ...(convertedRoutedToolSearchNames ? { convertedRoutedToolSearchNames } : {}), ...(convertedRoutedNamespaceToolAliases ? { convertedRoutedNamespaceToolAliases } : {}), ...(tierLog ? { tierLog } : {}), + ...(astraReasoningLog ? { reasoningLog: astraReasoningLog } : {}), }; }, diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d772205061..e07a56951a 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -892,6 +892,7 @@ "responses-forward-dangling-call.test.ts": "responses", "responses-forward-incomplete-quota.test.ts": "responses", "responses-forward-posit-continuation.test.ts": "responses", + "astra-effort-cache.test.ts": "responses", "responses-forward-prompt-envelope.test.ts": "responses", "responses-function-tool-repair.test.ts": "responses", "responses-image-gen-repair.test.ts": "responses", diff --git a/tests/responses/astra-effort-cache.test.ts b/tests/responses/astra-effort-cache.test.ts new file mode 100644 index 0000000000..a20748cd20 --- /dev/null +++ b/tests/responses/astra-effort-cache.test.ts @@ -0,0 +1,208 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync, mkdirSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { applyAstraEffortCache } from "../../src/adapters/astra-effort-cache"; +import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses"; +import { parseRequest } from "../../src/responses/parser"; +import { prepareCodexWsRequest, CODEX_RESPONSES_HTTP_URL } from "../../src/server/responses/codex-ws-request"; +import { recordAdapterReasoning, applyResponseLogMetadata, type RequestLogContext } from "../../src/server/request-log"; +import { withTestTranslatorBudget } from "../helpers/translator-budget"; + +let directory: string; +let oldHome: string | undefined; +let oldFlag: string | undefined; +beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), "astra-effort-")); + oldHome = process.env["OPENCODEX_HOME"]; + oldFlag = process.env["OCX_ASTRA_EFFORT_CACHE"]; + process.env["OPENCODEX_HOME"] = directory; + delete process.env["OCX_ASTRA_EFFORT_CACHE"]; +}); +afterEach(() => { + if (oldHome === undefined) delete process.env["OPENCODEX_HOME"]; else process.env["OPENCODEX_HOME"] = oldHome; + if (oldFlag === undefined) delete process.env["OCX_ASTRA_EFFORT_CACHE"]; else process.env["OCX_ASTRA_EFFORT_CACHE"] = oldFlag; + rmSync(directory, { recursive: true, force: true }); +}); +const user = (content: string) => ({ role: "user", content }); +const assistant = (content: string) => ({ type: "message", role: "assistant", content: [{ type: "output_text", text: content }] }); +const update = (effort: string) => ({ type: "configuration_update", reasoning: { effort } }); +const first = [user("Synthetic first turn")]; +const second = [...first, assistant("OK"), user("Synthetic second turn")]; +const third = [...second, assistant("OK"), user("Synthetic third turn")]; +function body(input: unknown[], effort = "medium", extra = {}) { + return { model: "gpt-6-astra", instructions: "Synthetic instructions", input, reasoning: { effort }, store: false, stream: true, ...extra }; +} +function run(input: unknown[], effort = "medium", thread = "thread-a", extra = {}, account = "test-account") { + const raw = body(input, effort, extra); + return applyAstraEffortCache(raw, raw, new Headers({ "thread-id": thread, "session-id": "shared-cache-key" }), + new Headers({ "chatgpt-account-id": account }), join(directory, "state")); +} +function statePath() { return join(directory, "state", readdirSync(join(directory, "state")).find(n => n.endsWith(".json"))!); } + +describe("Astra effort history", () => { + test("pins baseline, appends at the user boundary, and replays immutable updates", () => { + expect(run(first).status).toBe("baseline_reset"); + const low = run(second, "low"); + expect(low.body).toEqual(body([...first, assistant("OK"), update("low"), user("Synthetic second turn")])); + expect(low).toMatchObject({ baseline: "medium", effective: "low", status: "updated" }); + const replay = run(third, "low"); + expect((replay.body as any).input).toEqual([...(low.body as any).input, assistant("OK"), user("Synthetic third turn")]); + expect(second).toHaveLength(3); + }); + test("switches back using a second ordered update", () => { + run(first); run(second, "low"); + expect((run(third, "medium").body as any).input).toEqual([...first, assistant("OK"), update("low"), user("Synthetic second turn"), assistant("OK"), update("medium"), user("Synthetic third turn")]); + }); + test("same effort never adds an update", () => { + run(first); + expect(run(second).body).toEqual(body(second)); + }); + test("identical retries neither duplicate updates nor add state", () => { + run(first); const low = run(second, "low"); + const before = readFileSync(statePath(), "utf8"); + expect(run(second, "low")).toEqual(low.status === "updated" ? { ...low, status: "replay" } : low); + expect(readFileSync(statePath(), "utf8")).toBe(before); + }); + test("restart/resume reads disk state without process memory", () => { + run(first); run(second, "low"); + const loaded = JSON.parse(readFileSync(statePath(), "utf8")); + expect(loaded.snapshots[1].updates).toEqual([{ position: 2, effort: "low" }]); + expect((run(third, "high").body as any).reasoning.effort).toBe("medium"); + }); + test("missing history establishes a new baseline without guessing", () => { + expect(run(second, "low")).toMatchObject({ body: body(second, "low"), status: "baseline_reset", baseline: "low" }); + expect((run(third, "high").body as any).reasoning.effort).toBe("low"); + }); + test("forks and accounts do not borrow a sibling baseline", () => { + run(first); run(second, "low"); + expect(run(third, "high", "thread-b").status).toBe("baseline_reset"); + expect(run(third, "high", "thread-a", {}, "other-account").status).toBe("baseline_reset"); + }); + test("interleaved branches replay only their matching prefix", () => { + run(first); + run(second, "low"); + const sibling = [...first, assistant("Other answer"), user("Sibling")]; + const siblingLow = run(sibling, "high"); + expect((siblingLow.body as any).input).toEqual([...first, assistant("Other answer"), update("high"), user("Sibling")]); + expect((run(third, "low").body as any).input.filter((i: any) => i.type === "configuration_update")).toEqual([update("low")]); + }); + test("conflicting same-input retries use the requested effort and reset history", () => { + run(first); run(second, "low"); + expect(run(second, "high")).toMatchObject({ body: body(second, "high"), status: "conflicting_retry" }); + expect(run(third, "high").status).toBe("baseline_reset"); + }); + test("effort changes during tool continuation fall back without retaining old effort", () => { + run(first); run(second, "low"); + const continued = [...second, { type: "function_call", name: "test", call_id: "x", arguments: "{}" }, { type: "function_call_output", call_id: "x", output: "OK" }]; + expect(run(continued, "high")).toMatchObject({ body: body(continued, "high"), status: "missing_user_boundary" }); + }); + test("a newly observed historical user is not a current turn boundary", () => { + run(first); + const continued = [...second, assistant("Already answered"), { type: "function_call_output", call_id: "x", output: "OK" }]; + expect(run(continued, "high")).toMatchObject({ body: body(continued, "high"), status: "missing_user_boundary" }); + }); + test("an effort change belongs before the latest unseen user", () => { + run(first); + expect((run(third, "low").body as any).input).toEqual([...third.slice(0, -1), update("low"), third.at(-1)]); + }); + test("changed prefix and changed tools reset the baseline", () => { + run(first); run(second, "low"); + expect(run([user("Changed first"), ...third.slice(1)], "high").status).toBe("baseline_reset"); + expect(run(third, "high", "thread-a", { tools: [{ type: "function", name: "new_tool" }] }).status).toBe("baseline_reset"); + }); + test("state contains hashes and effort positions, never prompt or account data", () => { + run(first); run(second, "low"); + const state = readFileSync(statePath(), "utf8"); + for (const forbidden of ["Synthetic", "test-account", "thread-a", "shared-cache-key", "content", "authorization"]) expect(state).not.toContain(forbidden); + }); + test("corrupt state fails transparently without overwriting it", () => { + run(first); writeFileSync(statePath(), "invalid"); + expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low") }); + expect(readFileSync(statePath(), "utf8")).toBe("invalid"); + }); + test("invalid update positions fail state validation", () => { + run(first); run(second, "low"); + const saved = JSON.parse(readFileSync(statePath(), "utf8")); + saved.snapshots[1].updates[0].position = 99; + writeFileSync(statePath(), JSON.stringify(saved)); + expect(run(third, "low").status).toBe("invalid_state"); + }); + test("stored updates must still point to user messages", () => { + run(first); run(second, "low"); + const saved = JSON.parse(readFileSync(statePath(), "utf8")); + saved.snapshots[1].updates[0].position = 1; + writeFileSync(statePath(), JSON.stringify(saved)); + expect(run(third, "low")).toMatchObject({ status: "invalid_state", body: body(third, "low") }); + }); + test("a concurrent writer lock causes unchanged fallback", () => { + run(first); mkdirSync(statePath().replace(/\.json$/, ".lock")); + expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low") }); + }); + test.each(["compaction", "context_compaction", "compaction_trigger"])("%s disables rewriting", type => { + run(first); run(second, "low"); + const input = [...third, { type }]; + expect(run(input, "high")).toMatchObject({ status: "compaction", body: body(input, "high") }); + }); + test.each([ + [{ model: "gpt-5.6-luna" }, "unsupported_model"], + [{ reasoning: { effort: "ultra" } }, "unsupported_effort"], + [{ reasoning: { effort: "high", mode: "pro" } }, "unsupported_mode"], + [{ truncation: "auto" }, "automatic_context_management"], + [{ context_management: [{ type: "compaction", compact_threshold: 1000 }] }, "automatic_context_management"], + [{ multi_agent: {} }, "multi_agent"], + ])("unsupported settings retain the request: %j", (extra, status) => { + expect(run(first, "medium", "thread-a", extra as Record)).toMatchObject({ status, body: body(first, "medium", extra) }); + }); + test("explicit client updates pass through and report the last effective effort", () => { + const input = [...first, update("low"), user("next")]; + expect(run(input)).toMatchObject({ body: body(input), status: "client_managed", baseline: "medium", effective: "low" }); + }); + test("parent and cache identity alone cannot identify a side chat", () => { + const raw = body(first); + expect(applyAstraEffortCache(raw, raw, new Headers({ "x-codex-parent-thread-id": "parent", "session-id": "shared" }), new Headers({ "chatgpt-account-id": "account" }), directory).status).toBe("missing_thread_identity"); + }); +}); + +const provider = { adapter: "openai-responses" as const, authMode: "forward" as const, baseUrl: "https://chatgpt.com/backend-api/codex" }; +function adapterRequest(input: unknown[], effort = "medium", extra = {}, destination = provider) { + const adapter = withTestTranslatorBudget(createResponsesPassthroughAdapter(destination)); + return adapter.buildRequest(parseRequest(body(input, effort, extra)), { headers: new Headers({ authorization: "Bearer synthetic", "chatgpt-account-id": "test-account", "thread-id": "thread-a" }) }); +} +describe("Astra adapter integration", () => { + test("default remains disabled and performs no state writes", () => { + adapterRequest(first); const request = adapterRequest(second, "low"); + expect(JSON.parse(request.body).reasoning.effort).toBe("low"); + expect(readdirSync(directory)).toEqual([]); + }); + test("parser, native adapter, websocket framing and logs preserve the update", () => { + process.env["OCX_ASTRA_EFFORT_CACHE"] = "1"; + adapterRequest(first); + const request = adapterRequest(second, "low"); + const output = JSON.parse(request.body); + expect(output.reasoning.effort).toBe("medium"); + expect(output.input.filter((i: any) => i.type === "configuration_update")).toEqual([update("low")]); + expect(parseRequest(output)._rawBody).toEqual(output); + const ws = prepareCodexWsRequest(CODEX_RESPONSES_HTTP_URL, { headers: request.headers, body: request.body }); + expect(JSON.parse(ws!.frameText).input).toEqual(output.input); + expect(request.reasoningLog).toEqual({ effectiveEffort: "low", wireField: "reasoning.effort", wireValue: "medium" }); + const log = { requestedEffort: "low" } as RequestLogContext; + recordAdapterReasoning(log, request); + applyResponseLogMetadata(log, { reasoning: { effort: "medium" } }); + expect(log).toMatchObject({ requestedEffort: "low", effectiveEffort: "low", reasoningWireValue: "medium" }); + }); + test("auto truncation is detected before native parameter stripping", () => { + process.env["OCX_ASTRA_EFFORT_CACHE"] = "1"; + adapterRequest(first); + const request = adapterRequest(second, "low", { truncation: "auto" }); + expect(JSON.parse(request.body).reasoning.effort).toBe("low"); + expect(JSON.parse(request.body).input.some((i: any) => i.type === "configuration_update")).toBe(false); + }); + test("custom forward destinations never receive generated updates", () => { + process.env["OCX_ASTRA_EFFORT_CACHE"] = "1"; + const destination = { ...provider, baseUrl: "https://gateway.example.test" }; + adapterRequest(first, "medium", {}, destination); + expect(JSON.parse(adapterRequest(second, "low", {}, destination).body).reasoning.effort).toBe("low"); + expect(readdirSync(directory)).toEqual([]); + }); +}); From 7f3cece76048cc63c17739559d90dca101755942 Mon Sep 17 00:00:00 2001 From: nahuelb Date: Thu, 10 Sep 2026 15:33:47 -0300 Subject: [PATCH 2/7] fix: bound persistent effort state and recover crash-held locks --- .../docs/reference/configuration/server.md | 6 +- src/adapters/astra-effort-cache.ts | 55 +++++------- src/adapters/astra-effort-state.ts | 59 +++++++++++++ tests/responses/astra-effort-cache.test.ts | 86 ++++++++++++++++--- 4 files changed, 156 insertions(+), 50 deletions(-) create mode 100644 src/adapters/astra-effort-state.ts diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index b8095b8cd6..2f83794ea0 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -500,11 +500,11 @@ their requested effort unchanged. Confirm an `updated` diagnostic before treatin as supported by this opt-in path. State lives under `$OPENCODEX_HOME/astra-effort-cache/` (normally `~/.opencodex/astra-effort-cache/`). -Files contain hashed prefixes and envelope identities, effort values, and item positions. They contain -no conversation text, credentials, or raw account/task identifiers. State survives restart; a fork or +A private SQLite database contains hashed prefixes and envelope identities, effort values, and item positions. It contains +no conversation text, credentials, or raw account/task identifiers. SQLite releases locks when a process exits, including crashes. State survives restart; a fork or missing baseline starts a new baseline using the requested effort. Changed instructions or tools also start a new baseline. Each conversation/account is limited to 256 request snapshots and 2 MiB of state; -requests exceeding those limits use their requested effort unchanged. Conflicting retries and missing +requests exceeding those limits use their requested effort unchanged. Across conversations, the store retains at most 128 entries and 16 MiB of payload, evicting the least recently used entries. Entries expire after seven days without access; pruning runs on requests. The database is capped at 32 MiB, with a temporary rollback journal bounded by that size. The cache directory is registered once for uninstall cleanup. Conflicting retries and missing user boundaries reset history. A busy, corrupt, or unavailable state file causes unchanged fallback. Automatic context management, automatic truncation, multi-agent history, and compaction input disable diff --git a/src/adapters/astra-effort-cache.ts b/src/adapters/astra-effort-cache.ts index d65d3fc67a..1bd170a26b 100644 --- a/src/adapters/astra-effort-cache.ts +++ b/src/adapters/astra-effort-cache.ts @@ -1,9 +1,7 @@ import { createHash } from "node:crypto"; -import { mkdirSync, readFileSync, rmdirSync, statSync } from "node:fs"; import { join } from "node:path"; import { getConfigDir } from "../config/paths"; -import { atomicWriteFile } from "../config/atomic-write"; -import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; +import { withAstraEffortState } from "./astra-effort-state"; const EFFORTS = new Set(["low", "medium", "high", "xhigh", "max"]); const MAX_SNAPSHOTS = 256; @@ -131,40 +129,27 @@ export function applyAstraEffortCache( const account = servingHeaders.get("chatgpt-account-id"); if (!account) return fallback("missing_serving_identity"); const scope = digest(JSON.stringify([thread, account])); - const path = join(directory, `${scope}.json`); - const lock = join(directory, `${scope}.lock`); - let locked = false; try { - assertNotRealHomeUnderTest(directory); - mkdirSync(directory, { recursive: true, mode: 0o700 }); - mkdirSync(lock, { mode: 0o700 }); - locked = true; - let state: State = { version: 1, snapshots: [] }; - try { - if (statSync(path).size > MAX_STATE_BYTES) return fallback("state_limit"); - const loaded: unknown = JSON.parse(readFileSync(path, "utf8")); - if (!validState(loaded)) return fallback("invalid_state"); - state = loaded; - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ENOENT") return fallback("unavailable_state"); - } - const result = transform(body, state); - if (!result.snapshot) { - atomicWriteFile(path, JSON.stringify({ version: 1, snapshots: [] })); - return result; - } - const snapshot = result.snapshot; - if (!state.snapshots.some(s => JSON.stringify(s) === JSON.stringify(snapshot))) { - if (state.snapshots.length >= MAX_SNAPSHOTS) return fallback("state_limit"); - state.snapshots.push(snapshot); - const serialized = JSON.stringify(state); - if (serialized.length > MAX_STATE_BYTES) return fallback("state_limit"); - atomicWriteFile(path, serialized); - } - return { body: result.body, status: result.status, baseline: result.baseline, effective: result.effective }; + return withAstraEffortState(directory, scope, serialized => { + let state: State = { version: 1, snapshots: [] }; + if (serialized !== undefined) { + if (serialized.length > MAX_STATE_BYTES) return { value: fallback("state_limit") }; + const loaded: unknown = JSON.parse(serialized); + if (!validState(loaded)) return { value: fallback("invalid_state") }; + state = loaded; + } + const result = transform(body, state); + if (!result.snapshot) return { value: result, state: null }; + const snapshot = result.snapshot; + if (!state.snapshots.some(s => JSON.stringify(s) === JSON.stringify(snapshot))) { + if (state.snapshots.length >= MAX_SNAPSHOTS) return { value: fallback("state_limit") }; + state.snapshots.push(snapshot); + } + const next = JSON.stringify(state); + if (next.length > MAX_STATE_BYTES) return { value: fallback("state_limit") }; + return { value: { body: result.body, status: result.status, baseline: result.baseline, effective: result.effective }, state: next }; + }); } catch { return fallback("unavailable_state"); - } finally { - if (locked) { try { rmdirSync(lock); } catch {} } } } diff --git a/src/adapters/astra-effort-state.ts b/src/adapters/astra-effort-state.ts new file mode 100644 index 0000000000..167f8eff40 --- /dev/null +++ b/src/adapters/astra-effort-state.ts @@ -0,0 +1,59 @@ +import { Database } from "bun:sqlite"; +import { chmodSync, closeSync, lstatSync, mkdirSync, openSync } from "node:fs"; +import { join } from "node:path"; +import { getConfigDir } from "../config/paths"; +import { recordOwnedConfigPath } from "../lib/config-ownership"; +import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; +import { hardenSecretDir, hardenSecretPath } from "../lib/windows-secret-acl"; + +const MAX_DATABASE_BYTES = 32 * 1024 * 1024; +const MAX_PAYLOAD_BYTES = 16 * 1024 * 1024; +const MAX_CONVERSATIONS = 128; +const RETENTION_MS = 7 * 24 * 60 * 60 * 1000; + +export function withAstraEffortState( + directory: string, + scope: string, + work: (state: string | undefined) => { value: T; state?: string | null }, +): T { + assertNotRealHomeUnderTest(getConfigDir()); + assertNotRealHomeUnderTest(directory); + if (!recordOwnedConfigPath(getConfigDir(), directory)) throw new Error("Unowned effort state directory"); + mkdirSync(directory, { recursive: true, mode: 0o700 }); + if (!lstatSync(directory).isDirectory() || lstatSync(directory).isSymbolicLink()) throw new Error("Invalid effort state directory"); + if (process.platform !== "win32") chmodSync(directory, 0o700); + hardenSecretDir(directory, { required: true }); + const path = join(directory, "state.sqlite"); + try { closeSync(openSync(path, "wx", 0o600)); } + catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; } + const file = lstatSync(path); + if (!file.isFile() || file.isSymbolicLink() || file.nlink !== 1 || file.size > MAX_DATABASE_BYTES) throw new Error("Invalid effort state file"); + if (process.platform !== "win32") chmodSync(path, 0o600); + hardenSecretPath(path, { required: true }); + const db = new Database(path, { readwrite: true }); + try { + db.exec("PRAGMA busy_timeout = 0; PRAGMA journal_mode = DELETE; PRAGMA page_size = 4096; PRAGMA max_page_count = 8192"); + const pageSize = (db.query("PRAGMA page_size").get() as { page_size: number }).page_size; + db.exec(`PRAGMA max_page_count = ${Math.floor(MAX_DATABASE_BYTES / pageSize)}`); + db.exec("CREATE TABLE IF NOT EXISTS sessions (scope TEXT PRIMARY KEY, state TEXT NOT NULL CHECK(length(CAST(state AS BLOB)) <= 2097152), touched INTEGER NOT NULL)"); + return db.transaction(() => { + const now = Date.now(); + db.query("DELETE FROM sessions WHERE touched < ?").run(now - RETENTION_MS); + const row = db.query("SELECT state FROM sessions WHERE scope = ?").get(scope) as { state: string } | null; + const result = work(row?.state); + if (result.state === null) db.query("DELETE FROM sessions WHERE scope = ?").run(scope); + else if (result.state !== undefined) { + db.query("INSERT INTO sessions VALUES (?, ?, ?) ON CONFLICT(scope) DO UPDATE SET state = excluded.state, touched = excluded.touched") + .run(scope, result.state, now); + for (;;) { + const totals = db.query("SELECT COUNT(*) AS count, COALESCE(SUM(length(CAST(state AS BLOB))), 0) AS bytes FROM sessions").get() as { count: number; bytes: number }; + if (totals.count <= MAX_CONVERSATIONS && totals.bytes <= MAX_PAYLOAD_BYTES) break; + db.query("DELETE FROM sessions WHERE scope = (SELECT scope FROM sessions WHERE scope != ? ORDER BY touched, scope LIMIT 1)").run(scope); + } + } + return result.value; + }).immediate(); + } finally { + db.close(); + } +} diff --git a/tests/responses/astra-effort-cache.test.ts b/tests/responses/astra-effort-cache.test.ts index a20748cd20..1fd3b43a0b 100644 --- a/tests/responses/astra-effort-cache.test.ts +++ b/tests/responses/astra-effort-cache.test.ts @@ -1,7 +1,10 @@ +import { Database } from "bun:sqlite"; import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync, mkdirSync } from "node:fs"; +import { mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync, mkdirSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { withAstraEffortState } from "../../src/adapters/astra-effort-state"; +import { CONFIG_UNINSTALL_MANIFEST, removeOwnedConfigState } from "../../src/lib/config-ownership"; import { applyAstraEffortCache } from "../../src/adapters/astra-effort-cache"; import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses"; import { parseRequest } from "../../src/responses/parser"; @@ -38,7 +41,15 @@ function run(input: unknown[], effort = "medium", thread = "thread-a", extra = { return applyAstraEffortCache(raw, raw, new Headers({ "thread-id": thread, "session-id": "shared-cache-key" }), new Headers({ "chatgpt-account-id": account }), join(directory, "state")); } -function statePath() { return join(directory, "state", readdirSync(join(directory, "state")).find(n => n.endsWith(".json"))!); } +function statePath() { return join(directory, "state", "state.sqlite"); } +function readState() { + const db = new Database(statePath()); + try { return (db.query("SELECT state FROM sessions LIMIT 1").get() as { state: string }).state; } finally { db.close(); } +} +function writeState(state: string) { + const db = new Database(statePath()); + try { db.query("UPDATE sessions SET state = ?").run(state); } finally { db.close(); } +} describe("Astra effort history", () => { test("pins baseline, appends at the user boundary, and replays immutable updates", () => { @@ -60,13 +71,13 @@ describe("Astra effort history", () => { }); test("identical retries neither duplicate updates nor add state", () => { run(first); const low = run(second, "low"); - const before = readFileSync(statePath(), "utf8"); + const before = readState(); expect(run(second, "low")).toEqual(low.status === "updated" ? { ...low, status: "replay" } : low); - expect(readFileSync(statePath(), "utf8")).toBe(before); + expect(readState()).toBe(before); }); test("restart/resume reads disk state without process memory", () => { run(first); run(second, "low"); - const loaded = JSON.parse(readFileSync(statePath(), "utf8")); + const loaded = JSON.parse(readState()); expect(loaded.snapshots[1].updates).toEqual([{ position: 2, effort: "low" }]); expect((run(third, "high").body as any).reasoning.effort).toBe("medium"); }); @@ -113,7 +124,7 @@ describe("Astra effort history", () => { }); test("state contains hashes and effort positions, never prompt or account data", () => { run(first); run(second, "low"); - const state = readFileSync(statePath(), "utf8"); + const state = readState(); for (const forbidden of ["Synthetic", "test-account", "thread-a", "shared-cache-key", "content", "authorization"]) expect(state).not.toContain(forbidden); }); test("corrupt state fails transparently without overwriting it", () => { @@ -123,21 +134,25 @@ describe("Astra effort history", () => { }); test("invalid update positions fail state validation", () => { run(first); run(second, "low"); - const saved = JSON.parse(readFileSync(statePath(), "utf8")); + const saved = JSON.parse(readState()); saved.snapshots[1].updates[0].position = 99; - writeFileSync(statePath(), JSON.stringify(saved)); + writeState(JSON.stringify(saved)); expect(run(third, "low").status).toBe("invalid_state"); }); test("stored updates must still point to user messages", () => { run(first); run(second, "low"); - const saved = JSON.parse(readFileSync(statePath(), "utf8")); + const saved = JSON.parse(readState()); saved.snapshots[1].updates[0].position = 1; - writeFileSync(statePath(), JSON.stringify(saved)); + writeState(JSON.stringify(saved)); expect(run(third, "low")).toMatchObject({ status: "invalid_state", body: body(third, "low") }); }); test("a concurrent writer lock causes unchanged fallback", () => { - run(first); mkdirSync(statePath().replace(/\.json$/, ".lock")); - expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low") }); + run(first); + const lock = new Database(statePath()); + lock.exec("BEGIN IMMEDIATE"); + try { expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low") }); } + finally { lock.exec("ROLLBACK"); lock.close(); } + expect(run(second, "low").status).toBe("updated"); }); test.each(["compaction", "context_compaction", "compaction_trigger"])("%s disables rewriting", type => { run(first); run(second, "low"); @@ -206,3 +221,50 @@ describe("Astra adapter integration", () => { expect(readdirSync(directory)).toEqual([]); }); }); + +describe("durable effort state lifecycle", () => { + test("bounds task churn and registers one removable directory", () => { + for (let i = 0; i < 150; i++) run(first, "medium", `task-${i}`); + const db = new Database(statePath()); + try { expect((db.query("SELECT COUNT(*) AS count FROM sessions").get() as { count: number }).count).toBe(128); } + finally { db.close(); } + const manifest = JSON.parse(readFileSync(join(directory, CONFIG_UNINSTALL_MANIFEST), "utf8")); + expect(manifest.paths.filter((p: string) => p === "state" || p.startsWith("state/"))).toEqual(["state"]); + expect(removeOwnedConfigState(directory).status).toBe("removed"); + }); + test("bounds total payload bytes and expires abandoned conversations", () => { + const path = join(directory, "state"); + const payload = "x".repeat(1_500_000); + for (let i = 0; i < 14; i++) withAstraEffortState(path, String(i), () => ({ value: true, state: payload })); + const db = new Database(statePath()); + try { + expect((db.query("SELECT SUM(length(state)) AS bytes FROM sessions").get() as { bytes: number }).bytes).toBeLessThanOrEqual(16 * 1024 * 1024); + db.query("UPDATE sessions SET touched = 0").run(); + } finally { db.close(); } + expect(statSync(statePath()).size).toBeLessThanOrEqual(32 * 1024 * 1024); + withAstraEffortState(path, "fresh", () => ({ value: true, state: "fresh" })); + const fresh = new Database(statePath()); + try { expect(fresh.query("SELECT scope FROM sessions").all()).toEqual([{ scope: "fresh" }]); } + finally { fresh.close(); } + }); + test("recovers after a different process dies while holding a transaction", async () => { + run(first); + const child = Bun.spawn([process.execPath, "-e", `import { Database } from "bun:sqlite"; + const db = new Database(process.argv[1]); db.exec("BEGIN IMMEDIATE; UPDATE sessions SET state = 'uncommitted'"); + console.log("held"); await new Promise(() => {});`, statePath()], { stdout: "pipe", stderr: "pipe" }); + try { + const reader = child.stdout.getReader(); + const ready = await reader.read(); reader.releaseLock(); + expect(new TextDecoder().decode(ready.value)).toContain("held"); + expect(run(second, "low").status).toBe("unavailable_state"); + } finally { child.kill("SIGKILL"); await child.exited; } + expect(run(second, "low")).toMatchObject({ status: "updated", baseline: "medium", effective: "low" }); + }); + test("creates owner-only database and directory on POSIX", () => { + run(first); + if (process.platform !== "win32") { + expect(statSync(statePath()).mode & 0o777).toBe(0o600); + expect(statSync(join(directory, "state")).mode & 0o777).toBe(0o700); + } + }); +}); From df78d400aefdeae52113e15c0cac49977971502e Mon Sep 17 00:00:00 2001 From: nahuelb Date: Thu, 10 Sep 2026 16:39:07 -0300 Subject: [PATCH 3/7] fix: address Astra cache review diagnostics and test reliability --- .../docs/reference/configuration/server.md | 4 +++- src/adapters/astra-effort-cache.ts | 4 +++- src/adapters/openai-responses.ts | 4 ++-- tests/responses/astra-effort-cache.test.ts | 18 +++++++++++++----- 4 files changed, 21 insertions(+), 9 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 2f83794ea0..c5c23f9673 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -516,7 +516,9 @@ If clients supply updates themselves, that endpoint rejects them. OpenAI documen as an alternative, with a fresh update after compaction; automatic post-compaction rewriting is not implemented by this opt-in path. -Diagnostics tagged `[ocx:astra-effort-cache]` report a fixed status code, baseline, and effective effort. +Enable provider diagnostics with `ocx debug provider on` or `OCX_DEBUG=1`. Diagnostics tagged +`[ocx:openai-responses:astra-effort-cache]` report a fixed status code, baseline, and effective effort +through the shared debug buffer and stderr output. Request and usage logs preserve requested effort and record effective effort separately from the request-level wire value. The upstream response's `reasoning.effort` still reports the baseline, as specified by OpenAI. `baseline_reset`, `missing_thread_identity`, `compaction`, and `unavailable_state` diff --git a/src/adapters/astra-effort-cache.ts b/src/adapters/astra-effort-cache.ts index 1bd170a26b..262cad2a30 100644 --- a/src/adapters/astra-effort-cache.ts +++ b/src/adapters/astra-effort-cache.ts @@ -3,6 +3,8 @@ import { join } from "node:path"; import { getConfigDir } from "../config/paths"; import { withAstraEffortState } from "./astra-effort-state"; +// Explicit protocol gate; keep aligned with PROVIDER_REGISTRY openai-apikey.models/modelReasoningEfforts["gpt-6-astra"]. +const ASTRA_MODEL = "gpt-6-astra"; const EFFORTS = new Set(["low", "medium", "high", "xhigh", "max"]); const MAX_SNAPSHOTS = 256; const MAX_ITEMS = 20_000; @@ -45,7 +47,7 @@ function validState(value: unknown): value is State { } function unsupported(body: RecordValue, original: unknown, headers: Headers): string | undefined { - if (body.model !== "gpt-6-astra") return "unsupported_model"; + if (body.model !== ASTRA_MODEL) return "unsupported_model"; if (!record(body.reasoning) || !effort(body.reasoning.effort)) return "unsupported_effort"; if (body.reasoning.mode !== undefined && body.reasoning.mode !== "standard") return "unsupported_mode"; if (headers.has("x-openai-subagent") || body.multi_agent !== undefined || body.agents !== undefined) return "multi_agent"; diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index cdaf0ac931..7050d82d44 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -2522,8 +2522,8 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): if (effortResult.baseline && effortResult.effective) { astraReasoningLog = { effectiveEffort: effortResult.effective, wireField: "reasoning.effort", wireValue: effortResult.baseline }; } - console.info("[ocx:astra-effort-cache]", JSON.stringify({ status: effortResult.status, - baseline: effortResult.baseline, effective: effortResult.effective })); + debugProviderDiagnostic("openai-responses", "astra-effort-cache", { status: effortResult.status, + baseline: effortResult.baseline, effective: effortResult.effective }); } if (isCanonicalOpenAiForwardProvider(provider)) { // Spark closes Responses Lite streams before a terminal completion. Select compatibility diff --git a/tests/responses/astra-effort-cache.test.ts b/tests/responses/astra-effort-cache.test.ts index 1fd3b43a0b..b027e3e776 100644 --- a/tests/responses/astra-effort-cache.test.ts +++ b/tests/responses/astra-effort-cache.test.ts @@ -71,8 +71,9 @@ describe("Astra effort history", () => { }); test("identical retries neither duplicate updates nor add state", () => { run(first); const low = run(second, "low"); + expect(low.status).toBe("updated"); const before = readState(); - expect(run(second, "low")).toEqual(low.status === "updated" ? { ...low, status: "replay" } : low); + expect(run(second, "low")).toEqual({ ...low, status: "replay" }); expect(readState()).toBe(before); }); test("restart/resume reads disk state without process memory", () => { @@ -234,11 +235,11 @@ describe("durable effort state lifecycle", () => { }); test("bounds total payload bytes and expires abandoned conversations", () => { const path = join(directory, "state"); - const payload = "x".repeat(1_500_000); + const payload = "é".repeat(750_000); for (let i = 0; i < 14; i++) withAstraEffortState(path, String(i), () => ({ value: true, state: payload })); const db = new Database(statePath()); try { - expect((db.query("SELECT SUM(length(state)) AS bytes FROM sessions").get() as { bytes: number }).bytes).toBeLessThanOrEqual(16 * 1024 * 1024); + expect((db.query("SELECT SUM(length(CAST(state AS BLOB))) AS bytes FROM sessions").get() as { bytes: number }).bytes).toBeLessThanOrEqual(16 * 1024 * 1024); db.query("UPDATE sessions SET touched = 0").run(); } finally { db.close(); } expect(statSync(statePath()).size).toBeLessThanOrEqual(32 * 1024 * 1024); @@ -254,8 +255,15 @@ describe("durable effort state lifecycle", () => { console.log("held"); await new Promise(() => {});`, statePath()], { stdout: "pipe", stderr: "pipe" }); try { const reader = child.stdout.getReader(); - const ready = await reader.read(); reader.releaseLock(); - expect(new TextDecoder().decode(ready.value)).toContain("held"); + const decoder = new TextDecoder(); + let output = ""; + try { + while (!output.includes("held")) { + const chunk = await reader.read(); + if (chunk.done) throw new Error("Child exited before acquiring its transaction"); + output += decoder.decode(chunk.value, { stream: true }); + } + } finally { reader.releaseLock(); } expect(run(second, "low").status).toBe("unavailable_state"); } finally { child.kill("SIGKILL"); await child.exited; } expect(run(second, "low")).toMatchObject({ status: "updated", baseline: "medium", effective: "low" }); From 82cf472a3810a9c649c69e116dce634edfacd9fb Mon Sep 17 00:00:00 2001 From: nahuelb Date: Thu, 10 Sep 2026 17:22:27 -0300 Subject: [PATCH 4/7] feat: automatically preserve supported Astra effort cache prefixes --- .../docs/reference/configuration/server.md | 14 +++++++------- src/adapters/astra-effort-cache.ts | 4 ++++ src/adapters/openai-responses.ts | 4 ++-- tests/responses/astra-effort-cache.test.ts | 19 ++++++++++--------- 4 files changed, 23 insertions(+), 18 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index c5c23f9673..4c4dcf89e4 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -481,9 +481,10 @@ in `remoteGui.allowedTailscaleUsers`. Setting the origin without the user list p advertises itself correctly and then refuses every session; setting the user list without the origin produces sessions pointed at whichever origin the request happened to use. -## Experimental Astra effort cache preservation +## Astra effort cache preservation -Set `OCX_ASTRA_EFFORT_CACHE=1` in the proxy process environment to opt in. The default is disabled. +Effort cache preservation runs automatically for supported requests. There is no enable/disable +setting and no configuration is required. This applies only to `gpt-6-astra` on the canonical ChatGPT Codex forward destination in standard, single-agent mode. It does not enable the feature for Luna, Pro, public API destinations, or custom gateways. @@ -497,7 +498,7 @@ The caller must supply a distinct conversation identity through `thread-id` or `client_metadata.thread_id`. A parent task ID, session ID, or shared prompt-cache key alone is insufficient: side chats can share those values. Clients without a distinct identity continue with their requested effort unchanged. Confirm an `updated` diagnostic before treating a Desktop client -as supported by this opt-in path. +as supported by this path. State lives under `$OPENCODEX_HOME/astra-effort-cache/` (normally `~/.opencodex/astra-effort-cache/`). A private SQLite database contains hashed prefixes and envelope identities, effort values, and item positions. It contains @@ -514,7 +515,7 @@ configuration updates remain client-managed and pass through unchanged. The standalone `/responses/compact` path receives the client's history, without proxy-injected updates. If clients supply updates themselves, that endpoint rejects them. OpenAI documents `compaction_trigger` as an alternative, with a fresh update after compaction; automatic post-compaction rewriting is not -implemented by this opt-in path. +implemented by this path. Enable provider diagnostics with `ocx debug provider on` or `OCX_DEBUG=1`. Diagnostics tagged `[ocx:openai-responses:astra-effort-cache]` report a fixed status code, baseline, and effective effort @@ -524,9 +525,8 @@ request-level wire value. The upstream response's `reasoning.effort` still repor specified by OpenAI. `baseline_reset`, `missing_thread_identity`, `compaction`, and `unavailable_state` indicate that the optimization was not applied to that request. -Unset `OCX_ASTRA_EFFORT_CACHE` or set it to `0` in the proxy process environment to disable rewriting. -After a coordinated restart, ordinary request-level effort behavior resumes. Retained state files may -be removed while the proxy is stopped. Do not share one thread identity across independent conversations. +Retained state files may be removed while the proxy is stopped. The next request establishes a new +baseline. Do not share one thread identity across independent conversations. See OpenAI's [reasoning update compatibility](https://developers.openai.com/api/docs/guides/reasoning#change-reasoning-mid-conversation) and [prompt caching guidance](https://developers.openai.com/api/docs/guides/prompt-caching#change-reasoning-effort-without-rewriting-the-prefix). diff --git a/src/adapters/astra-effort-cache.ts b/src/adapters/astra-effort-cache.ts index 262cad2a30..3fa9d250ff 100644 --- a/src/adapters/astra-effort-cache.ts +++ b/src/adapters/astra-effort-cache.ts @@ -26,6 +26,10 @@ function record(value: unknown): value is RecordValue { return value !== null && typeof value === "object" && !Array.isArray(value); } +export function supportsAstraEffortCache(body: unknown): boolean { + return record(body) && body.model === ASTRA_MODEL; +} + function effort(value: unknown): value is string { return typeof value === "string" && EFFORTS.has(value); } diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index 7050d82d44..8cb5e53178 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -1,4 +1,4 @@ -import { applyAstraEffortCache } from "./astra-effort-cache"; +import { applyAstraEffortCache, supportsAstraEffortCache } from "./astra-effort-cache"; import { normalizeRoutedAgentMessages } from "./routed-agent-messages"; import { normalizeOpenCodeGoAdditionalTools } from "./opencode-go-additional-tools"; import { isXaiResponsesDestination } from "../providers/xai-transport"; @@ -2516,7 +2516,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): parsed.modelId, ); let astraReasoningLog: { effectiveEffort: string; wireField: "reasoning.effort"; wireValue: string } | undefined; - if (isCanonicalOpenAiForwardProvider(provider) && process.env["OCX_ASTRA_EFFORT_CACHE"] === "1") { + if (isCanonicalOpenAiForwardProvider(provider) && supportsAstraEffortCache(finalBody)) { const effortResult = applyAstraEffortCache(finalBody, parsed._rawBody, incoming.headers, new Headers(headers)); finalBody = effortResult.body; if (effortResult.baseline && effortResult.effective) { diff --git a/tests/responses/astra-effort-cache.test.ts b/tests/responses/astra-effort-cache.test.ts index b027e3e776..c30796c4e7 100644 --- a/tests/responses/astra-effort-cache.test.ts +++ b/tests/responses/astra-effort-cache.test.ts @@ -14,17 +14,13 @@ import { withTestTranslatorBudget } from "../helpers/translator-budget"; let directory: string; let oldHome: string | undefined; -let oldFlag: string | undefined; beforeEach(() => { directory = mkdtempSync(join(tmpdir(), "astra-effort-")); oldHome = process.env["OPENCODEX_HOME"]; - oldFlag = process.env["OCX_ASTRA_EFFORT_CACHE"]; process.env["OPENCODEX_HOME"] = directory; - delete process.env["OCX_ASTRA_EFFORT_CACHE"]; }); afterEach(() => { if (oldHome === undefined) delete process.env["OPENCODEX_HOME"]; else process.env["OPENCODEX_HOME"] = oldHome; - if (oldFlag === undefined) delete process.env["OCX_ASTRA_EFFORT_CACHE"]; else process.env["OCX_ASTRA_EFFORT_CACHE"] = oldFlag; rmSync(directory, { recursive: true, force: true }); }); const user = (content: string) => ({ role: "user", content }); @@ -186,13 +182,20 @@ function adapterRequest(input: unknown[], effort = "medium", extra = {}, destina return adapter.buildRequest(parseRequest(body(input, effort, extra)), { headers: new Headers({ authorization: "Bearer synthetic", "chatgpt-account-id": "test-account", "thread-id": "thread-a" }) }); } describe("Astra adapter integration", () => { - test("default remains disabled and performs no state writes", () => { - adapterRequest(first); const request = adapterRequest(second, "low"); + test("default preserves the baseline and inserts effort updates", () => { + adapterRequest(first); + const request = adapterRequest(second, "low"); + expect(JSON.parse(request.body).reasoning.effort).toBe("medium"); + expect(JSON.parse(request.body).input.filter((i: any) => i.type === "configuration_update")).toEqual([update("low")]); + expect(request.reasoningLog?.effectiveEffort).toBe("low"); + }); + test("other native models do not activate effort state or diagnostics", () => { + const request = adapterRequest(first, "low", { model: "gpt-5.6-luna" }); expect(JSON.parse(request.body).reasoning.effort).toBe("low"); + expect(request.reasoningLog).toBeUndefined(); expect(readdirSync(directory)).toEqual([]); }); test("parser, native adapter, websocket framing and logs preserve the update", () => { - process.env["OCX_ASTRA_EFFORT_CACHE"] = "1"; adapterRequest(first); const request = adapterRequest(second, "low"); const output = JSON.parse(request.body); @@ -208,14 +211,12 @@ describe("Astra adapter integration", () => { expect(log).toMatchObject({ requestedEffort: "low", effectiveEffort: "low", reasoningWireValue: "medium" }); }); test("auto truncation is detected before native parameter stripping", () => { - process.env["OCX_ASTRA_EFFORT_CACHE"] = "1"; adapterRequest(first); const request = adapterRequest(second, "low", { truncation: "auto" }); expect(JSON.parse(request.body).reasoning.effort).toBe("low"); expect(JSON.parse(request.body).input.some((i: any) => i.type === "configuration_update")).toBe(false); }); test("custom forward destinations never receive generated updates", () => { - process.env["OCX_ASTRA_EFFORT_CACHE"] = "1"; const destination = { ...provider, baseUrl: "https://gateway.example.test" }; adapterRequest(first, "medium", {}, destination); expect(JSON.parse(adapterRequest(second, "low", {}, destination).body).reasoning.effort).toBe("low"); From 0d666d604e0bc655c869774d85a1f14dd5c54fc4 Mon Sep 17 00:00:00 2001 From: nahuelb Date: Fri, 11 Sep 2026 01:05:36 -0300 Subject: [PATCH 5/7] Measure Astra effort-cache overhead and exercise proxy transports --- .../docs/reference/configuration/server.md | 51 +++++++ scripts/astra-effort-cache-eval.ts | 140 ++++++++++++++++++ scripts/astra-effort-cache-report.ts | 51 +++++++ scripts/test-layout/layout.json | 4 +- src/adapters/astra-effort-cache.ts | 69 ++++++--- src/adapters/astra-effort-state.ts | 93 +++++++----- src/adapters/base.ts | 2 + src/adapters/openai-responses.ts | 6 +- src/server/request-log.ts | 18 ++- src/usage/astra-effort-cache.ts | 50 +++++++ src/usage/log.ts | 5 + tests/fixtures/test-layout-expected.json | 4 +- tests/helpers/astra-effort-proxy.ts | 117 +++++++++++++++ .../astra-effort-cache-proxy.test.ts | 58 ++++++++ tests/responses/astra-effort-cache.test.ts | 7 +- .../usage/astra-effort-cache-metrics.test.ts | 56 +++++++ 16 files changed, 671 insertions(+), 60 deletions(-) create mode 100644 scripts/astra-effort-cache-eval.ts create mode 100644 scripts/astra-effort-cache-report.ts create mode 100644 src/usage/astra-effort-cache.ts create mode 100644 tests/helpers/astra-effort-proxy.ts create mode 100644 tests/responses/astra-effort-cache-proxy.test.ts create mode 100644 tests/usage/astra-effort-cache-metrics.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 4c4dcf89e4..2717e776be 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -530,3 +530,54 @@ baseline. Do not share one thread identity across independent conversations. See OpenAI's [reasoning update compatibility](https://developers.openai.com/api/docs/guides/reasoning#change-reasoning-mid-conversation) and [prompt caching guidance](https://developers.openai.com/api/docs/guides/prompt-caching#change-reasoning-effort-without-rewriting-the-prefix). + + +### Measure Astra effort-cache overhead + +Eligible Astra requests include `astraEffortCache` in the existing local `usage.jsonl` log, +including per-attempt records. No separate telemetry service or Lab activation is required. +The fields contain fixed status codes, counters, and durations, never prompts or account identifiers. + +`durationMs` measures the synchronous cache hook, including ownership checks, database setup, +history processing, and close. `setupMs`, `transactionMs`, `historyMs`, and `closeMs` expose those +phases. **History time is inside transaction time**, so do not add all phase durations together. +Measurements describe the last adapter preparation in each attempt, not cumulative retry work. +`stateOutcome` distinguishes skipped, committed, busy, and error paths; committed means the +transaction completed, not that the upstream accepted the request or returned cached tokens. +`inputItems` and `updateCount` count input items and outgoing configuration updates. + +From a source checkout, summarize the newest 1,000 usage rows: + +```bash +bun scripts/astra-effort-cache-report.ts 1000 +``` + +Set `OPENCODEX_HOME` to inspect another installation. An optional second argument filters by exact +request ID within the bounded recent window. The report prints aggregate counts, phase p50/p95/p99, +and cached-token totals from **reported** usage. Estimated or missing usage stays unknown, and +invalid token counts are excluded. Attempts replace their mirrored request summary when present. +Older installations have no timing samples; an empty report is not proof of zero overhead. +The existing Logs interface continues to show request duration, first output, and token usage. + +For a reproducible local benchmark with synthetic data and no model API calls: + +```bash +bun scripts/astra-effort-cache-eval.ts .tmp/astra-effort-eval 40 4 /path/to/clean-dev-worktree +``` + +The final argument is optional. When supplied, it runs the same HTTP/WebSocket fixture against +that checkout as a control. Use the same upstream `dev` commit on which the feature branch is based, +and install that checkout's dependencies first. The control does not disable the feature in production. + +The harness writes `report.json` and raw `samples.jsonl`. It records platform, Bun version, source +commit, and dirty-patch digest. Separate processes exercise a shared store with 1 KiB, 64 KiB, and +1 MiB synthetic user text; each fresh store's first call includes database creation. Subsequent calls +cover new conversations, effort switches, and replay. Concurrent writers can intentionally fall back +when SQLite is busy. Real proxy cells use concurrent HTTP and WebSocket clients with both HTTP/SSE +and WebSocket upstream fixtures. Client latency includes the whole local request; timer delay measures +blocking in that fixture process. These are local costs, not production latency or upstream cache-hit +proof. Synthetic token counts must never be interpreted as observed model cache savings. + +Run several trials on the target operating system, including Windows, before drawing rollout +conclusions. Local fixtures test routing and wire preservation; they cannot establish upstream +acceptance or Codex Desktop behavior. Use actual reported usage during normal work for that evidence. diff --git a/scripts/astra-effort-cache-eval.ts b/scripts/astra-effort-cache-eval.ts new file mode 100644 index 0000000000..5eeef7f7f9 --- /dev/null +++ b/scripts/astra-effort-cache-eval.ts @@ -0,0 +1,140 @@ +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { createHash } from "node:crypto"; +import { distribution, summarizeAstraEffortCache } from "./astra-effort-cache-report"; + +const [mode, ...args] = process.argv.slice(2); +const user = (content: string) => ({ role: "user", content }); +const body = (input: unknown[], effort = "medium") => ({ model: "gpt-6-astra", instructions: "Synthetic fixture", input, reasoning: { effort }, store: false, stream: true }); + +if (mode === "--worker") { + const [home, worker, countArg, bytesArg] = args; + process.env.OPENCODEX_HOME = join(home, "config"); + const { applyAstraEffortCache } = await import("../src/adapters/astra-effort-cache"); + const count = Number(countArg); + const first = [user("x".repeat(Number(bytesArg)))]; + const second = [...first, { type: "message", role: "assistant", content: [{ type: "output_text", text: "OK" }] }, user("next")]; + if (worker === "0") { + const { recordOwnedConfigPath } = await import("../src/lib/config-ownership"); + if (!recordOwnedConfigPath(process.env.OPENCODEX_HOME!, join(process.env.OPENCODEX_HOME!, "astra-effort-cache"))) throw new Error("Cannot own synthetic cache directory"); + } + writeFileSync(join(home, `ready-${worker}`), ""); + const deadline = Date.now() + 15_000; + while (!existsSync(join(home, "go"))) { + if (Date.now() > deadline) throw new Error("Synthetic worker barrier timeout"); + await Bun.sleep(5); + } + const samples = []; + for (let i = 0; i < count; i++) { + const scenario = i % 4; + const request = body(scenario === 0 ? first : second, scenario === 0 ? "medium" : "low"); + const headers = new Headers({ "thread-id": `worker-${worker}-conversation-${Math.floor(i / 4)}` }); + const started = performance.now(); + const result = applyAstraEffortCache(request, request, headers, new Headers({ "chatgpt-account-id": "synthetic-account" })); + samples.push({ worker: Number(worker), scenario: ["new-conversation", "switch", "replay", "replay"][scenario], wallMs: performance.now() - started, ...result.metrics }); + } + console.log(JSON.stringify(samples)); +} else if (mode === "--proxy") { + const [native, countArg, concurrencyArg, bytesArg, runtimeRoot] = args; + const { startAstraEffortProxy } = await import("../tests/helpers/astra-effort-proxy"); + const { readRecentUsageEntries } = await import("../src/usage/log"); + const fixture = await startAstraEffortProxy(native === "true", runtimeRoot); + const samples: Array<{ transport: string; wallMs: number }> = []; + const count = Number(countArg); + const concurrency = Number(concurrencyArg); + const started = performance.now(); + let maxTimerDelayMs = 0; + let tick = performance.now(); + const timer = setInterval(() => { const now = performance.now(); maxTimerDelayMs = Math.max(maxTimerDelayMs, now - tick - 5); tick = now; }, 5); + try { + for (const transport of ["http", "websocket"]) { + await Promise.all(Array.from({ length: concurrency }, async (_, worker) => { + const thread = `${transport}-${worker}`; + const ws = transport === "websocket" ? fixture.websocket(thread) : undefined; + let input: unknown[] = []; + try { + for (let i = 0; i < count; i++) { + input.push(user(i === 0 ? "x".repeat(Number(bytesArg)) : "next")); + const request = body(input, i % 2 ? "low" : "medium"); + const start = performance.now(); + const response = ws ? await ws.turn(request) : await fixture.http(request, thread); + samples.push({ transport, wallMs: performance.now() - start }); + input = [...input, ...response.output]; + } + } finally { ws?.close(); } + })); + } + await Bun.sleep(10); + const elapsedMs = performance.now() - started; + const measurements = summarizeAstraEffortCache(readRecentUsageEntries(10_000, fixture.home)); + console.log(JSON.stringify({ samples, elapsedMs, requestsPerSecond: samples.length / elapsedMs * 1000, maxTimerDelayMs, measurements })); + } finally { clearInterval(timer); await fixture.stop(); } +} else { + const outDir = mode; + const [countArg = "40", concurrencyArg = "4", controlRoot] = args; + const count = Number(countArg), concurrency = Number(concurrencyArg); + if (!outDir || !Number.isSafeInteger(count) || count < 4 || count > 200 || !Number.isSafeInteger(concurrency) || concurrency < 1 || concurrency > 16) { + throw new Error("Usage: bun scripts/astra-effort-cache-eval.ts [4..200 turns] [1..16 workers] [control worktree]"); + } + mkdirSync(outDir, { recursive: true, mode: 0o700 }); + const scratch = mkdtempSync(join(tmpdir(), "astra-eval-")); + mkdirSync(join(scratch, "codex")); + const children: ReturnType[] = []; + const root = resolve(import.meta.dir, ".."); + const env = { ...process.env, HOME: scratch, USERPROFILE: scratch, OPENCODEX_HOME: scratch, CODEX_HOME: join(scratch, "codex") }; + for (const key of Object.keys(env)) if (/^(http|https|all)_proxy$/i.test(key)) delete (env as Record)[key]; + function spawn(childArgs: string[]) { + const child = Bun.spawn([process.execPath, import.meta.path, ...childArgs], { env, stdout: "pipe", stderr: "pipe" }); + children.push(child); + return child; + } + async function result(child: ReturnType) { + const [text, error, code] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited]); + if (code !== 0) throw new Error(`Synthetic benchmark child failed (${code}): ${error.slice(-1000)}`); + const json = text.trim().split("\n").at(-1)!; + return JSON.parse(json); + } + function revision(path: string) { + const head = Bun.spawnSync(["git", "rev-parse", "HEAD"], { cwd: path }); + const diff = Bun.spawnSync(["git", "diff", "HEAD"], { cwd: path }); + const untracked = Bun.spawnSync(["git", "ls-files", "--others", "--exclude-standard", "-z", "--", ".", ":!node_modules"], { cwd: path }); + if (head.exitCode || diff.exitCode || untracked.exitCode) throw new Error("Cannot identify benchmark checkout"); + const hash = createHash("sha256").update(diff.stdout); + const files = untracked.stdout.toString().split("\0").filter(Boolean).sort(); + for (const file of files) hash.update(file).update(readFileSync(join(path, file))); + return { commit: head.stdout.toString().trim(), dirty: diff.stdout.length > 0 || files.length > 0, patchSha256: hash.digest("hex") }; + } + const cells = []; + try { + for (const bytes of [1024, 64 * 1024, 1024 * 1024]) { + for (const workers of [...new Set([1, concurrency])]) { + const home = join(scratch, `direct-${bytes}-${workers}`); mkdirSync(home); + const tasks = Array.from({ length: workers }, (_, i) => spawn(["--worker", home, String(i), String(count), String(bytes)])); + const deadline = Date.now() + 15_000; + while (!tasks.every((_, i) => existsSync(join(home, `ready-${i}`)))) { + for (const task of tasks) if (task.exitCode !== null) await result(task); + if (Date.now() > deadline) throw new Error("Synthetic workers failed to reach barrier"); + await Bun.sleep(5); + } + writeFileSync(join(home, "go"), ""); + const samples = (await Promise.all(tasks.map(result))).flat(); + cells.push({ kind: "synchronous-hook", inputTextBytes: bytes, workers, samples, wallMs: distribution(samples.map(row => row.wallMs)), statuses: samples.reduce((counts, row) => { counts[row.status] = (counts[row.status] ?? 0) + 1; return counts; }, {} as Record) }); + } + } + for (const native of [false, true]) { + for (const [arm, runtime] of [["treatment", undefined], ...(controlRoot ? [["control", resolve(controlRoot)]] : [])] as const) { + cells.push({ kind: "proxy", arm, nativeUpstreamWebSocket: native, inputTextBytes: 64 * 1024, workers: concurrency, + ...await result(spawn(["--proxy", String(native), String(count), String(concurrency), String(64 * 1024), ...(runtime ? [runtime] : [])])) }); + } + } + const report = { schemaVersion: 1, synthetic: true, platform: process.platform, arch: process.arch, bun: Bun.version, treatment: revision(root), ...(controlRoot ? { control: revision(resolve(controlRoot)) } : {}), cells }; + writeFileSync(join(outDir, "samples.jsonl"), cells.flatMap((cell, i) => cell.samples.map((sample: unknown) => JSON.stringify({ cell: i, sample }))).join("\n") + "\n"); + writeFileSync(join(outDir, "report.json"), JSON.stringify({ ...report, cells: cells.map(({ samples, ...cell }) => ({ ...cell, sampleCount: samples.length, ...(cell.kind === "proxy" ? { wallMs: distribution(samples.map((row: any) => row.wallMs)) } : {}) })) }, null, 2) + "\n"); + console.log("Synthetic benchmark complete: report.json and samples.jsonl"); + } finally { + for (const child of children) if (child.exitCode === null) child.kill(); + await Promise.all(children.map(child => child.exited)); + rmSync(scratch, { recursive: true, force: true }); + } +} diff --git a/scripts/astra-effort-cache-report.ts b/scripts/astra-effort-cache-report.ts new file mode 100644 index 0000000000..6b6d252c7a --- /dev/null +++ b/scripts/astra-effort-cache-report.ts @@ -0,0 +1,51 @@ +import { readRecentUsageEntries, type PersistedUsageEntry } from "../src/usage/log"; +import { normalizeAstraEffortCacheMetrics } from "../src/usage/astra-effort-cache"; + +export function distribution(values: number[]) { + const sorted = [...values].sort((a, b) => a - b); + const percentile = (p: number) => sorted.length ? sorted[Math.ceil(p * sorted.length) - 1] : null; + return { samples: sorted.length, p50: percentile(0.5), p95: percentile(0.95), p99: percentile(0.99), max: sorted.at(-1) ?? null }; +} + +export function summarizeAstraEffortCache(entries: PersistedUsageEntry[]) { + const statuses: Record = {}; + const stateOutcomes: Record = {}; + const timings = Object.fromEntries(["durationMs", "setupMs", "transactionMs", "historyMs", "closeMs"].map(key => [key, [] as number[]])); + const cache = { hit: 0, miss: 0, unknown: 0, invalid: 0, inputTokens: 0, cachedInputTokens: 0 }; + let samples = 0; + for (const entry of entries) { + for (const row of entry.attempts?.length ? entry.attempts : [entry]) { + const metrics = normalizeAstraEffortCacheMetrics(row.astraEffortCache); + if (!metrics) continue; + samples++; + statuses[metrics.status] = (statuses[metrics.status] ?? 0) + 1; + stateOutcomes[metrics.stateOutcome] = (stateOutcomes[metrics.stateOutcome] ?? 0) + 1; + for (const key of Object.keys(timings) as (keyof typeof metrics)[]) { + const value = metrics[key]; + if (typeof value === "number") timings[key].push(value); + } + const input = row.usage?.inputTokens; + const cached = row.usage?.cachedInputTokens; + if (row.usageStatus !== "reported" || input === undefined || cached === undefined) cache.unknown++; + else if (!Number.isSafeInteger(input) || !Number.isSafeInteger(cached) || input < 0 || cached < 0 || cached > input) cache.invalid++; + else { + cache[cached > 0 ? "hit" : "miss"]++; + cache.inputTokens += input; + cache.cachedInputTokens += cached; + } + } + } + return { + rowsRead: entries.length, samples, statuses, stateOutcomes, + timingsMs: Object.fromEntries(Object.entries(timings).map(([key, values]) => [key, distribution(values)])), + cache: { ...cache, cachedInputRatio: cache.inputTokens > 0 ? cache.cachedInputTokens / cache.inputTokens : null }, + }; +} + +if (import.meta.main) { + const [limitArg = "1000", requestId] = process.argv.slice(2); + const limit = Number(limitArg); + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 10_000) throw new Error("Usage: bun scripts/astra-effort-cache-report.ts [1..10000 recent rows] [request-id]"); + const entries = readRecentUsageEntries(limit); + console.log(JSON.stringify(summarizeAstraEffortCache(requestId ? entries.filter(row => row.requestId === requestId) : entries), null, 2)); +} diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index e088020334..f64110ee68 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1332,7 +1332,9 @@ "zhipu-bigmodel-responses-quota.test.ts": "providers", "zz-ci-api-usage-isolation.test.ts": "ci-workflows", "zz-ci-storage-policy-isolation.test.ts": "ci-workflows", - "zz-pr-coderabbit-readiness-revalidation.test.ts": "ci-workflows" + "zz-pr-coderabbit-readiness-revalidation.test.ts": "ci-workflows", + "astra-effort-cache-metrics.test.ts": "usage", + "astra-effort-cache-proxy.test.ts": "responses" }, "migrated": [ "adapters", diff --git a/src/adapters/astra-effort-cache.ts b/src/adapters/astra-effort-cache.ts index 3fa9d250ff..b2d6e8aba9 100644 --- a/src/adapters/astra-effort-cache.ts +++ b/src/adapters/astra-effort-cache.ts @@ -1,3 +1,4 @@ +import { normalizeAstraEffortCacheMetrics, type AstraEffortCacheMetrics, type AstraEffortStoreMeasurement } from "../usage/astra-effort-cache"; import { createHash } from "node:crypto"; import { join } from "node:path"; import { getConfigDir } from "../config/paths"; @@ -20,6 +21,7 @@ export interface AstraEffortResult { status: string; baseline?: string; effective?: string; + metrics?: AstraEffortCacheMetrics; } function record(value: unknown): value is RecordValue { @@ -113,12 +115,13 @@ function transform(body: RecordValue, state: State): AstraEffortResult & { snaps status, baseline, effective: requested, snapshot }; } -export function applyAstraEffortCache( +function applyAstraEffortCacheInner( body: unknown, original: unknown, headers: Headers, servingHeaders: Headers, - directory = join(getConfigDir(), "astra-effort-cache"), + directory: string, + measurement: AstraEffortStoreMeasurement & { historyMs?: number }, ): AstraEffortResult { if (!record(body)) return { body, status: "unsupported_input" }; const requested = record(body.reasoning) && effort(body.reasoning.effort) ? body.reasoning.effort : undefined; @@ -137,25 +140,51 @@ export function applyAstraEffortCache( const scope = digest(JSON.stringify([thread, account])); try { return withAstraEffortState(directory, scope, serialized => { - let state: State = { version: 1, snapshots: [] }; - if (serialized !== undefined) { - if (serialized.length > MAX_STATE_BYTES) return { value: fallback("state_limit") }; - const loaded: unknown = JSON.parse(serialized); - if (!validState(loaded)) return { value: fallback("invalid_state") }; - state = loaded; - } - const result = transform(body, state); - if (!result.snapshot) return { value: result, state: null }; - const snapshot = result.snapshot; - if (!state.snapshots.some(s => JSON.stringify(s) === JSON.stringify(snapshot))) { - if (state.snapshots.length >= MAX_SNAPSHOTS) return { value: fallback("state_limit") }; - state.snapshots.push(snapshot); - } - const next = JSON.stringify(state); - if (next.length > MAX_STATE_BYTES) return { value: fallback("state_limit") }; - return { value: { body: result.body, status: result.status, baseline: result.baseline, effective: result.effective }, state: next }; - }); + const started = performance.now(); + try { + let state: State = { version: 1, snapshots: [] }; + if (serialized !== undefined) { + if (serialized.length > MAX_STATE_BYTES) return { value: fallback("state_limit") }; + const loaded: unknown = JSON.parse(serialized); + if (!validState(loaded)) return { value: fallback("invalid_state") }; + state = loaded; + } + const result = transform(body, state); + if (!result.snapshot) return { value: result, state: null }; + const snapshot = result.snapshot; + if (!state.snapshots.some(s => JSON.stringify(s) === JSON.stringify(snapshot))) { + if (state.snapshots.length >= MAX_SNAPSHOTS) return { value: fallback("state_limit") }; + state.snapshots.push(snapshot); + } + const next = JSON.stringify(state); + if (next.length > MAX_STATE_BYTES) return { value: fallback("state_limit") }; + return { value: { body: result.body, status: result.status, baseline: result.baseline, effective: result.effective }, state: next }; + } finally { measurement.historyMs = performance.now() - started; } + }, measurement); } catch { return fallback("unavailable_state"); } } + +export function applyAstraEffortCache( + body: unknown, + original: unknown, + headers: Headers, + servingHeaders: Headers, + directory = join(getConfigDir(), "astra-effort-cache"), +): AstraEffortResult { + const started = performance.now(); + const measurement: AstraEffortStoreMeasurement & { historyMs?: number } = { outcome: "skipped" }; + const result = applyAstraEffortCacheInner(body, original, headers, servingHeaders, directory, measurement); + const input = record(body) && Array.isArray(body.input) ? body.input : []; + const output = record(result.body) && Array.isArray(result.body.input) ? result.body.input : []; + let updateCount = 0; + for (const item of output) if (record(item) && item.type === "configuration_update") updateCount++; + const metrics = normalizeAstraEffortCacheMetrics({ + status: result.status, stateOutcome: measurement.outcome, durationMs: performance.now() - started, + setupMs: measurement.setupMs, transactionMs: measurement.transactionMs, historyMs: measurement.historyMs, + closeMs: measurement.closeMs, inputItems: input.length, + updateCount, + }); + return { ...result, ...(metrics ? { metrics } : {}) }; +} diff --git a/src/adapters/astra-effort-state.ts b/src/adapters/astra-effort-state.ts index 167f8eff40..cc73429161 100644 --- a/src/adapters/astra-effort-state.ts +++ b/src/adapters/astra-effort-state.ts @@ -1,3 +1,4 @@ +import type { AstraEffortStoreMeasurement } from "../usage/astra-effort-cache"; import { Database } from "bun:sqlite"; import { chmodSync, closeSync, lstatSync, mkdirSync, openSync } from "node:fs"; import { join } from "node:path"; @@ -15,45 +16,71 @@ export function withAstraEffortState( directory: string, scope: string, work: (state: string | undefined) => { value: T; state?: string | null }, + measurement?: AstraEffortStoreMeasurement, ): T { - assertNotRealHomeUnderTest(getConfigDir()); - assertNotRealHomeUnderTest(directory); - if (!recordOwnedConfigPath(getConfigDir(), directory)) throw new Error("Unowned effort state directory"); - mkdirSync(directory, { recursive: true, mode: 0o700 }); - if (!lstatSync(directory).isDirectory() || lstatSync(directory).isSymbolicLink()) throw new Error("Invalid effort state directory"); - if (process.platform !== "win32") chmodSync(directory, 0o700); - hardenSecretDir(directory, { required: true }); - const path = join(directory, "state.sqlite"); - try { closeSync(openSync(path, "wx", 0o600)); } - catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; } - const file = lstatSync(path); - if (!file.isFile() || file.isSymbolicLink() || file.nlink !== 1 || file.size > MAX_DATABASE_BYTES) throw new Error("Invalid effort state file"); - if (process.platform !== "win32") chmodSync(path, 0o600); - hardenSecretPath(path, { required: true }); - const db = new Database(path, { readwrite: true }); + const started = performance.now(); + let db: Database | undefined; + if (measurement) measurement.outcome = "error"; try { + assertNotRealHomeUnderTest(getConfigDir()); + assertNotRealHomeUnderTest(directory); + if (!recordOwnedConfigPath(getConfigDir(), directory)) throw new Error("Unowned effort state directory"); + mkdirSync(directory, { recursive: true, mode: 0o700 }); + if (!lstatSync(directory).isDirectory() || lstatSync(directory).isSymbolicLink()) throw new Error("Invalid effort state directory"); + if (process.platform !== "win32") chmodSync(directory, 0o700); + hardenSecretDir(directory, { required: true }); + const path = join(directory, "state.sqlite"); + try { closeSync(openSync(path, "wx", 0o600)); } + catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; } + const file = lstatSync(path); + if (!file.isFile() || file.isSymbolicLink() || file.nlink !== 1 || file.size > MAX_DATABASE_BYTES) throw new Error("Invalid effort state file"); + if (process.platform !== "win32") chmodSync(path, 0o600); + hardenSecretPath(path, { required: true }); + db = new Database(path, { readwrite: true }); + const database = db; db.exec("PRAGMA busy_timeout = 0; PRAGMA journal_mode = DELETE; PRAGMA page_size = 4096; PRAGMA max_page_count = 8192"); - const pageSize = (db.query("PRAGMA page_size").get() as { page_size: number }).page_size; + const pageSize = (database.query("PRAGMA page_size").get() as { page_size: number }).page_size; db.exec(`PRAGMA max_page_count = ${Math.floor(MAX_DATABASE_BYTES / pageSize)}`); db.exec("CREATE TABLE IF NOT EXISTS sessions (scope TEXT PRIMARY KEY, state TEXT NOT NULL CHECK(length(CAST(state AS BLOB)) <= 2097152), touched INTEGER NOT NULL)"); - return db.transaction(() => { - const now = Date.now(); - db.query("DELETE FROM sessions WHERE touched < ?").run(now - RETENTION_MS); - const row = db.query("SELECT state FROM sessions WHERE scope = ?").get(scope) as { state: string } | null; - const result = work(row?.state); - if (result.state === null) db.query("DELETE FROM sessions WHERE scope = ?").run(scope); - else if (result.state !== undefined) { - db.query("INSERT INTO sessions VALUES (?, ?, ?) ON CONFLICT(scope) DO UPDATE SET state = excluded.state, touched = excluded.touched") - .run(scope, result.state, now); - for (;;) { - const totals = db.query("SELECT COUNT(*) AS count, COALESCE(SUM(length(CAST(state AS BLOB))), 0) AS bytes FROM sessions").get() as { count: number; bytes: number }; - if (totals.count <= MAX_CONVERSATIONS && totals.bytes <= MAX_PAYLOAD_BYTES) break; - db.query("DELETE FROM sessions WHERE scope = (SELECT scope FROM sessions WHERE scope != ? ORDER BY touched, scope LIMIT 1)").run(scope); + if (measurement) measurement.setupMs = performance.now() - started; + const transactionStarted = performance.now(); + let result: T; + try { + result = database.transaction(() => { + const now = Date.now(); + database.query("DELETE FROM sessions WHERE touched < ?").run(now - RETENTION_MS); + const row = database.query("SELECT state FROM sessions WHERE scope = ?").get(scope) as { state: string } | null; + const result = work(row?.state); + if (result.state === null) database.query("DELETE FROM sessions WHERE scope = ?").run(scope); + else if (result.state !== undefined) { + database.query("INSERT INTO sessions VALUES (?, ?, ?) ON CONFLICT(scope) DO UPDATE SET state = excluded.state, touched = excluded.touched") + .run(scope, result.state, now); + for (;;) { + const totals = database.query("SELECT COUNT(*) AS count, COALESCE(SUM(length(CAST(state AS BLOB))), 0) AS bytes FROM sessions").get() as { count: number; bytes: number }; + if (totals.count <= MAX_CONVERSATIONS && totals.bytes <= MAX_PAYLOAD_BYTES) break; + database.query("DELETE FROM sessions WHERE scope = (SELECT scope FROM sessions WHERE scope != ? ORDER BY touched, scope LIMIT 1)").run(scope); + } } - } - return result.value; - }).immediate(); + return result.value; + }).immediate(); + } finally { + if (measurement) measurement.transactionMs = performance.now() - transactionStarted; + } + if (measurement) measurement.outcome = "committed"; + return result; + } catch (error) { + if (measurement) { + measurement.setupMs ??= performance.now() - started; + const code = (error as { code?: unknown } | null)?.code; + measurement.outcome = code === "SQLITE_BUSY" || code === "SQLITE_LOCKED" ? "busy" : "error"; + } + throw error; } finally { - db.close(); + if (db) { + const closeStarted = performance.now(); + try { db.close(); } + catch (error) { if (measurement) measurement.outcome = "error"; throw error; } + finally { if (measurement) measurement.closeMs = performance.now() - closeStarted; } + } } } diff --git a/src/adapters/base.ts b/src/adapters/base.ts index f5a22b2969..73487bbce4 100644 --- a/src/adapters/base.ts +++ b/src/adapters/base.ts @@ -1,3 +1,4 @@ +import type { AstraEffortCacheMetrics } from "../usage/astra-effort-cache"; import type { AdapterEvent, OcxParsedRequest } from "../types"; import type { TranslatorBudget } from "../lib/translator-budget"; import type { AdapterTierMetadata } from "../providers/fastwire"; @@ -79,6 +80,7 @@ export interface ProviderAdapter { } export interface AdapterRequest { + astraEffortCache?: AstraEffortCacheMetrics; url: string; method: string; headers: Record; diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index 8cb5e53178..8bd7139386 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -1,3 +1,4 @@ +import type { AstraEffortCacheMetrics } from "../usage/astra-effort-cache"; import { applyAstraEffortCache, supportsAstraEffortCache } from "./astra-effort-cache"; import { normalizeRoutedAgentMessages } from "./routed-agent-messages"; import { normalizeOpenCodeGoAdditionalTools } from "./opencode-go-additional-tools"; @@ -2515,15 +2516,17 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): provider, parsed.modelId, ); + let astraEffortCache: AstraEffortCacheMetrics | undefined; let astraReasoningLog: { effectiveEffort: string; wireField: "reasoning.effort"; wireValue: string } | undefined; if (isCanonicalOpenAiForwardProvider(provider) && supportsAstraEffortCache(finalBody)) { const effortResult = applyAstraEffortCache(finalBody, parsed._rawBody, incoming.headers, new Headers(headers)); finalBody = effortResult.body; + astraEffortCache = effortResult.metrics; if (effortResult.baseline && effortResult.effective) { astraReasoningLog = { effectiveEffort: effortResult.effective, wireField: "reasoning.effort", wireValue: effortResult.baseline }; } debugProviderDiagnostic("openai-responses", "astra-effort-cache", { status: effortResult.status, - baseline: effortResult.baseline, effective: effortResult.effective }); + baseline: effortResult.baseline, effective: effortResult.effective, metrics: astraEffortCache }); } if (isCanonicalOpenAiForwardProvider(provider)) { // Spark closes Responses Lite streams before a terminal completion. Select compatibility @@ -2569,6 +2572,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): ...(convertedRoutedNamespaceToolAliases ? { convertedRoutedNamespaceToolAliases } : {}), ...(tierLog ? { tierLog } : {}), ...(astraReasoningLog ? { reasoningLog: astraReasoningLog } : {}), + ...(astraEffortCache ? { astraEffortCache } : {}), }; }, diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 6c92aad3ae..209c741e94 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -1,3 +1,4 @@ +import { normalizeAstraEffortCacheMetrics, type AstraEffortCacheMetrics } from "../usage/astra-effort-cache"; import { existsSync, readFileSync } from "node:fs"; import { randomBytes } from "node:crypto"; import type { ResponsesTerminalStatus } from "../bridge"; @@ -90,6 +91,7 @@ export interface RequestLogContext { effectiveEffort?: string; reasoningWireField?: string; reasoningWireValue?: string | number | boolean; + astraEffortCache?: AstraEffortCacheMetrics; callerServiceTier?: string; requestedServiceTier?: string; requestedSpeedLabel?: string; @@ -182,6 +184,7 @@ export interface RequestLogEntry { effectiveEffort?: string; reasoningWireField?: string; reasoningWireValue?: string | number | boolean; + astraEffortCache?: AstraEffortCacheMetrics; callerServiceTier?: string; requestedServiceTier?: string; requestedSpeedLabel?: string; @@ -301,6 +304,7 @@ export function requestLogEntryFromPersistedUsage(entry: PersistedUsageEntry): R ...(entry.effectiveEffort ? { effectiveEffort: entry.effectiveEffort } : {}), ...(entry.reasoningWireField ? { reasoningWireField: entry.reasoningWireField } : {}), ...(entry.reasoningWireValue !== undefined ? { reasoningWireValue: entry.reasoningWireValue } : {}), + ...(normalizeAstraEffortCacheMetrics(entry.astraEffortCache) ? { astraEffortCache: normalizeAstraEffortCacheMetrics(entry.astraEffortCache) } : {}), ...(entry.callerServiceTier ? { callerServiceTier: entry.callerServiceTier } : {}), ...(entry.requestedServiceTier ? { requestedServiceTier: entry.requestedServiceTier } : {}), ...(entry.requestedSpeedLabel ? { requestedSpeedLabel: entry.requestedSpeedLabel } : {}), @@ -382,10 +386,13 @@ export function addRequestLog(entry: RequestLogEntry) { // sanitization bug because the safe surface is the one you check. const shadowCallRewrittenFrom = sanitizeLogMetadataString(entry.shadowCallRewrittenFrom); const claudeCompatibility = normalizeClaudeCompatibilityUsageLog(entry.claudeCompatibility); - const retained: RequestLogEntry = shadowCallRewrittenFrom === entry.shadowCallRewrittenFrom && entry.claudeCompatibility === undefined + const retained: RequestLogEntry = shadowCallRewrittenFrom === entry.shadowCallRewrittenFrom && entry.claudeCompatibility === undefined && entry.astraEffortCache === undefined ? entry : { ...entry, ...(shadowCallRewrittenFrom ? { shadowCallRewrittenFrom } : {}) }; if (!shadowCallRewrittenFrom && retained !== entry) delete retained.shadowCallRewrittenFrom; + const astraMetrics = normalizeAstraEffortCacheMetrics(entry.astraEffortCache); + if (astraMetrics) retained.astraEffortCache = astraMetrics; + else if (retained !== entry) delete retained.astraEffortCache; if (claudeCompatibility) retained.claudeCompatibility = claudeCompatibility; else if (retained !== entry) delete retained.claudeCompatibility; entry = retained; @@ -428,6 +435,7 @@ export function addRequestLog(entry: RequestLogEntry) { ...(entry.effectiveEffort ? { effectiveEffort: entry.effectiveEffort } : {}), ...(entry.reasoningWireField ? { reasoningWireField: entry.reasoningWireField } : {}), ...(entry.reasoningWireValue !== undefined ? { reasoningWireValue: entry.reasoningWireValue } : {}), + ...(normalizeAstraEffortCacheMetrics(entry.astraEffortCache) ? { astraEffortCache: normalizeAstraEffortCacheMetrics(entry.astraEffortCache) } : {}), ...(entry.callerServiceTier ? { callerServiceTier: entry.callerServiceTier } : {}), ...(entry.requestedServiceTier ? { requestedServiceTier: entry.requestedServiceTier } : {}), ...(entry.requestedSpeedLabel ? { requestedSpeedLabel: entry.requestedSpeedLabel } : {}), @@ -508,12 +516,19 @@ export function recordAdapterReasoning( delete attempt.reasoningWireField; delete attempt.reasoningWireValue; } + delete logCtx.astraEffortCache; + if (attempt) delete attempt.astraEffortCache; recordAttemptRequestedEffort(logCtx); // Diagnostics must never make an otherwise valid upstream request fail. Config files // written by older versions (or edited by hand) can contain values that violate the // current TypeScript shape, so validate the runtime object before redacting strings. try { + const astraMetrics = normalizeAstraEffortCacheMetrics(request.astraEffortCache); + if (astraMetrics) { + logCtx.astraEffortCache = astraMetrics; + if (attempt) attempt.astraEffortCache = astraMetrics; + } const raw: unknown = request.reasoningLog; if (!raw || typeof raw !== "object" || Array.isArray(raw)) return; const reasoning = raw as Record; @@ -1056,6 +1071,7 @@ export function addFinalRequestLog( ...(logCtx.effectiveEffort ? { effectiveEffort: logCtx.effectiveEffort } : {}), ...(logCtx.reasoningWireField ? { reasoningWireField: logCtx.reasoningWireField } : {}), ...(logCtx.reasoningWireValue !== undefined ? { reasoningWireValue: logCtx.reasoningWireValue } : {}), + ...(normalizeAstraEffortCacheMetrics(logCtx.astraEffortCache) ? { astraEffortCache: normalizeAstraEffortCacheMetrics(logCtx.astraEffortCache) } : {}), ...(logCtx.callerServiceTier ? { callerServiceTier: logCtx.callerServiceTier } : {}), ...(logCtx.requestedServiceTier ? { requestedServiceTier: logCtx.requestedServiceTier } : {}), ...(logCtx.requestedSpeedLabel ? { requestedSpeedLabel: logCtx.requestedSpeedLabel } : {}), diff --git a/src/usage/astra-effort-cache.ts b/src/usage/astra-effort-cache.ts new file mode 100644 index 0000000000..d3798245b3 --- /dev/null +++ b/src/usage/astra-effort-cache.ts @@ -0,0 +1,50 @@ +export const ASTRA_EFFORT_CACHE_STATUSES = [ + "unsupported_input", "client_managed", "unsupported_model", "unsupported_effort", "unsupported_mode", + "multi_agent", "automatic_context_management", "compaction", "missing_thread_identity", "missing_serving_identity", + "state_limit", "invalid_state", "unavailable_state", "ambiguous_history", "conflicting_retry", + "missing_user_boundary", "baseline_reset", "replay", "updated", +] as const; + +export interface AstraEffortCacheMetrics { + status: typeof ASTRA_EFFORT_CACHE_STATUSES[number]; + stateOutcome: "skipped" | "committed" | "busy" | "error"; + durationMs: number; + setupMs?: number; + transactionMs?: number; + historyMs?: number; + closeMs?: number; + inputItems: number; + updateCount: number; +} + +export interface AstraEffortStoreMeasurement { + outcome: AstraEffortCacheMetrics["stateOutcome"]; + setupMs?: number; + transactionMs?: number; + closeMs?: number; +} + +function normalizeMetrics(value: unknown): AstraEffortCacheMetrics | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const row = value as Record; + const duration = (n: unknown): n is number => typeof n === "number" && Number.isFinite(n) && n >= 0 && n <= 3_600_000; + const count = (n: unknown): n is number => typeof n === "number" && Number.isSafeInteger(n) && n >= 0; + if (!ASTRA_EFFORT_CACHE_STATUSES.includes(row.status as AstraEffortCacheMetrics["status"]) + || typeof row.stateOutcome !== "string" || !["skipped", "committed", "busy", "error"].includes(row.stateOutcome) + || !duration(row.durationMs) || !count(row.inputItems) || !count(row.updateCount)) return undefined; + const metrics: AstraEffortCacheMetrics = { + status: row.status as AstraEffortCacheMetrics["status"], stateOutcome: row.stateOutcome as AstraEffortCacheMetrics["stateOutcome"], + durationMs: row.durationMs, inputItems: row.inputItems, updateCount: row.updateCount, + }; + for (const key of ["setupMs", "transactionMs", "historyMs", "closeMs"] as const) { + if (row[key] !== undefined) { + if (!duration(row[key])) return undefined; + metrics[key] = row[key]; + } + } + return metrics; +} + +export function normalizeAstraEffortCacheMetrics(value: unknown): AstraEffortCacheMetrics | undefined { + try { return normalizeMetrics(value); } catch { return undefined; } +} diff --git a/src/usage/log.ts b/src/usage/log.ts index 2944c22f9a..f595de5e05 100644 --- a/src/usage/log.ts +++ b/src/usage/log.ts @@ -1,3 +1,4 @@ +import { normalizeAstraEffortCacheMetrics, type AstraEffortCacheMetrics } from "./astra-effort-cache"; import { createHash, type Hash } from "node:crypto"; import { chmodSync, closeSync, existsSync, fstatSync, mkdirSync, openSync, readFileSync, readSync, appendFileSync } from "node:fs"; import { join } from "node:path"; @@ -116,6 +117,7 @@ export interface PersistedUsageAttempt { effectiveEffort?: string; reasoningWireField?: string; reasoningWireValue?: string | number | boolean; + astraEffortCache?: AstraEffortCacheMetrics; /** Adapter-produced tier fact for this physical attempt; absent on pre-B0 rows. */ tierOutcome?: AttemptTierOutcome; } @@ -147,6 +149,7 @@ export interface PersistedUsageEntry { effectiveEffort?: string; reasoningWireField?: string; reasoningWireValue?: string | number | boolean; + astraEffortCache?: AstraEffortCacheMetrics; /** Raw caller tier captured before routing, sanitized and bounded for durable logs. */ callerServiceTier?: string; requestedServiceTier?: string; @@ -484,6 +487,7 @@ function normalizeUsageAttempt(raw: unknown): PersistedUsageAttempt | null { ...(typeof attempt.reasoningWireField === "string" && attempt.reasoningWireField ? { reasoningWireField: capMetadataString(attempt.reasoningWireField) } : {}), + ...(normalizeAstraEffortCacheMetrics(attempt.astraEffortCache) ? { astraEffortCache: normalizeAstraEffortCacheMetrics(attempt.astraEffortCache) } : {}), ...(isValidReasoningWireValue(attempt.reasoningWireField, attempt.reasoningWireValue) ? typeof attempt.reasoningWireValue === "string" ? { reasoningWireValue: capMetadataString(attempt.reasoningWireValue) } @@ -569,6 +573,7 @@ function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry { ...(typeof entry.reasoningWireField === "string" && entry.reasoningWireField ? { reasoningWireField: capMetadataString(entry.reasoningWireField) } : {}), + ...(normalizeAstraEffortCacheMetrics(entry.astraEffortCache) ? { astraEffortCache: normalizeAstraEffortCacheMetrics(entry.astraEffortCache) } : {}), ...(isValidReasoningWireValue(entry.reasoningWireField, entry.reasoningWireValue) ? typeof entry.reasoningWireValue === "string" ? { reasoningWireValue: capMetadataString(entry.reasoningWireValue) } diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d47a939822..1ca22cd810 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1167,5 +1167,7 @@ "zhipu-bigmodel-responses-quota.test.ts": "providers", "zz-ci-api-usage-isolation.test.ts": "ci-workflows", "zz-ci-storage-policy-isolation.test.ts": "ci-workflows", - "zz-pr-coderabbit-readiness-revalidation.test.ts": "ci-workflows" + "zz-pr-coderabbit-readiness-revalidation.test.ts": "ci-workflows", + "astra-effort-cache-metrics.test.ts": "usage", + "astra-effort-cache-proxy.test.ts": "responses" } diff --git a/tests/helpers/astra-effort-proxy.ts b/tests/helpers/astra-effort-proxy.ts new file mode 100644 index 0000000000..7c50b42a79 --- /dev/null +++ b/tests/helpers/astra-effort-proxy.ts @@ -0,0 +1,117 @@ +import { mkdtempSync, mkdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { pathToFileURL } from "node:url"; +import { fakeChatGptJwt } from "./fake-chatgpt-jwt"; + +export async function startAstraEffortProxy(nativeWebSocket = false, runtimeRoot?: string) { + const home = mkdtempSync(join(tmpdir(), "astra-proxy-")); + const oldEnv = { ...process.env }; + for (const key of Object.keys(process.env)) { + if (/^(OPENAI_|CODEX_|OPENCODEX_)/.test(key) || /^(http|https|all)_proxy$/i.test(key)) delete process.env[key]; + } + Object.assign(process.env, { HOME: home, USERPROFILE: home, OPENCODEX_HOME: join(home, "ocx"), CODEX_HOME: join(home, "codex"), OPENCODEX_API_AUTH_TOKEN: "fixture-admission", NO_PROXY: "127.0.0.1,localhost" }); + mkdirSync(process.env.OPENCODEX_HOME!, { recursive: true }); + mkdirSync(process.env.CODEX_HOME!, { recursive: true }); + const realFetch = globalThis.fetch; + const RealWebSocket = globalThis.WebSocket; + const captured: Array<{ transport: "http" | "websocket"; compact: boolean; body: any }> = []; + let serial = 0; + function events(body: any, transport: "http" | "websocket", compact = false) { + captured.push({ transport, compact, body }); + const id = `resp_fixture_${++serial}`; + const output = [{ id: `msg_fixture_${serial}`, type: "message", role: "assistant", status: "completed", content: [{ type: "output_text", text: "OK", annotations: [] }] }]; + return [ + { type: "response.created", response: { id, status: "in_progress", output: [] } }, + { type: "response.output_text.delta", output_index: 0, content_index: 0, item_id: output[0].id, delta: "OK" }, + { type: "response.output_item.done", output_index: 0, item: output[0] }, + { type: "response.completed", response: { id, object: "response", status: "completed", output, usage: { input_tokens: 100, output_tokens: 1, input_tokens_details: { cached_tokens: 80 } } } }, + ]; + } + const upstream = Bun.serve({ + hostname: "127.0.0.1", port: 0, + async fetch(request, server) { + if (request.headers.get("upgrade") === "websocket" && server.upgrade(request)) return; + if (request.method !== "POST") return Response.json({}); + const body = await request.json(); + const compact = new URL(request.url).pathname.endsWith("/compact"); + const frames = events(body, "http", compact); + if (compact) return Response.json({ id: "cmp_fixture", object: "response.compaction", output: [{ type: "compaction", encrypted_content: "synthetic" }] }); + return new Response(frames.map(frame => `event: ${frame.type}\ndata: ${JSON.stringify(frame)}\n\n`).join(""), { headers: { "content-type": "text/event-stream" } }); + }, + websocket: { message(ws, data) { for (const event of events(JSON.parse(String(data)), "websocket")) ws.send(JSON.stringify(event)); } }, + }); + globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(typeof input === "string" ? input : input instanceof URL ? input.href : input.url); + if (url.origin === "https://chatgpt.com" && url.pathname.startsWith("/backend-api/codex")) { + return realFetch(new URL(url.pathname.slice("/backend-api/codex".length) || "/", upstream.url), init); + } + if (url.hostname === "127.0.0.1") return realFetch(input, init); + return Promise.reject(new Error("Synthetic Astra fixture denies external fetch")); + }) as typeof fetch; + globalThis.WebSocket = new Proxy(RealWebSocket, { + construct(target, args) { + const url = new URL(String(args[0])); + if (url.origin === "wss://chatgpt.com") { + if (!nativeWebSocket) throw new Error("Synthetic HTTP-only upstream"); + const local = new URL("/responses", upstream.url); local.protocol = "ws:"; + return Reflect.construct(target, [local.href, ...args.slice(1)]); + } + if (url.hostname !== "127.0.0.1") throw new Error("Synthetic Astra fixture denies external websocket"); + return Reflect.construct(target, args); + }, + }); + let proxy: Awaited> | undefined; + async function stop() { + try { await proxy?.stop(true); await upstream.stop(true); } + finally { + globalThis.fetch = realFetch; globalThis.WebSocket = RealWebSocket; + for (const key of Object.keys(process.env)) if (!(key in oldEnv)) delete process.env[key]; + Object.assign(process.env, oldEnv); + rmSync(home, { recursive: true, force: true }); + } + } + try { + const [{ saveConfig }, { startServer }] = await Promise.all(runtimeRoot + ? [import(pathToFileURL(join(runtimeRoot, "src/config.ts")).href), import(pathToFileURL(join(runtimeRoot, "src/server/index.ts")).href)] + : [import("../../src/config"), import("../../src/server")]); + saveConfig({ port: 0, hostname: "127.0.0.1", defaultProvider: "openai", openaiProviderTierVersion: 2, websockets: true, + providers: { openai: { adapter: "openai-responses", baseUrl: "https://chatgpt.com/backend-api/codex", authMode: "forward", codexAccountMode: "direct" } } }); + proxy = startServer(0); + } catch (error) { await stop(); throw error; } + const url = new URL("/v1/responses", proxy!.url); + function headers(thread: string) { + return { "content-type": "application/json", "x-opencodex-api-key": "fixture-admission", authorization: `Bearer ${fakeChatGptJwt({ chatgpt_account_id: "fixture-account" })}`, "chatgpt-account-id": "fixture-account", "thread-id": thread, "session-id": thread, "x-codex-parent-thread-id": thread }; + } + async function http(body: unknown, thread: string, compact = false) { + const response = await realFetch(compact ? `${url}/compact` : url, { method: "POST", headers: headers(thread), body: JSON.stringify(body), signal: AbortSignal.timeout(10_000) }); + const text = await response.text(); + if (!response.ok) throw new Error(`Synthetic request failed: ${response.status}: ${text.slice(0, 300)}`); + return compact ? JSON.parse(text) : JSON.parse(text.split("\n").find(line => line.startsWith("data:") && line.includes('"type":"response.completed"'))!.slice(5)).response; + } + function websocket(thread: string) { + const wsUrl = new URL(url); wsUrl.protocol = "ws:"; + const ws = new RealWebSocket(wsUrl, { headers: headers(thread) } as unknown as string[]); + const ready = new Promise((resolve, reject) => { + const timer = setTimeout(() => { ws.close(); reject(new Error("Synthetic websocket open timeout")); }, 10_000); + ws.addEventListener("open", () => { clearTimeout(timer); resolve(); }, { once: true }); + ws.addEventListener("error", () => { clearTimeout(timer); reject(new Error("Synthetic websocket open failed")); }, { once: true }); + }); + return { close: () => ws.close(), async turn(body: object) { + await ready; + return new Promise((resolve, reject) => { + const finish = (error?: Error, value?: unknown) => { clearTimeout(timer); ws.removeEventListener("message", onMessage); ws.removeEventListener("close", onClose); error ? reject(error) : resolve(value); }; + const onMessage = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)); + if (frame.type === "response.completed") finish(undefined, frame.response); + else if (["error", "response.failed"].includes(frame.type)) finish(new Error("Synthetic websocket request failed")); + }; + const onClose = () => finish(new Error("Synthetic websocket closed before completion")); + const timer = setTimeout(() => { finish(new Error("Synthetic websocket turn timeout")); ws.close(); }, 10_000); + ws.addEventListener("message", onMessage); ws.addEventListener("close", onClose, { once: true }); + ws.send(JSON.stringify({ ...body, type: "response.create" })); + }); + } }; + } + return { home: join(home, "ocx"), captured, http, websocket, stop }; +} diff --git a/tests/responses/astra-effort-cache-proxy.test.ts b/tests/responses/astra-effort-cache-proxy.test.ts new file mode 100644 index 0000000000..4d0c2f10c8 --- /dev/null +++ b/tests/responses/astra-effort-cache-proxy.test.ts @@ -0,0 +1,58 @@ +import { Database } from "bun:sqlite"; +import { join } from "node:path"; +import { expect, test } from "bun:test"; +import { startAstraEffortProxy } from "../helpers/astra-effort-proxy"; +import { readRecentUsageEntries } from "../../src/usage/log"; + +const user = (content: string) => ({ role: "user", content }); +const body = (input: unknown[], effort = "medium", extra = {}) => ({ model: "gpt-6-astra", instructions: "Synthetic", input, reasoning: { effort }, stream: true, store: false, ...extra }); +const updates = (input: any[]) => input.filter(item => item.type === "configuration_update").map(item => item.reasoning.effort); + +for (const native of [false, true]) { + test(`Astra real HTTP and WebSocket ingress, continuation and compact with ${native ? "WebSocket" : "HTTP"} upstream`, async () => { + const fixture = await startAstraEffortProxy(native); + const ws = fixture.websocket("ws-fixture"); + try { + for (const [thread, send] of [["http-fixture", (b: object) => fixture.http(b, "http-fixture")], ["ws-fixture", ws.turn]] as const) { + const first = [user("one")]; + const r1 = await send(body(first)); + const second = [...first, ...r1.output, user("two")]; + const r2 = await send(body(second, "low")); + expect(fixture.captured.at(-1)!.body.reasoning.effort).toBe("medium"); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual(["low"]); + await send(body([user("three")], "medium", { previous_response_id: r2.id })); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual(["low", "medium"]); + expect(updates(second)).toEqual([]); + await fixture.http(body(second, "high"), `${thread}-sibling`); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual([]); + } + const resumeInput = [user("resume-one")]; + const initial = await fixture.http(body(resumeInput), "resume-fixture"); + const reconnect = fixture.websocket("resume-fixture"); + try { + await reconnect.turn(body([user("resume-two")], "low", { previous_response_id: initial.id })); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual(["low"]); + } finally { reconnect.close(); } + const lock = new Database(join(fixture.home, "astra-effort-cache", "state.sqlite")); + lock.exec("BEGIN IMMEDIATE"); + try { + await fixture.http(body([...resumeInput, ...initial.output, user("busy")], "high"), "resume-fixture"); + expect(fixture.captured.at(-1)!.body.reasoning.effort).toBe("high"); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual([]); + } finally { lock.exec("ROLLBACK"); lock.close(); } + await fixture.http(body([user("after compaction"), { type: "compaction_trigger" }], "high"), "resume-fixture"); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual([]); + const compact = await fixture.http(body([user("compact")], "high"), "http-fixture", true); + expect(compact.object).toBe("response.compaction"); + expect(fixture.captured.at(-1)!.compact).toBe(true); + expect(updates(fixture.captured.at(-1)!.body.input)).toEqual([]); + await Promise.all(Array.from({ length: 4 }, (_, i) => fixture.http(body([user("concurrent")]), `concurrent-${i}`))); + const rows = readRecentUsageEntries(100, fixture.home); + expect(rows.some(row => row.astraEffortCache?.status === "updated")).toBe(true); + expect(rows.some(row => row.astraEffortCache?.stateOutcome === "busy")).toBe(true); + expect(rows.some(row => row.attempts?.some(attempt => attempt.astraEffortCache?.status === "updated"))).toBe(true); + expect(rows.filter(row => row.astraEffortCache).every(row => row.astraEffortCache!.durationMs >= 0)).toBe(true); + expect(fixture.captured.some(row => row.transport === (native ? "websocket" : "http"))).toBe(true); + } finally { ws.close(); await fixture.stop(); } + }, 40_000); +} diff --git a/tests/responses/astra-effort-cache.test.ts b/tests/responses/astra-effort-cache.test.ts index c30796c4e7..ee98dc8189 100644 --- a/tests/responses/astra-effort-cache.test.ts +++ b/tests/responses/astra-effort-cache.test.ts @@ -69,7 +69,8 @@ describe("Astra effort history", () => { run(first); const low = run(second, "low"); expect(low.status).toBe("updated"); const before = readState(); - expect(run(second, "low")).toEqual({ ...low, status: "replay" }); + const { metrics: _metrics, ...lowResult } = low; + expect(run(second, "low")).toMatchObject({ ...lowResult, status: "replay" }); expect(readState()).toBe(before); }); test("restart/resume reads disk state without process memory", () => { @@ -126,7 +127,7 @@ describe("Astra effort history", () => { }); test("corrupt state fails transparently without overwriting it", () => { run(first); writeFileSync(statePath(), "invalid"); - expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low") }); + expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low"), metrics: { stateOutcome: "error" } }); expect(readFileSync(statePath(), "utf8")).toBe("invalid"); }); test("invalid update positions fail state validation", () => { @@ -147,7 +148,7 @@ describe("Astra effort history", () => { run(first); const lock = new Database(statePath()); lock.exec("BEGIN IMMEDIATE"); - try { expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low") }); } + try { expect(run(second, "low")).toMatchObject({ status: "unavailable_state", body: body(second, "low"), metrics: { stateOutcome: "busy" } }); } finally { lock.exec("ROLLBACK"); lock.close(); } expect(run(second, "low").status).toBe("updated"); }); diff --git a/tests/usage/astra-effort-cache-metrics.test.ts b/tests/usage/astra-effort-cache-metrics.test.ts new file mode 100644 index 0000000000..77cd02f28c --- /dev/null +++ b/tests/usage/astra-effort-cache-metrics.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync, readFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { normalizeAstraEffortCacheMetrics } from "../../src/usage/astra-effort-cache"; +import { appendUsageEntry, readRecentUsageEntries, type PersistedUsageEntry } from "../../src/usage/log"; +import { recordAdapterReasoning, type RequestLogContext } from "../../src/server/request-log"; +import { summarizeAstraEffortCache } from "../../scripts/astra-effort-cache-report"; + +const metrics = { status: "updated", stateOutcome: "committed", durationMs: 3, setupMs: 1, transactionMs: 1, historyMs: 0.5, closeMs: 0.1, inputItems: 3, updateCount: 1 } as const; +const row: PersistedUsageEntry = { requestId: "fixture", timestamp: 1, provider: "openai", model: "gpt-6-astra", status: 200, durationMs: 10, usageStatus: "reported", usage: { inputTokens: 100, outputTokens: 1, cachedInputTokens: 80 }, astraEffortCache: metrics }; + +describe("Astra effort cache measurements", () => { + test("allowlists metadata and rejects malformed values without leaking strings", () => { + expect(normalizeAstraEffortCacheMetrics({ ...metrics, body: "private", account: "private" })).toEqual(metrics); + for (const change of [{ durationMs: NaN }, { closeMs: -1 }, { inputItems: 1.2 }, { status: "private" }, { stateOutcome: "private" }]) { + expect(normalizeAstraEffortCacheMetrics({ ...metrics, ...change })).toBeUndefined(); + } + expect(normalizeAstraEffortCacheMetrics({ get status() { throw new Error("private"); } })).toBeUndefined(); + }); + test("persists bounded metadata through the existing usage ledger", () => { + const home = mkdtempSync(join(tmpdir(), "astra-usage-")); + const oldHome = process.env.OPENCODEX_HOME; + process.env.OPENCODEX_HOME = home; + try { + appendUsageEntry({ ...row, astraEffortCache: { ...metrics, privateField: "private-sentinel" } as typeof metrics }); + expect(readRecentUsageEntries(10, home)[0].astraEffortCache).toEqual(metrics); + expect(readFileSync(join(home, "usage.jsonl"), "utf8")).not.toContain("private-sentinel"); + } finally { + if (oldHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = oldHome; + rmSync(home, { recursive: true, force: true }); + } + }); + test("records metrics without reasoning metadata and clears them on another adapter build", () => { + const ctx = { activeAttempt: {} } as RequestLogContext; + recordAdapterReasoning(ctx, { url: "http://localhost", method: "POST", headers: {}, body: "{}", astraEffortCache: metrics }); + expect(ctx.astraEffortCache).toEqual(metrics); + expect(ctx.activeAttempt?.astraEffortCache).toEqual(metrics); + recordAdapterReasoning(ctx, { url: "http://localhost", method: "POST", headers: {}, body: "{}" }); + expect(ctx.astraEffortCache).toBeUndefined(); + expect(ctx.activeAttempt?.astraEffortCache).toBeUndefined(); + }); + test("counts attempts once and separates measured cache usage from unknown or invalid usage", () => { + const attempt = { ...row, ordinal: 1, adapter: "openai-responses", sendCount: 1, recoveryKinds: [] }; + const report = summarizeAstraEffortCache([ + { ...row, attempts: [attempt] }, + { ...row, usageStatus: "estimated" }, { ...row, usage: { inputTokens: 100, outputTokens: 1 } }, + { ...row, usage: { inputTokens: 100, outputTokens: 1, cachedInputTokens: 0 } }, + { ...row, usage: { inputTokens: 10, outputTokens: 1, cachedInputTokens: 80 } }, + { ...row, astraEffortCache: undefined }, + ]); + expect(report.samples).toBe(5); + expect(report.cache).toEqual({ hit: 1, miss: 1, unknown: 2, invalid: 1, inputTokens: 200, cachedInputTokens: 80, cachedInputRatio: 0.4 }); + expect(summarizeAstraEffortCache([]).timingsMs.durationMs.p99).toBeNull(); + }); +}); From 2ecb749d832e28c746ac3e507181147ee62bb900 Mon Sep 17 00:00:00 2001 From: nahuelb Date: Fri, 11 Sep 2026 01:38:22 -0300 Subject: [PATCH 6/7] Respect runtime transport gates in Astra integration checks --- .../src/content/docs/reference/configuration/server.md | 4 +++- scripts/astra-effort-cache-eval.ts | 7 ++++--- tests/responses/astra-effort-cache-proxy.test.ts | 4 +++- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 2717e776be..544d84a523 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -574,7 +574,9 @@ commit, and dirty-patch digest. Separate processes exercise a shared store with 1 MiB synthetic user text; each fresh store's first call includes database creation. Subsequent calls cover new conversations, effort switches, and replay. Concurrent writers can intentionally fall back when SQLite is busy. Real proxy cells use concurrent HTTP and WebSocket clients with both HTTP/SSE -and WebSocket upstream fixtures. Client latency includes the whole local request; timer delay measures +and WebSocket upstream fixtures. `upstreamWebSocketAvailable` describes the fixture; +`observedUpstreamTransports` records what the proxy actually used. Runtime capability gates can +select HTTP fallback even when the fixture supports WebSockets, including on prerelease Bun builds. Client latency includes the whole local request; timer delay measures blocking in that fixture process. These are local costs, not production latency or upstream cache-hit proof. Synthetic token counts must never be interpreted as observed model cache savings. diff --git a/scripts/astra-effort-cache-eval.ts b/scripts/astra-effort-cache-eval.ts index 5eeef7f7f9..fffbc020b7 100644 --- a/scripts/astra-effort-cache-eval.ts +++ b/scripts/astra-effort-cache-eval.ts @@ -68,7 +68,8 @@ if (mode === "--worker") { await Bun.sleep(10); const elapsedMs = performance.now() - started; const measurements = summarizeAstraEffortCache(readRecentUsageEntries(10_000, fixture.home)); - console.log(JSON.stringify({ samples, elapsedMs, requestsPerSecond: samples.length / elapsedMs * 1000, maxTimerDelayMs, measurements })); + const observedUpstreamTransports = [...new Set(fixture.captured.map(row => row.transport))].sort(); + console.log(JSON.stringify({ samples, elapsedMs, requestsPerSecond: samples.length / elapsedMs * 1000, maxTimerDelayMs, observedUpstreamTransports, measurements })); } finally { clearInterval(timer); await fixture.stop(); } } else { const outDir = mode; @@ -124,11 +125,11 @@ if (mode === "--worker") { } for (const native of [false, true]) { for (const [arm, runtime] of [["treatment", undefined], ...(controlRoot ? [["control", resolve(controlRoot)]] : [])] as const) { - cells.push({ kind: "proxy", arm, nativeUpstreamWebSocket: native, inputTextBytes: 64 * 1024, workers: concurrency, + cells.push({ kind: "proxy", arm, upstreamWebSocketAvailable: native, inputTextBytes: 64 * 1024, workers: concurrency, ...await result(spawn(["--proxy", String(native), String(count), String(concurrency), String(64 * 1024), ...(runtime ? [runtime] : [])])) }); } } - const report = { schemaVersion: 1, synthetic: true, platform: process.platform, arch: process.arch, bun: Bun.version, treatment: revision(root), ...(controlRoot ? { control: revision(resolve(controlRoot)) } : {}), cells }; + const report = { schemaVersion: 1, synthetic: true, platform: process.platform, arch: process.arch, bun: Bun.version, bunVersionWithSha: Bun.version_with_sha, treatment: revision(root), ...(controlRoot ? { control: revision(resolve(controlRoot)) } : {}), cells }; writeFileSync(join(outDir, "samples.jsonl"), cells.flatMap((cell, i) => cell.samples.map((sample: unknown) => JSON.stringify({ cell: i, sample }))).join("\n") + "\n"); writeFileSync(join(outDir, "report.json"), JSON.stringify({ ...report, cells: cells.map(({ samples, ...cell }) => ({ ...cell, sampleCount: samples.length, ...(cell.kind === "proxy" ? { wallMs: distribution(samples.map((row: any) => row.wallMs)) } : {}) })) }, null, 2) + "\n"); console.log("Synthetic benchmark complete: report.json and samples.jsonl"); diff --git a/tests/responses/astra-effort-cache-proxy.test.ts b/tests/responses/astra-effort-cache-proxy.test.ts index 4d0c2f10c8..eb0fd00d15 100644 --- a/tests/responses/astra-effort-cache-proxy.test.ts +++ b/tests/responses/astra-effort-cache-proxy.test.ts @@ -3,6 +3,7 @@ import { join } from "node:path"; import { expect, test } from "bun:test"; import { startAstraEffortProxy } from "../helpers/astra-effort-proxy"; import { readRecentUsageEntries } from "../../src/usage/log"; +import { bunSupportsBoundedCodexWsRelay, currentBunRuntimeIdentity } from "../../src/server/responses/ws-upstream"; const user = (content: string) => ({ role: "user", content }); const body = (input: unknown[], effort = "medium", extra = {}) => ({ model: "gpt-6-astra", instructions: "Synthetic", input, reasoning: { effort }, stream: true, store: false, ...extra }); @@ -52,7 +53,8 @@ for (const native of [false, true]) { expect(rows.some(row => row.astraEffortCache?.stateOutcome === "busy")).toBe(true); expect(rows.some(row => row.attempts?.some(attempt => attempt.astraEffortCache?.status === "updated"))).toBe(true); expect(rows.filter(row => row.astraEffortCache).every(row => row.astraEffortCache!.durationMs >= 0)).toBe(true); - expect(fixture.captured.some(row => row.transport === (native ? "websocket" : "http"))).toBe(true); + const expectedTransport = native && bunSupportsBoundedCodexWsRelay(currentBunRuntimeIdentity()) ? "websocket" : "http"; + expect(fixture.captured.filter(row => !row.compact).every(row => row.transport === expectedTransport)).toBe(true); } finally { ws.close(); await fixture.stop(); } }, 40_000); } From 98cb59428a79691386ba6826e8338387d09ebe46 Mon Sep 17 00:00:00 2001 From: nahuelb Date: Fri, 11 Sep 2026 01:43:55 -0300 Subject: [PATCH 7/7] Match provider integration transport expectations to Bun capability --- tests/adapters/openai/openai-provider-option-e2e.test.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/adapters/openai/openai-provider-option-e2e.test.ts b/tests/adapters/openai/openai-provider-option-e2e.test.ts index 903d85c82a..6d0220e44a 100644 --- a/tests/adapters/openai/openai-provider-option-e2e.test.ts +++ b/tests/adapters/openai/openai-provider-option-e2e.test.ts @@ -624,9 +624,10 @@ describe("OpenAI provider-option integration spine", () => { removeTreeWithRetry(migrationRoot); } - expect(new Set(blockedUpstreamWebSocketUrls)).toEqual(new Set([ - "wss://chatgpt.com/backend-api/codex/responses", - ])); + const { bunSupportsBoundedCodexWsRelay, currentBunRuntimeIdentity } = await import("../../../src/server/responses/ws-upstream"); + const expectedWebSocketUrls = bunSupportsBoundedCodexWsRelay(currentBunRuntimeIdentity()) + ? ["wss://chatgpt.com/backend-api/codex/responses"] : []; + expect(new Set(blockedUpstreamWebSocketUrls)).toEqual(new Set(expectedWebSocketUrls)); if (process.platform === "win32") { expect(aclSeamCalls).toBeGreaterThan(0); expect(principalSeamCalls).toBeGreaterThan(0);