diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 40078fa17c..e0b349c689 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -425,7 +425,7 @@ "open-sse/executors/codex.ts": 1505, "open-sse/executors/cursor.ts": 1759, "open-sse/executors/muse-spark-web.ts": 1405, - "open-sse/handlers/chatCore.ts": 6021, + "open-sse/handlers/chatCore.ts": 6026, "open-sse/handlers/imageGeneration.ts": 3259, "open-sse/handlers/search.ts": 1789, "open-sse/mcp-server/schemas/tools.ts": 1621, @@ -437,7 +437,7 @@ "open-sse/translator/response/openai-responses.ts": 1466, "open-sse/utils/cursorAgentProtobuf.ts": 1547, "open-sse/utils/proxyFetch.ts": 1271, - "open-sse/utils/stream.ts": 3080, + "open-sse/utils/stream.ts": 3098, "open-sse/vendor/codex-chatgpt-web/adapters/chatgpt-web/browser-worker.ts": 4398, "open-sse/vendor/codex-chatgpt-web/bridge.ts": 1335, "src/app/(dashboard)/dashboard/HomePageClient.tsx": 1344, @@ -655,5 +655,6 @@ "_rebaseline_2026_09_04_12641_continuation_effective_input": "PR #12641 crescimento proprio: src/sse/handlers/chat.ts 2450->2454 (+4). A continuacao por previous_response_id encadeava a partir de clientRawRequest.body.input, que e capturado ANTES da reconstrucao do proprio chat.ts; quando o turno anterior ja era uma continuacao, esse campo guarda so o delta do cliente, e o erro se acumulava a cada salto ate a reconstrucao virar itens de tool sem prefixo. Persistir o input EFETIVO exige as linhas no ponto onde a reconstrucao termina, dentro do fluxo de despacho. Coberto por tests/unit/responses-continuation-store.test.ts (22/22 focados na leva).", "_rebaseline_2026_09_05_12671_combos_usage_guide_external_store": "combos/page.tsx 5018 -> 5066: #12671 replaces the effect-based localStorage read with useSyncExternalStore; the +48 lines are the store helpers (subscribe/getSnapshot/getServerSnapshot/emit) hoisted to module scope, which is the sanctioned shape and what let the react-hooks/set-state-in-effect suppression be dropped.", "_rebaseline_2026_09_07_chatcore_nonstreaming_regression_fixes": "Own growth: open-sse/handlers/chatCore.ts 5984->6021 (+37). Two of my own PRs on top of #12867: #12963 pins the ok variant of the non-streaming leg result in its own binding (the discriminated-union narrowing was lost across the tool-loop reassignment, 13 TS2339 under tsconfig.typecheck-api.json), and #12990 restores four behaviours the same refactor dropped — abort classification through isLocalStreamLifecycleError, the omitted synthetic clientResponse, the claudePromptCacheLogMeta rebuild on the leg path, and the lazy fail-closed fence identity. Irreducible at the existing chokepoints: each edit sits where chatCore already owns the decision, and the helpers themselves (nonStreamingProviderLeg.ts, serverOwnedToolLoopWire.ts) are under cap. Covered by tests/unit/chatcore-translation-paths.test.ts (72/74; the 2 open are issue #13043).", - "_rebaseline_2026_09_07_virtualfactory_crosses_the_new_file_cap": "open-sse/services/autoCombo/virtualFactory.ts crosses the 1200 new-file cap for the first time (1187 on the pre-wave tip, 1219 after the wave). Growth is spread across the routing/free-tier wave, not one extractable block: #12794 feeds observed breaker state and model quality into snapshot scoring instead of neutral constants, #12792 adds the reliability factor the snapshot path was still ignoring and the pooled-latency bootstrap, and #12744 tightens the free-model predicate the factory consumes, and #12795 records which filter stage emptied an auto/* pool. FROZEN RATHER THAN SPLIT, deliberately, and this is debt: two cohesive extraction candidates are ready when someone owns the move — computeSnapshotWeights (~85 lines) and the credential-eligibility group hasUsableOAuthToken/hasProviderSpecificSessionData/isKeylessEligibleConnection/hasUsableConnectionCredential (~70 lines). Either alone clears 1200 from here. Splitting three contributors' just-merged work mid-batch was the larger risk." + "_rebaseline_2026_09_07_virtualfactory_crosses_the_new_file_cap": "open-sse/services/autoCombo/virtualFactory.ts crosses the 1200 new-file cap for the first time (1187 on the pre-wave tip, 1219 after the wave). Growth is spread across the routing/free-tier wave, not one extractable block: #12794 feeds observed breaker state and model quality into snapshot scoring instead of neutral constants, #12792 adds the reliability factor the snapshot path was still ignoring and the pooled-latency bootstrap, and #12744 tightens the free-model predicate the factory consumes, and #12795 records which filter stage emptied an auto/* pool. FROZEN RATHER THAN SPLIT, deliberately, and this is debt: two cohesive extraction candidates are ready when someone owns the move — computeSnapshotWeights (~85 lines) and the credential-eligibility group hasUsableOAuthToken/hasProviderSpecificSessionData/isKeylessEligibleConnection/hasUsableConnectionCredential (~70 lines). Either alone clears 1200 from here. Splitting three contributors' just-merged work mid-batch was the larger risk.", + "_rebaseline_2026_09_07_streaming_wave": "Stacked growth from the SSE/streaming wave. open-sse/handlers/chatCore.ts 6021->6026 (+5): #12854 seeds the in-memory pending continuation state synchronously, before saveCallLogOperation's first await, closing the window where resolvePreviousResponseState finds nothing because the artifact write has not landed yet. open-sse/utils/stream.ts 3080->3098 (+18): #12828 emits the trailing usage-estimate chunk on the translate flush (#12151 had only covered passthrough, so translate-mode clients never saw token counts) and #12718 stops rebuilding a truncated summary from the collector's cap-dropped event array. Irreducible at the existing chokepoints — both are the flush/finalization points themselves. Covered by the continuation-store, translate-usage and collector-truncation suites." } diff --git a/src/lib/db/responsesContinuationStore.ts b/src/lib/db/responsesContinuationStore.ts index 90d96e8f6b..ef2ae5224f 100644 --- a/src/lib/db/responsesContinuationStore.ts +++ b/src/lib/db/responsesContinuationStore.ts @@ -16,6 +16,19 @@ * lightweight `call_logs.response_id` index (154_call_logs_response_id.sql) * is new. Every lookup is scoped by `api_key_id` -- one client can never * resolve another client's stored conversation. + * + * Pending-write bridge: `saveCallLog` writes the durable `call_logs` row + * (and the artifact this module reads) through a queued, single-worker disk + * write (see callLogArtifactWriter.ts) that runs *after* the response has + * already been streamed to the client. A client that fires its next turn + * immediately -- normal in a tight tool-calling loop -- can reach this + * module before that write lands, and would otherwise see a false "not + * found" for a response id OmniRoute itself minted moments earlier. + * `seedPendingContinuationState` / `clearPendingContinuationState` (called + * from callLogs.ts, synchronously around that same write) bridge exactly + * that window, applying the identical extraction/fail-closed rules via + * extractContinuationState below; the durable row remains the sole source + * of truth once it exists. */ import { getDbInstance } from "./core"; @@ -47,38 +60,23 @@ function containsTruncatedArrayMarker(items: readonly unknown[]): boolean { return items.some((item) => isPlainRecord(item) && item[TRUNCATED_ARRAY_MARKER] === true); } +export type ContinuationPipeline = { + clientRawRequest?: { body?: unknown; effectiveInput?: unknown }; + clientResponse?: { output?: unknown; summary?: { output?: unknown }; _truncated?: unknown }; +}; + /** - * Resolve the full input + output a prior Responses API call produced, so - * the caller can reconstruct `full_input = stored.input + stored.output + - * new_delta`. Returns null on any lookup/read/shape failure (unknown id, - * wrong tenant, artifact missing, or an artifact whose pipeline payload was - * size-limit-omitted -- see MAX_CALL_LOG_ARTIFACT_BYTES in - * callLogArtifacts.ts) so the caller can fail closed and ask the client to - * resend full history, exactly like a real `previous_response_not_found` - * from OpenAI itself. + * Reconstruct { input, output } from one call's own pipeline payload, or + * fail closed to null -- shared by the durable (call_logs + artifact) path + * below and the in-memory pending bridge, so a fail-closed rule added here + * (video redaction, truncation markers, an empty/aborted response) protects + * both instead of only whichever path someone remembered to update. */ -export function resolvePreviousResponseState( - responseId: string, - apiKeyId: string | null | undefined +function extractContinuationState( + pipeline: ContinuationPipeline | null | undefined, + videoContentRemoved: boolean ): ResponsesContinuationState | null { - if (!responseId) return null; - - const db = getDbInstance(); - const row = db - .prepare( - `SELECT artifact_relpath, api_key_id, video_content_removed FROM call_logs - WHERE response_id = ? AND detail_state = 'ready' - ORDER BY timestamp DESC LIMIT 1` - ) - .get(responseId) as - | { artifact_relpath: string | null; api_key_id: string | null; video_content_removed: number } - | undefined; - - if (!row || !row.artifact_relpath) return null; - // Tenant isolation: a response id is only ever handed back to the API key - // that created it. A stored row with no api_key_id at all (no-log/legacy) - // can never be resolved by any key -- fail closed rather than guess. - if (!apiKeyId || row.api_key_id !== apiKeyId) return null; + if (!pipeline) return null; // #12150 P2 surface 2: the persisted clientRawRequest snapshot on this row had // its video transcript cues structurally redacted to [redacted-video-transcript] // before storage (videoBridgeSnapshotRedaction, marker written by the call-log @@ -86,15 +84,10 @@ export function resolvePreviousResponseState( // text -- reconstructing a continuation off it would forward the placeholder // upstream as if it were genuine history. Fail closed so the client resends // full history, exactly like a real previous_response_not_found. - if (row.video_content_removed === 1) return null; + if (videoContentRemoved) return null; - const { artifact, state } = readCallArtifact(row.artifact_relpath); - if (state !== "ready" || !artifact?.pipeline) return null; - - const clientRawRequest = artifact.pipeline.clientRawRequest as - { body?: unknown; effectiveInput?: unknown } | undefined; - const clientResponse = artifact.pipeline.clientResponse as - { output?: unknown; summary?: { output?: unknown } } | undefined; + const clientRawRequest = pipeline.clientRawRequest; + const clientResponse = pipeline.clientResponse; // clientRawRequest, not providerRequest: this store only ever fires for // sourceFormat === OPENAI_RESPONSES (see chat.ts), so the client's own @@ -156,6 +149,113 @@ export function resolvePreviousResponseState( return { input, output }; } +type PendingContinuationEntry = { + apiKeyId: string | null; + state: ResponsesContinuationState; + expiresAt: number; +}; + +// Bounds how long a seeded entry can stand in for the durable row. Comfortably +// longer than realistic artifact-write queue latency (single worker, see +// MAX_QUEUED_JOBS in callLogArtifactWriter.ts) but short enough that a save +// which never reaches "ready" (e.g. detailState becomes "missing") falls back +// to the same permanent not-found the durable path already gives today. +const PENDING_CONTINUATION_TTL_MS = 60_000; +const pendingContinuationStates = new Map(); + +/** + * Called from callLogs.ts the moment a response id and its pipeline payload + * are known, before the artifact write is even queued. Runs the response + * through the same extractContinuationState fail-closed rules as the durable + * path -- a video-redacted, truncated, or empty-output response never gets + * bridged, matching what the durable row would (eventually) say anyway. + */ +export function seedPendingContinuationState( + responseId: string, + apiKeyId: string | null, + pipeline: ContinuationPipeline | null | undefined, + videoContentRemoved: boolean +): void { + if (!responseId) return; + const state = extractContinuationState(pipeline, videoContentRemoved); + if (!state) return; + pendingContinuationStates.set(responseId, { + apiKeyId, + state, + expiresAt: Date.now() + PENDING_CONTINUATION_TTL_MS, + }); +} + +/** Called from callLogs.ts once the durable row lands -- the DB is now + * authoritative and the bridge entry would otherwise just idle until its TTL. */ +export function clearPendingContinuationState(responseId: string): void { + if (!responseId) return; + pendingContinuationStates.delete(responseId); +} + +function resolvePendingContinuationState( + responseId: string, + apiKeyId: string | null | undefined +): ResponsesContinuationState | null { + const entry = pendingContinuationStates.get(responseId); + if (!entry) return null; + if (Date.now() > entry.expiresAt) { + pendingContinuationStates.delete(responseId); + return null; + } + // Same tenant-isolation rule as the durable lookup below: a response id is + // only ever handed back to the API key that created it. + if (!apiKeyId || entry.apiKeyId !== apiKeyId) return null; + return entry.state; +} + +/** + * Resolve the full input + output a prior Responses API call produced, so + * the caller can reconstruct `full_input = stored.input + stored.output + + * new_delta`. Returns null on any lookup/read/shape failure (unknown id, + * wrong tenant, artifact missing, or an artifact whose pipeline payload was + * size-limit-omitted -- see MAX_CALL_LOG_ARTIFACT_BYTES in + * callLogArtifacts.ts) so the caller can fail closed and ask the client to + * resend full history, exactly like a real `previous_response_not_found` + * from OpenAI itself. + */ +export function resolvePreviousResponseState( + responseId: string, + apiKeyId: string | null | undefined +): ResponsesContinuationState | null { + if (!responseId) return null; + + const db = getDbInstance(); + const row = db + .prepare( + `SELECT artifact_relpath, api_key_id, video_content_removed FROM call_logs + WHERE response_id = ? AND detail_state = 'ready' + ORDER BY timestamp DESC LIMIT 1` + ) + .get(responseId) as + | { artifact_relpath: string | null; api_key_id: string | null; video_content_removed: number } + | undefined; + + if (!row) { + // No durable row at all yet -- this is exactly the write-in-flight + // window the pending bridge exists for, not a genuinely unknown id. + return resolvePendingContinuationState(responseId, apiKeyId); + } + if (!row.artifact_relpath) return null; + // Tenant isolation: a response id is only ever handed back to the API key + // that created it. A stored row with no api_key_id at all (no-log/legacy) + // can never be resolved by any key -- fail closed rather than guess. + if (!apiKeyId || row.api_key_id !== apiKeyId) return null; + + const { artifact, state } = readCallArtifact(row.artifact_relpath); + if (state !== "ready" || !artifact?.pipeline) return null; + + return extractContinuationState( + artifact.pipeline as ContinuationPipeline, + row.video_content_removed === 1 + ); +} + /** * Resolve the call-log id that produced `responseId`, for the dashboard's * "continues from" link. Reuses the same `call_logs.response_id` index and diff --git a/src/lib/usage/callLogs.ts b/src/lib/usage/callLogs.ts index 3e1ed314de..7a2181b0dd 100644 --- a/src/lib/usage/callLogs.ts +++ b/src/lib/usage/callLogs.ts @@ -13,6 +13,11 @@ import { getDbInstance } from "../db/core"; import { getRequestDetailLogByCallLogId } from "../db/detailedLogs"; import { shouldPersistToDisk } from "./migrations"; import { getCallLogApiKeyContext } from "./callLogApiKeyContext"; +import { + seedPendingContinuationState, + clearPendingContinuationState, + type ContinuationPipeline, +} from "../db/responsesContinuationStore"; import { getLoggedInputTokens, getLoggedOutputTokens, @@ -470,6 +475,26 @@ async function saveCallLogOperation(entry: any): Promise { ); const protectedError = sanitizeErrorForLog(entry.error); + // Bridges the window before this row's own artifact write (queued below, + // async) lands with detail_state = 'ready': a client that fires its next + // turn immediately -- normal in a tight tool-calling loop -- can reach + // resolvePreviousResponseState before that write exists at all. Seeded + // synchronously, before any await, from the same protected pipeline + // payload the artifact will eventually hold (and the same + // videoContentRemoved/fail-closed rules), so it is available the instant + // this function is called. See responsesContinuationStore.ts. + if (typeof entry.responseId === "string" && entry.responseId.length > 0) { + seedPendingContinuationState( + entry.responseId, + apiKeyId, + // The store's own contract, not a looser restatement of it: the inline shape + // widened both fields to `unknown`, which does not assign to + // ContinuationPipeline's typed members (TS2345 under typecheck:core). + protectedPipelinePayloads as ContinuationPipeline | null, + Boolean(entry.videoContentRemoved) + ); + } + const account = await resolveAccountName(entry.connectionId || null); const rawProvider: string = entry.provider || "-"; const rawRequestedModel: string | null = entry.requestedModel || null; @@ -604,6 +629,12 @@ async function saveCallLogOperation(entry: any): Promise { requestSummary, }); + if (detailState === "ready" && typeof logEntry.responseId === "string") { + // The durable row is now authoritative; drop the bridge entry instead + // of letting it idle until its TTL. + clearPendingContinuationState(logEntry.responseId); + } + scheduleCallLogRotation(); } catch (error) { console.error( diff --git a/tests/unit/responses-continuation-store.test.ts b/tests/unit/responses-continuation-store.test.ts index 6d8cb66c5f..cfddd18556 100644 --- a/tests/unit/responses-continuation-store.test.ts +++ b/tests/unit/responses-continuation-store.test.ts @@ -15,6 +15,7 @@ process.env.DATA_DIR = TEST_DATA_DIR; const core = await import("../../src/lib/db/core.ts"); const store = await import("../../src/lib/db/responsesContinuationStore.ts"); +const callLogs = await import("../../src/lib/usage/callLogs.ts"); test.after(() => { core.resetDbInstance(); @@ -471,3 +472,133 @@ test("resolvePreviousResponseState returns null when detail logging was never ca assert.equal(store.resolvePreviousResponseState("resp_no_detail", "key-1"), null); }); + +// Proven live in production (2026-09-06, nvidia/nemotron-3.5-lightning:free via +// OpenRouter): a client that fires its next turn immediately after receiving a +// response id -- normal behavior in a tight tool-calling loop -- can reach +// resolvePreviousResponseState before saveCallLog's own artifact write (queued, +// see writeCallArtifactAsync) has landed and flipped detail_state to "ready". +// Before the pending-continuation bridge, OmniRoute answered a well-formed 400 +// previous_response_not_found for an id it minted seconds earlier; the wire +// capture showed the client recovering by resending full history, exactly like +// a real OpenAI-issued rejection -- but every one of those resends was an +// avoidable full-history resend, not a genuine unknown id. This exercises the +// real saveCallLog pipeline end to end, not a pre-inserted "ready" row. +test("resolvePreviousResponseState resolves via the pending bridge while saveCallLog's artifact write is still queued", async () => { + const save = callLogs.saveCallLog({ + method: "POST", + path: "/v1/responses", + status: 200, + model: "nvidia/nemotron-3.5-lightning:free", + provider: "openrouter", + apiKeyId: "key-1", + duration: 8169, + responseId: "resp_gen-race-abc123", + requestBody: { input: [{ type: "message", role: "user", content: "hi" }], store: true }, + responseBody: { id: "resp_gen-race-abc123" }, + pipeline: { + clientRawRequest: { body: { input: [{ type: "message", role: "user", content: "hi" }] } }, + clientResponse: { + id: "resp_gen-race-abc123", + output: [{ type: "message", role: "assistant", content: "hello" }], + }, + }, + }); + + // The client's next turn can arrive before the queued artifact write below + // has even started -- the bridge, seeded synchronously inside saveCallLog + // before this call returns, must already answer correctly. + assert.deepEqual(store.resolvePreviousResponseState("resp_gen-race-abc123", "key-1"), { + input: [{ type: "message", role: "user", content: "hi" }], + output: [{ type: "message", role: "assistant", content: "hello" }], + }); + + await save; + + // Once the durable row lands, the same id must still resolve -- now from + // call_logs/the artifact, with the bridge entry already cleared. + assert.deepEqual(store.resolvePreviousResponseState("resp_gen-race-abc123", "key-1"), { + input: [{ type: "message", role: "user", content: "hi" }], + output: [{ type: "message", role: "assistant", content: "hello" }], + }); +}); + +test("resolvePreviousResponseState never lets the pending bridge cross tenants", async () => { + const save = callLogs.saveCallLog({ + method: "POST", + path: "/v1/responses", + status: 200, + model: "nvidia/nemotron-3.5-lightning:free", + provider: "openrouter", + apiKeyId: "key-a", + duration: 4000, + responseId: "resp_gen-tenant-bridge", + pipeline: { + clientRawRequest: { body: { input: [{ role: "user", content: "secret" }] } }, + clientResponse: { id: "resp_gen-tenant-bridge", output: [{ role: "assistant", content: "reply" }] }, + }, + }); + + assert.equal(store.resolvePreviousResponseState("resp_gen-tenant-bridge", "key-b"), null); + assert.equal(store.resolvePreviousResponseState("resp_gen-tenant-bridge", null), null); + assert.notEqual(store.resolvePreviousResponseState("resp_gen-tenant-bridge", "key-a"), null); + + await save; +}); + +test("resolvePreviousResponseState does not bridge a response id that saveCallLog never seeded (no-log or no pipeline)", async () => { + // noLog: the entry is redacted before it would ever reach the bridge. + await callLogs.saveCallLog({ + method: "POST", + path: "/v1/responses", + status: 200, + model: "gpt-5.4-pro", + provider: "openai", + apiKeyId: "key-1", + noLog: true, + responseId: "resp_gen-nolog", + pipeline: { + clientRawRequest: { body: { input: [{ role: "user", content: "hi" }] } }, + clientResponse: { id: "resp_gen-nolog", output: [{ role: "assistant", content: "hi" }] }, + }, + }); + assert.equal(store.resolvePreviousResponseState("resp_gen-nolog", "key-1"), null); + + // No pipeline payload at all -- nothing to reconstruct from. + await callLogs.saveCallLog({ + method: "POST", + path: "/v1/responses", + status: 200, + model: "gpt-5.4-pro", + provider: "openai", + apiKeyId: "key-1", + responseId: "resp_gen-no-pipeline", + }); + assert.equal(store.resolvePreviousResponseState("resp_gen-no-pipeline", "key-1"), null); +}); + +test("the pending bridge shares the durable path's fail-closed rules (video-redacted turns never bridge)", async () => { + const save = callLogs.saveCallLog({ + method: "POST", + path: "/v1/responses", + status: 200, + model: "gpt-5.4-pro", + provider: "openai", + apiKeyId: "key-1", + duration: 100, + responseId: "resp_gen-video-bridge", + videoContentRemoved: true, + pipeline: { + clientRawRequest: { body: { input: [{ role: "user", content: "[redacted-video-transcript]" }] } }, + clientResponse: { id: "resp_gen-video-bridge", output: [{ role: "assistant", content: "ok" }] }, + }, + }); + + // Even mid-flight (bridge-only, durable row not yet written), a + // video-redacted turn must fail closed exactly like the durable path does. + assert.equal(store.resolvePreviousResponseState("resp_gen-video-bridge", "key-1"), null); + + await save; + + assert.equal(store.resolvePreviousResponseState("resp_gen-video-bridge", "key-1"), null); +});