diff --git a/src/bridge.ts b/src/bridge.ts index 3b31746c94..c8880d5426 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -198,6 +198,16 @@ export function bridgeToResponsesSSE( declaredToolNames?: ReadonlySet; /** Declared parameter schema per tool name; repairs integral-float integer args (#1611). */ toolParameterSchemas?: ReadonlyMap>; + /** + * Wire keep-alive shape. Codex-rs parses at the EVENT level (timeout(idle_timeout, + * stream.next()) over an eventsource_stream), so an SSE comment line dispatches no event + * and does NOT re-arm its idle timer — the keep-alive must be a typed frame the parser + * ignores via its catch-all (110 RCA, 30_patch-direction.md). grok-build's strict + * async-openai fork is the opposite: it dies on the unknown `response.heartbeat` + * variant but, being eventsource-based at the byte level, its idle handling tolerates + * comment lines. Default stays the typed frame; the grok surface opts into comments. + */ + heartbeatStyle?: "typed" | "comment"; translatorBudget?: TranslatorBudget; /** * Conversation identity for the reasoning replay cache (issue #950). @@ -326,12 +336,13 @@ export function bridgeToResponsesSSE( clearOwnedWatchdog(); }; // RC3 keep-alive: Codex's idle timer is timeout(idle_timeout, stream.next()) over an - // eventsource_stream; ANY received bytes re-arm it. An SSE comment line (a line starting - // with `:`) is discarded by every eventsource parser without producing an event, so it - // keeps the wire alive without triggering deserialization. Emit a comment line whenever the - // *wire* has been silent, even if invisible adapter heartbeats are still flowing (web-search - // buffering + raw-byte progress). Upstream activity only resets the stall watchdog. Parity - // with the passthrough relay's `: opencodex keepalive` (relay.ts). + // eventsource_stream, which parses at the EVENT level — a comment-only frame dispatches no + // event, so it does NOT re-arm the timer (110 RCA). The default keep-alive is therefore a + // typed `response.heartbeat` frame the codex-rs parser ignores via `_ => Ok(None)`. The + // grok surface (strict async-openai decoder that dies on unknown variants) opts into SSE + // comment lines instead via options.heartbeatStyle. Emit whenever the *wire* has been + // silent, even if invisible adapter heartbeats are still flowing (web-search buffering + + // raw-byte progress). Upstream activity only resets the stall watchdog. let upstreamActivity = false; let wireActivity = false; let beat: unknown; @@ -398,7 +409,9 @@ export function bridgeToResponsesSSE( ...(endTurn !== undefined ? { end_turn: endTurn } : {}), }); - const heartbeatFrame = encoder.encode(': opencodex heartbeat\n\n'); + const heartbeatFrame = options?.heartbeatStyle === "comment" + ? encoder.encode(': opencodex heartbeat\n\n') + : encoder.encode('event: response.heartbeat\ndata: {"type":"response.heartbeat"}\n\n'); let stallTicks = 0; const stallSec = resolveStallTimeoutSec(options?.stallTimeoutSec); const maxStallTicks = Math.ceil((stallSec * 1000) / heartbeatMs); diff --git a/src/codex/auth-api.ts b/src/codex/auth-api.ts index f2d08188b5..b0b34bcd02 100644 --- a/src/codex/auth-api.ts +++ b/src/codex/auth-api.ts @@ -854,6 +854,15 @@ function reconcileFreshPoolAccountPlans(runtimeConfig: OcxConfig, updates: Fresh accepted.push(update); if (persistedAccount.plan !== update.plan) { persistedAccount.plan = update.plan; + // WHAM is the authoritative plan source: stamp provenance so a later JWT + // reconcile cannot overwrite this observation within the same credential + // generation (src/codex/plan-from-token.ts jwtMayWritePlan). Stamped only + // alongside a real plan change: a steady-state refresh whose plan is + // unchanged must stay write-free (no-config-write contract), and an + // unchanged value needs no fence — a JWT rewrite to the same text is a + // no-op under the caller's own equality check. + persistedAccount.planSource = "wham"; + persistedAccount.planCredentialGeneration = update.credentialGeneration; changed = true; } } @@ -873,6 +882,8 @@ function reconcileFreshPoolAccountPlans(runtimeConfig: OcxConfig, updates: Fresh const liveAccount = configuredPoolAccount(runtimeConfig, update.accountId); if (liveAccount) { liveAccount.plan = update.plan; + liveAccount.planSource = "wham"; + liveAccount.planCredentialGeneration = update.credentialGeneration; } } } diff --git a/src/codex/plan-from-token.ts b/src/codex/plan-from-token.ts index 40549364aa..be2585ec4e 100644 --- a/src/codex/plan-from-token.ts +++ b/src/codex/plan-from-token.ts @@ -29,6 +29,19 @@ function jwtPlanFromPoolCredential(accountId: string): string | undefined { return cred ? extractChatgptPlanType(undefined, cred.accessToken) : undefined; } +/** + * WHAM-wins gate (release-audit fix). A JWT-derived plan may be persisted only when no + * WHAM-sourced plan exists for the CURRENT credential generation. A token refresh bumps the + * generation, and the refreshed JWT is then genuinely newer information than the previous + * generation's WHAM read, so it may write again until WHAM re-observes. Records without + * provenance (legacy) stay writable so the original #1989 recovery still works. + */ +function jwtMayWritePlan(account: CodexAccount, generation: number): boolean { + if (account.planSource !== "wham") return true; + const whamGeneration = account.planCredentialGeneration; + return whamGeneration !== undefined && generation > whamGeneration; +} + function collectJwtPoolPlanUpdates(runtimeConfig: OcxConfig): FreshPoolPlanUpdate[] { const updates: FreshPoolPlanUpdate[] = []; for (const account of (runtimeConfig.codexAccounts ?? []).filter(isSelectableCodexPoolAccount)) { @@ -36,6 +49,7 @@ function collectJwtPoolPlanUpdates(runtimeConfig: OcxConfig): FreshPoolPlanUpdat if (!jwtPlan || codexPlanValue(account.plan) === jwtPlan) continue; const generation = readCodexAccountRecord(account.id)?.generation; if (generation === undefined) continue; + if (!jwtMayWritePlan(account, generation)) continue; updates.push({ accountId: account.id, plan: jwtPlan, credentialGeneration: generation }); } return updates; @@ -55,11 +69,19 @@ function persistJwtPlanUpdates(runtimeConfig: OcxConfig, updates: FreshPoolPlanU const liveAccount = configuredPoolAccount(runtimeConfig, update.accountId); const persistedAccount = configuredPoolAccount(persistedConfig, update.accountId); if (!liveAccount || !persistedAccount) continue; + // Re-check against the PERSISTED row: another process may have landed a WHAM + // observation between collect and this mutation. + if (!jwtMayWritePlan(persistedAccount, update.credentialGeneration)) continue; accepted.push(update); if (persistedAccount.plan !== update.plan) { persistedAccount.plan = update.plan; changed = true; } + if (persistedAccount.planSource !== "jwt" || persistedAccount.planCredentialGeneration !== update.credentialGeneration) { + persistedAccount.planSource = "jwt"; + persistedAccount.planCredentialGeneration = update.credentialGeneration; + changed = true; + } } return { changed, value: accepted }; }); @@ -73,6 +95,8 @@ function persistJwtPlanUpdates(runtimeConfig: OcxConfig, updates: FreshPoolPlanU const liveAccount = configuredPoolAccount(runtimeConfig, update.accountId); if (liveAccount) { liveAccount.plan = update.plan; + liveAccount.planSource = "jwt"; + liveAccount.planCredentialGeneration = update.credentialGeneration; appliedJwtPlans.set(update.accountId, update.plan); } } @@ -107,6 +131,7 @@ export function noteCodexAccountAccessToken( appliedJwtPlans.set(accountId, jwtPlan); return; } + if (!jwtMayWritePlan(live, credentialGeneration)) return; persistJwtPlanUpdates(runtimeConfig, [{ accountId, plan: jwtPlan, credentialGeneration }]); if (codexPlanValue(live.plan) === jwtPlan) appliedJwtPlans.set(accountId, jwtPlan); } catch { diff --git a/src/providers/service-tier.ts b/src/providers/service-tier.ts index 4eebbcbe05..a06c42c7b4 100644 --- a/src/providers/service-tier.ts +++ b/src/providers/service-tier.ts @@ -210,9 +210,18 @@ export function serviceTierSupportForModel( /** Compatibility projection shared by catalog, routing, and request logging. */ export function serviceTierSupportFromPolicy( - policy: Pick, + policy: Pick, ): boolean | undefined { if (policy.eligibility === "eligible") return true; - if (policy.eligibility === "unclassified") return undefined; + if (policy.eligibility === "unclassified") { + // B1 regression guard: an unclassified chat-wire route whose final adapter will not + // forward any tier cannot serialize service_tier, so projecting "unknown" would let + // require.serviceTier: "unsupported" routing stop matching groq/ollama-class providers + // that main projected as false. Chat + no forwarding stays a definitive false; a + // chat route with chatServiceTier: true (forwarding allowed) keeps the historical + // unknown, as does every unclassified Responses-wire route. + if (policy.adapter === "openai-chat" && !policy.forwardCallerTier) return false; + return undefined; + } return false; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index ff201a0aac..0c36d0cb82 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -3505,6 +3505,9 @@ async function handleResponsesInner( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true } : {}), + // grok-build's strict decoder dies on the typed response.heartbeat frame; its + // eventsource layer tolerates comment keep-alives. Codex needs the opposite. + ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), onUsage: usage => { // Raw adapter usage, pre wire-normalization: the bridged SSE now always carries // zero-default detail objects, so provenance must come from here (cache_detail_missing). @@ -4312,6 +4315,8 @@ async function handleResponsesInner( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true } : {}), + // Same grok-surface split as the runTurn branch above. + ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), onUsage: usage => { // Raw adapter usage, pre wire-normalization (see the runTurn branch above). logCtx.usageFromBridge = true; diff --git a/src/types.ts b/src/types.ts index d54860a8ba..ecf375f265 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1843,6 +1843,15 @@ export interface CodexAccount { /** User-owned display label; never participates in routing or identity checks. */ alias?: string; plan?: string; + /** + * Provenance of `plan`. WHAM (live quota API) is authoritative; the JWT + * `chatgpt_plan_type` claim is a fallback that may lag a plan change. A JWT write + * must never overwrite a WHAM-sourced plan observed for the same credential + * generation — only a newer generation (token refresh after the WHAM read) may. + */ + planSource?: "jwt" | "wham"; + /** Credential generation at which `plan`/`planSource` was recorded. */ + planCredentialGeneration?: number; chatgptAccountId?: string; logLabel?: string; isMain: boolean; diff --git a/tests/bridge-lifecycle.test.ts b/tests/bridge-lifecycle.test.ts index 47d77166bd..7caaec643d 100644 --- a/tests/bridge-lifecycle.test.ts +++ b/tests/bridge-lifecycle.test.ts @@ -246,8 +246,10 @@ describe("bridge stream lifecycle (RC1 / RC2)", () => { expect(aborted).toBe(true); }); - test("RC3: emits an SSE comment keep-alive (no response.heartbeat event) during upstream silence", async () => { - // heartbeatMs = 10 so the keep-alive fires quickly; hangs() goes silent after one delta. + test("RC3: default keep-alive is a typed response.heartbeat frame (codex idle timer re-arm)", async () => { + // Codex-rs parses at the EVENT level: a comment-only frame dispatches no event and does + // NOT re-arm timeout(idle_timeout, stream.next()) — 110 RCA. The typed frame is ignored + // by its catch-all (_ => Ok(None)) but still yields an event. const stream = bridgeToResponsesSSE(hangs(), "routed/model", undefined, undefined, undefined, undefined, 10); const reader = stream.getReader(); const dec = new TextDecoder(); @@ -258,8 +260,26 @@ describe("bridge stream lifecycle (RC1 / RC2)", () => { if (value) text += dec.decode(value, { stream: true }); } await reader.cancel(); - // Keep-alives are SSE comment lines, not typed events: any client parser discards - // them without deserializing, so strict Responses decoders stay alive and quiet. + expect(text).toContain("event: response.heartbeat"); + expect(text).not.toContain(": opencodex heartbeat"); + }); + + test("RC3: the grok surface opts into comment keep-alives (strict decoder safety)", async () => { + // grok-build's async-openai fork dies on the unknown response.heartbeat variant; its + // eventsource layer tolerates comment lines. heartbeatStyle: "comment" preserves that. + const stream = bridgeToResponsesSSE( + hangs(), "routed/model", undefined, undefined, undefined, undefined, 10, + { heartbeatStyle: "comment" }, + ); + const reader = stream.getReader(); + const dec = new TextDecoder(); + let text = ""; + for (let i = 0; i < 12; i++) { + const { value, done } = await reader.read(); + if (done) break; + if (value) text += dec.decode(value, { stream: true }); + } + await reader.cancel(); expect(text).toContain(": opencodex heartbeat\n\n"); expect(text).not.toContain("event: response.heartbeat"); }); diff --git a/tests/bridge.test.ts b/tests/bridge.test.ts index 8c92b4ef9d..5db8fdd054 100644 --- a/tests/bridge.test.ts +++ b/tests/bridge.test.ts @@ -803,11 +803,12 @@ describe("Responses bridge reasoning and usage parity", () => { expect(frames.some(f => f.data.type === "heartbeat")).toBe(false); }); - test("wire keepalive comment keeps firing while only adapter heartbeats flow", async () => { + test("wire keepalive keeps firing while only adapter heartbeats flow", async () => { // Issue #521: web-search buffers semantic events and yields invisible adapter heartbeats from // raw-byte progress. Those must not suppress wire keepalives, or Codex Desktop idle-timeouts - // (~5 min) while OCX still considers the upstream alive. The wire keepalive is an SSE comment - // line (": opencodex heartbeat") so it never triggers deserialization on any client. + // (~5 min) while OCX still considers the upstream alive. The default keep-alive is the typed + // response.heartbeat frame (codex-rs re-arms only on parsed EVENTS — 110 RCA); the grok + // surface swaps to comment lines via heartbeatStyle. const heartbeatMs = 50; const stallTimeoutSec = 1; const cycles = 4; @@ -870,15 +871,14 @@ describe("Responses bridge reasoning and usage parity", () => { const lines = trimmed.split("\n"); const event = lines.find(l => l.startsWith("event: "))?.slice(7); const dataLine = lines.find(l => l.startsWith("data: ")); - // Skip comment-only frames (e.g. ": opencodex heartbeat"); they have no data - // line and must not become fake deserializable events. + // Skip data-less frames; a keep-alive frame carries its own data line now. if (!dataLine) continue; frames.push({ event, data: JSON.parse(dataLine?.slice(6) ?? "{}") as Record }); } - // Wire keepalives are SSE comment lines (": opencodex heartbeat") — they keep the - // idle timer alive without producing a typed event any client must deserialize. - const keepaliveCount = (rawText.match(/^: opencodex heartbeat$/gm) ?? []).length; + // Wire keepalives are typed response.heartbeat frames — codex-rs ignores the unknown + // variant but its eventsource layer still yields an event, re-arming the idle timer. + const keepaliveCount = (rawText.match(/^event: response.heartbeat$/gm) ?? []).length; expect(keepaliveCount).toBeGreaterThan(1); expect(frames.some(f => f.event === "response.completed")).toBe(true); expect(frames.some(f => (f.data.response as Record | undefined)?.incomplete_details)).toBe(false); diff --git a/tests/codex-auth-api.test.ts b/tests/codex-auth-api.test.ts index 027c28301e..9f035ebc10 100644 --- a/tests/codex-auth-api.test.ts +++ b/tests/codex-auth-api.test.ts @@ -1318,6 +1318,11 @@ describe("codex-auth API", () => { id: "pool-plan-unchanged", email: "pool-plan-unchanged@example.com", plan: "plus", + // Provenance already stamped: steady state. The FIRST WHAM observation after the + // provenance feature landed performs one migration write; that case is covered by + // the WHAM-wins gate tests. Steady-state refreshes must stay write-free. + planSource: "wham", + planCredentialGeneration: 1, }); saveConfig(structuredClone(config)); let configCommits = 0; diff --git a/tests/codex-plan.test.ts b/tests/codex-plan.test.ts index 7654a5f780..0c4b9c0936 100644 --- a/tests/codex-plan.test.ts +++ b/tests/codex-plan.test.ts @@ -124,3 +124,77 @@ describe("getMainAccountPlan JWT fallback", () => { expect(getMainAccountPlan()).toBe("pro"); }); }); + +describe("WHAM-wins plan provenance gate (release-audit fix)", () => { + test("a same-generation JWT cannot overwrite a WHAM-sourced plan", () => { + const config: OcxConfig = { + port: 10100, + providers: {}, + defaultProvider: "openai", + codexAccounts: [{ + id: "pool-wham-fence", email: "fence@example.test", plan: "pro", + planSource: "wham", planCredentialGeneration: 1, isMain: false, + }], + }; + saveConfig(config); + saveCodexAccountCredential("pool-wham-fence", { + accessToken: chatgptPlanJwt("plus", "acct-pool-wham-fence"), + refreshToken: "refresh-pool-wham-fence", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "acct-pool-wham-fence", + }); + // credential save above starts at generation 1 == fence generation + reconcileCodexPlansFromTokens(config); + expect(config.codexAccounts?.[0]?.plan).toBe("pro"); + expect(loadConfig().codexAccounts?.[0]?.plan).toBe("pro"); + }); + + test("a newer-generation JWT (token refresh after the WHAM read) may write again", () => { + const config: OcxConfig = { + port: 10100, + providers: {}, + defaultProvider: "openai", + codexAccounts: [{ + id: "pool-wham-stale", email: "stale@example.test", plan: "pro", + planSource: "wham", planCredentialGeneration: 0, isMain: false, + }], + }; + saveConfig(config); + saveCodexAccountCredential("pool-wham-stale", { + accessToken: chatgptPlanJwt("plus", "acct-pool-wham-stale"), + refreshToken: "refresh-pool-wham-stale", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "acct-pool-wham-stale", + }); + // saved credential generation (>=1) is newer than the WHAM fence at 0 + reconcileCodexPlansFromTokens(config); + expect(config.codexAccounts?.[0]?.plan).toBe("plus"); + const persisted = loadConfig().codexAccounts?.[0]; + expect(persisted?.plan).toBe("plus"); + expect(persisted?.planSource).toBe("jwt"); + }); + + test("the gate survives a restart because provenance is persisted, not in-memory", () => { + const config: OcxConfig = { + port: 10100, + providers: {}, + defaultProvider: "openai", + codexAccounts: [{ + id: "pool-wham-restart", email: "restart@example.test", plan: "pro", + planSource: "wham", planCredentialGeneration: 1, isMain: false, + }], + }; + saveConfig(config); + saveCodexAccountCredential("pool-wham-restart", { + accessToken: chatgptPlanJwt("plus", "acct-pool-wham-restart"), + refreshToken: "refresh-pool-wham-restart", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "acct-pool-wham-restart", + }); + resetJwtPlanNotesForTests(); // simulate a fresh process: in-memory notes gone + const reloaded = loadConfig(); // startup path reads persisted config + reconcileCodexPlansFromTokens(reloaded); + expect(loadConfig().codexAccounts?.[0]?.plan).toBe("pro"); + }); +}); + diff --git a/tests/service-tier-capability.test.ts b/tests/service-tier-capability.test.ts index c781208ea8..2e397be86b 100644 --- a/tests/service-tier-capability.test.ts +++ b/tests/service-tier-capability.test.ts @@ -11,7 +11,7 @@ import { providerConfigSeed, enrichProviderFromRegistry } from "../src/providers import { getProviderRegistryEntry } from "../src/providers/registry"; import type { RequestLogContext } from "../src/server/request-log"; import { applyServiceTierGate, handleResponses } from "../src/server/responses/core"; -import { canForwardServiceTierForModel, supportsServiceTierForModel } from "../src/providers/service-tier"; +import { canForwardServiceTierForModel, serviceTierSupportForModel, supportsServiceTierForModel } from "../src/providers/service-tier"; import { serviceTierAdapterForModel } from "../src/providers/service-tier"; import { candidateCapabilityEvidence } from "../src/routing/capability"; import { resolveProductionBehaviorValues } from "../src/routing/compatibility/behavior"; @@ -311,3 +311,21 @@ describe("the gate fires on the live handleResponses path", () => { expect(undeclared).not.toHaveProperty("service_tier"); }); }); + +describe("unclassified chat-wire tier projection (release-audit fix)", () => { + test("unclassified openai-chat without chatServiceTier projects false (require.serviceTier unsupported keeps matching)", () => { + const provider = { adapter: "openai-chat" } as OcxProviderConfig; + expect(serviceTierSupportForModel(provider, "some-model")).toBe(false); + }); + + test("unclassified openai-chat WITH chatServiceTier: true keeps the historical unknown", () => { + const provider = { adapter: "openai-chat", chatServiceTier: true } as OcxProviderConfig; + expect(serviceTierSupportForModel(provider, "some-model")).toBeUndefined(); + }); + + test("unclassified Responses-wire provider stays unknown", () => { + const provider = { adapter: "openai-responses" } as OcxProviderConfig; + expect(serviceTierSupportForModel(provider, "some-model")).toBeUndefined(); + }); +}); +