feat: Change Astra reasoning mid-conversation without breaking cache

docs-site/src/content/docs/reference/configuration/server.md

@@ -480,3 +480,106 @@ 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.
+
+## Astra effort cache preservation
+
+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.
+
+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 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
+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. 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
+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 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
+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`
+indicate that the optimization was not applied to that request.
+
+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).
+
+
+### 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. `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.
+
+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.

scripts/astra-effort-cache-eval.ts

@@ -0,0 +1,141 @@
+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));
+    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;
+  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 <outDir> [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<typeof Bun.spawn>[] = [];
+  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<string, string | undefined>)[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<typeof spawn>) {
+    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<string, number>) });
+      }
+    }
+    for (const native of [false, true]) {
+      for (const [arm, runtime] of [["treatment", undefined], ...(controlRoot ? [["control", resolve(controlRoot)]] : [])] as const) {
+        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, 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");
+  } 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 });
+  }
+}