diff --git a/src/adapters/google-truncation.ts b/src/adapters/google-truncation.ts index 208291796..5ac6de1cd 100644 --- a/src/adapters/google-truncation.ts +++ b/src/adapters/google-truncation.ts @@ -11,3 +11,14 @@ export function vertexTruncationErrorMessage(reason?: string): string { const suffix = reason ? ` (${redactSecretString(reason).slice(0, 160)})` : ""; return `Vertex AI response truncated upstream before the turn completed${suffix}`; } + +/** + * Whether a finished turn must fail closed. A truncation reason arriving mid tool call always + * does. MALFORMED_FUNCTION_CALL fails closed even with zero started calls: the malformed call + * is dropped upstream and usually never materializes as a part, so the turn is incomplete + * despite looking empty. MAX_TOKENS with no started call stays a plain token-limit stop. + */ +export function isVertexTruncatedTurn(finishReason: string | undefined, toolCallsStarted: number): boolean { + if (!isVertexTruncationReason(finishReason)) return false; + return toolCallsStarted > 0 || finishReason === "MALFORMED_FUNCTION_CALL"; +} diff --git a/src/adapters/google.ts b/src/adapters/google.ts index c9d7aedfc..58374ddcd 100644 --- a/src/adapters/google.ts +++ b/src/adapters/google.ts @@ -17,7 +17,7 @@ import { contentPartsToText, parseDataUrl } from "./image"; import { getVertexAccessToken } from "../lib/gcp-adc"; import { fetchAntigravityWithRetry, fetchVertexWithRetry } from "./google-http"; import { safeAntigravityHttpErrorMessage, safeVertexHttpErrorMessage } from "./google-errors"; -import { isVertexTruncationReason, vertexTruncationErrorMessage } from "./google-truncation"; +import { isVertexTruncatedTurn, vertexTruncationErrorMessage } from "./google-truncation"; import { ANTIGRAVITY_REQUEST_UA, antigravitySessionId, isLikelyRealThoughtSignature, sanitizeAntigravityClaudeSignatures } from "./google-antigravity-wire"; import { compileGoogleWireBody } from "./google-wire-compiler"; import { identifyRoutedModel } from "./identity"; @@ -596,7 +596,7 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte // Fail-closed: a turn cut off mid tool call (MAX_TOKENS / MALFORMED_FUNCTION_CALL) surfaces // an error instead of a silently-incomplete done. Mirrors kiro-truncation. if ((provider.googleMode === "vertex" || provider.googleMode === "cloud-code-assist") - && toolCallsStarted > 0 && isVertexTruncationReason(lastFinishReason)) { + && isVertexTruncatedTurn(lastFinishReason, toolCallsStarted)) { yield { type: "error", message: vertexTruncationErrorMessage(lastFinishReason) }; return; } @@ -752,7 +752,7 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte // Fail-closed truncation, same as the stream path: a non-stream turn cut off mid tool call // (MAX_TOKENS / MALFORMED_FUNCTION_CALL) surfaces an error instead of a silent done. if ((provider.googleMode === "vertex" || provider.googleMode === "cloud-code-assist") - && toolCallsStarted > 0 && isVertexTruncationReason(candidates?.[0]?.finishReason)) { + && isVertexTruncatedTurn(candidates?.[0]?.finishReason, toolCallsStarted)) { return finish([{ type: "error", message: vertexTruncationErrorMessage(candidates?.[0]?.finishReason) }]); } diff --git a/src/adapters/kiro-events.ts b/src/adapters/kiro-events.ts index 0def314cc..eab61ea51 100644 --- a/src/adapters/kiro-events.ts +++ b/src/adapters/kiro-events.ts @@ -3,7 +3,8 @@ import { kiroTruncationReason } from "./kiro-truncation"; export type ParsedKiroEvent = | { type: "content"; data?: string; modelId?: string } - | { type: "reasoning"; data?: string } + | { type: "reasoning"; data?: string; redactedContent?: string } + | { type: "context_usage"; contextUsagePercentage: number } | { type: "tool"; name?: string; toolUseId?: string; input?: string; stop?: boolean } | { type: "truncation"; data: string } | { type: "metadata"; usage?: OcxUsage; contextUsagePercentage?: number; stopReason?: string } @@ -17,6 +18,10 @@ const KNOWN_EVENT_TYPES = new Set([ "toolUseEvent", "messageMetadataEvent", "metadataEvent", + // Authoritative context pressure. Every capture (kiro-cli 2.14.1 and 2.16.0) put the percentage + // HERE and left `metadataEvent` carrying only `stopReason`; metadataEvent's own + // contextUsagePercentage stays supported as a fallback rather than being dropped. + "contextUsageEvent", "invalidStateEvent", "error", ]); @@ -114,11 +119,17 @@ export function parseKiroEvent(eventType: string, payload: Uint8Array): ParsedKi : {}), }; case "reasoningContentEvent": + // `text` is plaintext reasoning; `redactedContent` is the encrypted blob the GPT-5.6 family + // (sol/terra/luna) actually returns — they never send `text`. Keyed off the wire field, not + // the model id. Both may be absent on a bare event. return { type: "reasoning", ...(optionalString(eventType, parsed, "text") !== undefined ? { data: optionalString(eventType, parsed, "text") } : {}), + ...(optionalString(eventType, parsed, "redactedContent") !== undefined + ? { redactedContent: optionalString(eventType, parsed, "redactedContent") } + : {}), }; case "toolUseEvent": return { @@ -161,6 +172,13 @@ export function parseKiroEvent(eventType: string, payload: Uint8Array): ParsedKi ...(stopReason !== undefined ? { stopReason } : {}), }; } + case "contextUsageEvent": { + const contextUsagePercentage = parsed.contextUsagePercentage; + if (typeof contextUsagePercentage !== "number" || !Number.isFinite(contextUsagePercentage)) { + return malformed(eventType, "contextUsagePercentage must be a finite number"); + } + return { type: "context_usage", contextUsagePercentage }; + } case "invalidStateEvent": return { type: "invalid_state", message: optionalString(eventType, parsed, "message") }; case "error": diff --git a/src/adapters/kiro.ts b/src/adapters/kiro.ts index 3e2440045..2ea1bc84d 100644 --- a/src/adapters/kiro.ts +++ b/src/adapters/kiro.ts @@ -99,7 +99,11 @@ interface KiroUserInputMessage { } interface KiroHistoryEntry { userInputMessage?: KiroUserInputMessage; - assistantResponseMessage?: { content: string; toolUses?: KiroToolUse[] }; + assistantResponseMessage?: { + content: string; + toolUses?: KiroToolUse[]; + reasoningContent?: { redactedContent: string }; + }; } function kiroToolWireNames(tools: readonly unknown[]): string[] { @@ -326,7 +330,7 @@ function validateKiroCapabilities(parsed: OcxParsedRequest): void { type KiroTurn = | { kind: "user"; content: string; images: KiroImage[]; toolResults: KiroToolResult[] } - | { kind: "assistant"; content: string; toolUses: KiroToolUse[] }; + | { kind: "assistant"; content: string; toolUses: KiroToolUse[]; redactedReasoning?: string }; function appendTurnText(target: string, next: string): string { if (!next) return target; @@ -471,13 +475,15 @@ export function buildKiroPayload( turns.push({ kind: "user", content, images: [...images], toolResults: [...toolResults] }); } }; - const pushAssistant = (content: string, toolUses: KiroToolUse[]): void => { + const pushAssistant = (content: string, toolUses: KiroToolUse[], redactedReasoning?: string): void => { const last = turns.at(-1); if (last?.kind === "assistant") { last.content = appendTurnText(last.content, content); last.toolUses.push(...toolUses); + // Merged turns keep the newest blob: it covers the reasoning up to the merged turn's end. + if (redactedReasoning) last.redactedReasoning = redactedReasoning; } else { - turns.push({ kind: "assistant", content, toolUses: [...toolUses] }); + turns.push({ kind: "assistant", content, toolUses: [...toolUses], ...(redactedReasoning ? { redactedReasoning } : {}) }); } }; @@ -507,7 +513,7 @@ export function buildKiroPayload( const hasReasoning = aMsg.content.some(part => part.type === "thinking" && part.thinking.trim()); if (hasReasoning) continue; } - pushAssistant(text, toolUses); + pushAssistant(text, toolUses, aMsg.kiroRedactedReasoning); } else if (msg.role === "toolResult") { const tr = msg as OcxToolResultMessage; if (tr.containsEncryptedContent) { @@ -561,6 +567,7 @@ export function buildKiroPayload( assistantResponseMessage: { content: turn.content, ...(turn.toolUses.length > 0 ? { toolUses: turn.toolUses } : {}), + ...(turn.redactedReasoning ? { reasoningContent: { redactedContent: turn.redactedReasoning } } : {}), }, } : { @@ -1183,6 +1190,12 @@ async function* parseKiroAttemptEvents( if (ev.data) { yield* emitRetained(stage({ type: "reasoning_raw_delta", text: ev.data })); } + if (ev.redactedContent) { + yield* emitRetained(stage({ type: "kiro_redacted_reasoning", data: ev.redactedContent })); + } + break; + case "context_usage": + if (ev.contextUsagePercentage > 0) contextUsagePercentage = ev.contextUsagePercentage; break; case "tool": { for (const contentEvent of thinking.flush()) { diff --git a/src/bridge.ts b/src/bridge.ts index 21fa85118..cc121e2e5 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -442,6 +442,29 @@ export function bridgeToResponsesSSE( retainFinishedItem(item as OutputItem, bytesOf(encrypted), "reasoning"); outputIndex++; }; + // Kiro reasoning round-trip. Kiro sends its encrypted blob at the END of a turn, while the + // assistant message is still open, so this CANNOT emit on arrival: the open message still + // owns `outputIndex` (it only advances on close), and an item emitted here would both reuse + // that index and land BEFORE the message — where the parser's backwards pairing drops it as + // orphaned. Stash it and flush after `done` has closed every open item instead. + let pendingKiroRedacted: string | undefined; + let pendingKiroRedactedBytes = 0; + const flushKiroRedactedReasoning = () => { + if (!pendingKiroRedacted) return; + const previousBytes = pendingKiroRedactedBytes; + const encrypted = encodeReasoningEnvelope({ krc: pendingKiroRedacted }); + const reservation = budget?.reserveTransient(bytesOf(encrypted), { kind: "reasoning" }); + pendingKiroRedacted = undefined; + pendingKiroRedactedBytes = 0; + reservation?.commitRetained(); + budget?.releaseRetained(previousBytes, { kind: "reasoning" }); + const itemId = `rs_${uuid()}`; + const item = { type: "reasoning", id: itemId, summary: [] as never[], encrypted_content: encrypted }; + emit("response.output_item.added", { output_index: outputIndex, item }); + emit("response.output_item.done", { output_index: outputIndex, item }); + retainFinishedItem(item as OutputItem, bytesOf(encrypted), "reasoning"); + outputIndex++; + }; // Full assistant text of a compaction turn (across message boundaries) — becomes the // synthetic compaction item's payload on done. let compactionText = ""; @@ -869,6 +892,12 @@ export function bridgeToResponsesSSE( pendingRedacted.push(event.data); break; } + case "kiro_redacted_reasoning": { + // Stash only — see flushKiroRedactedReasoning. One blob per turn, so last wins. + pendingKiroRedactedBytes = replaceRetainedString(pendingKiroRedactedBytes, event.data, "reasoning"); + pendingKiroRedacted = event.data; + break; + } case "reasoning_raw_delta": { if (options?.hideThinkingSummary) { ({ value: hiddenRawReasoningText, bytes: hiddenRawReasoningBytes } = appendString( @@ -1039,6 +1068,9 @@ export function bridgeToResponsesSSE( // Redacted-only turns (or hidden thinking without a trailing signature event) still // need their envelope-only reasoning item so the blocks replay next turn. flushHiddenReasoningEnvelope(); + // After every close above, so the blob lands AFTER the assistant message it belongs + // to and the parser's backwards pairing finds it. + flushKiroRedactedReasoning(); if (options?.compaction) { // Exactly one compaction item per turn; codex-rs takes the first and fatals on 0. const item = { @@ -1361,6 +1393,10 @@ function buildResponseJSONWithBudget( let batchSignatureBytes = 0; let batchRedacted: string[] = []; let batchRedactedBytes = 0; + // Kiro reasoning blob, held until after the trailing flushes so it lands AFTER the assistant + // message (see the streaming path). Retained because it outlives releaseTranslatedEvent. + let batchKiroRedacted: string | undefined; + let batchKiroRedactedBytes = 0; let currentToolCallId = ""; let currentToolCallName = ""; let currentToolCallArgs = ""; @@ -1528,6 +1564,16 @@ function buildResponseJSONWithBudget( } batchRedacted.push(e.data); break; + case "kiro_redacted_reasoning": + // Stash only — pushed after the trailing flushes. One blob per turn, so last wins. + { + const dataBytes = bytesOf(e.data); + budget?.chargeRetained(dataBytes, { kind: "reasoning" }); + if (batchKiroRedactedBytes > 0) budget?.releaseRetained(batchKiroRedactedBytes, { kind: "reasoning" }); + batchKiroRedactedBytes = dataBytes; + } + batchKiroRedacted = e.data; + break; case "reasoning_raw_delta": if (currentText) flushText("commentary"); if (currentSummaryReasoning) flushSummaryReasoning(); @@ -1626,6 +1672,15 @@ function buildResponseJSONWithBudget( flushRawReasoning(); // Open tool call on a failed/incomplete turn must not land as status:"completed". if (currentToolCallId) flushToolCall(errorEvent || incompleteEvent ? "incomplete" : "completed"); + if (batchKiroRedacted) { + // pushOutput reserves the item itself and releases the retained raw blob it replaces. + pushOutput({ + type: "reasoning", id: `rs_${uuid()}`, summary: [], + encrypted_content: encodeReasoningEnvelope({ krc: batchKiroRedacted }), + }, batchKiroRedactedBytes, "reasoning"); + batchKiroRedacted = undefined; + batchKiroRedactedBytes = 0; + } // A truncated turn must never be installed as replacement history: emit the // compaction item only when the turn actually completed (#422). if ( diff --git a/src/claude/inbound.ts b/src/claude/inbound.ts index d27169c51..86cdfc104 100644 --- a/src/claude/inbound.ts +++ b/src/claude/inbound.ts @@ -106,6 +106,10 @@ function toolResultOutput(block: Rec): string | Rec[] { } else if (item.type === "image") { const img = imageBlockToInputImage(item); if (img) out.push(img); + } else if (item.type === "document") { + // Same marker as the user-message document case below: the model should see the + // attachment happened instead of an empty tool output. + out.push({ type: "input_text", text: `[document${typeof item.title === "string" ? `: ${item.title}` : ""}]` }); } } if (isError) out.unshift({ type: "input_text", text: "[tool error]" }); diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index 0a9c81b2a..e025762f8 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -187,6 +187,8 @@ interface OpenBlock { argsBufBytes?: number; webSearchArgsEmitted?: boolean; callId?: string; + /** Last reasoning part identity (item + summary/content index) seen by this thinking block. */ + reasoningPartKey?: string; } /** Streaming: Responses SSE bytes -> Anthropic Messages SSE bytes. */ @@ -353,6 +355,21 @@ export function responsesSseToAnthropicSse( case "response.reasoning_text.delta": { if (typeof data.delta !== "string" || data.delta.length === 0) break; ensureBlock("thinking"); + // The JSON path joins reasoning summary/content parts with "\n\n" + // (responsesJsonToAnthropicMessage); mirror that at part and item boundaries + // so multi-part summaries do not glue into one run-on paragraph. Frames + // without part indices produce a constant key and never get a separator. + const slot = eventName === "response.reasoning_summary_text.delta" + ? `s${String(data.summary_index)}` + : `c${String(data.content_index)}`; + const partKey = `${String(data.item_id)}:${slot}`; + if (open!.reasoningPartKey !== undefined && open!.reasoningPartKey !== partKey) { + emit("content_block_delta", { + type: "content_block_delta", index: open!.index, + delta: { type: "thinking_delta", thinking: "\n\n" }, + }); + } + open!.reasoningPartKey = partKey; emit("content_block_delta", { type: "content_block_delta", index: open!.index, delta: { type: "thinking_delta", thinking: data.delta }, diff --git a/src/lib/bun-stream-caps.ts b/src/lib/bun-stream-caps.ts index 188f10b51..3b5265e70 100644 --- a/src/lib/bun-stream-caps.ts +++ b/src/lib/bun-stream-caps.ts @@ -6,7 +6,8 @@ * PR #32120, merged 2026-06-21). No RELEASED Bun version is proven to carry * that fix yet, so `MIN_FIXED_BUN_VERSION` is null: every runtime is * "known-bad" until a bundle-bump commit sets it. Windows no-rewrite traffic - * follows this runtime/config decision. Darwin no-rewrite traffic stays on tee + * follows this runtime/config decision, preserving the explicit legacy-tee + * safety pin. Darwin no-rewrite traffic stays on tee * for `auto` regardless of runtime capability and reaches eager relay only via * explicit `streamMode: "eager-relay"` opt-in (see * devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md). diff --git a/src/responses/parser.ts b/src/responses/parser.ts index 339fd0002..cdfb1ff33 100644 --- a/src/responses/parser.ts +++ b/src/responses/parser.ts @@ -416,6 +416,18 @@ export function parseRequest(body: unknown): OcxParsedRequest { : null; const thinkingText = envelope?.txt || text; + // Kiro reasoning round-trip: a krc-only item carries nothing renderable — it is provider + // state for the assistant turn that ALREADY closed, because Kiro emits its + // reasoningContentEvent at the END of a turn (after content AND tool calls, verified + // against kiro-cli 2.14.1/2.16.0). Folding it into the FOLLOWING turn like ordinary + // reasoning would attach turn N's blob to turn N+1, so attach it backwards instead. With + // no assistant turn to own it the blob is dropped rather than mis-paired. + if (envelope?.krc && thinkingText.length === 0) { + const previous = messages[messages.length - 1]; + if (previous?.role === "assistant") previous.kiroRedactedReasoning = envelope.krc; + continue; + } + // Native/non-ocxr1 encrypted-only reasoning is opaque here. Do not create a detached // assistant turn or invent replayable plaintext/signatures from the encrypted payload. if (thinkingText.length > 0) { diff --git a/src/responses/reasoning-envelope.ts b/src/responses/reasoning-envelope.ts index 18c6bc639..1735f775f 100644 --- a/src/responses/reasoning-envelope.ts +++ b/src/responses/reasoning-envelope.ts @@ -24,6 +24,12 @@ export interface ReasoningEnvelope { * so replay needs it even though the visible summary was suppressed. */ txt?: string; + /** + * Kiro `reasoningContentEvent.redactedContent`: a KMS-encrypted reasoning blob that is opaque to + * the proxy. Kiro's own CLI replays it on the matching `assistantResponseMessage` to preserve + * model reasoning across turns, so it round-trips here the same way a signature does. + */ + krc?: string; } export function encodeReasoningEnvelope(envelope: ReasoningEnvelope): string { @@ -45,7 +51,9 @@ export function decodeReasoningEnvelope(encryptedContent: string): ReasoningEnve } const txt = (parsed as { txt?: unknown }).txt; if (typeof txt === "string" && txt.length > 0) envelope.txt = txt; - return envelope.sig || envelope.red || envelope.txt ? envelope : null; + const krc = (parsed as { krc?: unknown }).krc; + if (typeof krc === "string" && krc.length > 0) envelope.krc = krc; + return envelope.sig || envelope.red || envelope.txt || envelope.krc ? envelope : null; } catch { return null; } diff --git a/src/server/index.ts b/src/server/index.ts index dc56c01ea..1ef97c64a 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -251,18 +251,15 @@ function attachLiveSidebandUpstream(ws: ServerWebSocket): void { // if (isEventStream && upstreamResponse.body) { // const repairConfig = route.provider.responsesItemIdRepair; // const needsClientRewrite = imageGenCallAliases.size > 0 -// #314 gated shape: win32 no-rewrite traffic follows runtime/config policy; darwin no-rewrite -// traffic requires explicit config-eager opt-in (`auto` always stays tee on darwin). Default OFF -// on the bundled known-bad runtime; policy lives in 260731_macos_rss_retention phase 100. +// #314 gated shape: win32 always uses the terminal-aware eager relay so a keep-alive +// upstream cannot hold Codex open after response.completed; darwin no-rewrite traffic +// requires explicit config-eager opt-in (`auto` always stays tee on darwin). // selectEagerPath(process.platform, needsClientRewrite, config.streamMode ?? "auto") // relaySseEagerBounded(upstreamResponse.body, turnAc, // new Response(eagerBody, // Default shape (tee + background inspection): // upstreamResponse.body.tee() // const repairedBody = hasResponsesItemIdRepair(repairConfig) -// process.platform === "win32" -// && !needsClientRewrite -// ? nativeBody // relaySseWithFailedTail(repairedBody, upstream) // new Response(clientBody // markNativePassthroughSseResponse diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index ee8ac88e7..5c1241f22 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -24,7 +24,7 @@ * up to the drain window. */ -import { buildFailedTailPayload } from "./relay"; +import { buildFailedTailPayload, createSseTerminalOutputBoundary } from "./relay"; import { nextSseBlock, replaceSseDataPayload, @@ -92,6 +92,7 @@ export function relaySseEagerBounded( const now = opts?.now ?? Date.now; const reader = body.getReader(); + const terminalBoundary = createSseTerminalOutputBoundary(); const rewrite = hooks.rewritePayload; const rewriteDecoder = rewrite ? new TextDecoder() : null; const rewriteEncoder = rewrite ? new TextEncoder() : null; @@ -161,6 +162,7 @@ export function relaySseEagerBounded( let queuedBytes = 0; let cancelled = false; let done = false; + const terminalSentinel = new TextEncoder().encode("data: [DONE]\n\n"); // Pause gate: resolved by client pull, client cancel, or upstream abort so a // paused producer ALWAYS resumes (audit blocker 2 — no deadlock; onDone and // turn unregistration stay reachable, drainAndShutdown never hangs). @@ -216,12 +218,17 @@ export function relaySseEagerBounded( if (upstream.signal.aborted) break; if (upstreamDone) { hooks.finishInspection(); + const boundedTail = terminalBoundary.finish(); if (rewrite) { - const tail = flushRewriteTail(); + const rewritten = rewriteOutbound(boundedTail); + const tail = joinUint8Arrays(rewritten, flushRewriteTail()); if (tail.byteLength > 0 && !cancelled) { queuedBytes += tail.byteLength; try { controllerRef?.enqueue(tail); } catch { /* client already gone */ } } + } else if (boundedTail.byteLength > 0 && !cancelled) { + queuedBytes += boundedTail.byteLength; + try { controllerRef?.enqueue(boundedTail); } catch { /* client already gone */ } } if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { syntheticKind = "incomplete"; @@ -237,17 +244,30 @@ export function relaySseEagerBounded( } continue; } - const outbound = rewrite ? rewriteOutbound(value) : value; - if (outbound.byteLength === 0) continue; - queuedBytes += outbound.byteLength; - try { - controllerRef?.enqueue(outbound); - } catch { - // Controller already torn down (client went away without cancel()). - cancelled = true; - drainDeadline = now() + drainMs; - armDrainTimer(); - continue; + const terminalBounded = terminalBoundary.feed(value); + const outbound = rewrite ? rewriteOutbound(terminalBounded) : terminalBounded; + if (outbound.byteLength > 0) { + queuedBytes += outbound.byteLength; + try { + controllerRef?.enqueue(outbound); + } catch { + // Controller already torn down (client went away without cancel()). + cancelled = true; + drainDeadline = now() + drainMs; + armDrainTimer(); + continue; + } + } + if (terminalBoundary.terminalSeen()) { + // The Responses terminal event ends the turn even when a compatible + // gateway keeps its HTTP connection alive. Add the conventional + // sentinel and stop the single-reader relay at that protocol boundary. + if (!terminalBoundary.doneSeen()) { + queuedBytes += terminalSentinel.byteLength; + try { controllerRef?.enqueue(terminalSentinel); } catch { /* client already gone */ } + } + reader.cancel("Responses terminal event received").catch(() => {}); + break; } while (queuedBytes > maxQueueBytes && !cancelled && !upstream.signal.aborted) { await paused(); @@ -279,6 +299,7 @@ export function relaySseEagerBounded( try { rewriteBudget.releaseRetained(frameBufferBytes, { kind: "live_transient" }); } catch { /* teardown must not throw */ } frameBufferBytes = 0; } + terminalBoundary.dispose(); if (syntheticKind) hooks.onSynthetic(syntheticKind); if (cancelled && !hooks.sawTerminal()) { hooks.onClientCancel(); @@ -318,3 +339,12 @@ export function relaySseEagerBounded( }, }); } + +function joinUint8Arrays(first: Uint8Array, second: Uint8Array): Uint8Array { + if (first.byteLength === 0) return second; + if (second.byteLength === 0) return first; + const joined = new Uint8Array(first.byteLength + second.byteLength); + joined.set(first); + joined.set(second, first.byteLength); + return joined; +} diff --git a/src/server/relay.ts b/src/server/relay.ts index 06d1fec18..3b5aae236 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -95,6 +95,79 @@ export function buildFailedTailPayload(err: unknown): string { }); } +export type SseTerminalOutputBoundary = { + feed(chunk: Uint8Array): Uint8Array; + finish(): Uint8Array; + terminalSeen(): boolean; + doneSeen(): boolean; + dispose(): void; +}; + +/** + * Frame-aware client output boundary shared by both native Responses relays. + * It buffers only the current incomplete SSE block, forwards complete blocks + * through the first Responses terminal, and drops every later block/byte. + */ +export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { + let decoder: TextDecoder | null = new TextDecoder(); + const encoder = new TextEncoder(); + let buffer = ""; + let terminal = false; + let done = false; + let disposed = false; + + const process = (flush: boolean): Uint8Array => { + if (disposed || terminal) return new Uint8Array(0); + let output = ""; + let responsesTerminal = false; + for (;;) { + const next = nextSseBlock(buffer); + if (!next) break; + buffer = next.rest; + const payload = sseDataPayload(next.block); + if (!responsesTerminal) output += next.block + next.delimiter; + if (payload === "[DONE]") { + done = true; + if (responsesTerminal) output += next.block + next.delimiter; + continue; + } + if (!responsesTerminal && payload && terminalStatusFromSsePayload(payload)) { + responsesTerminal = true; + } + } + if (responsesTerminal) { + terminal = true; + buffer = ""; + } + if (flush && !terminal && buffer.length > 0) { + output += buffer; + buffer = ""; + } + return encoder.encode(output); + }; + + return { + feed(chunk) { + if (disposed || terminal) return new Uint8Array(0); + buffer += decoder!.decode(chunk, { stream: true }); + return process(false); + }, + finish() { + if (disposed || terminal) return new Uint8Array(0); + buffer += decoder!.decode(); + return process(true); + }, + terminalSeen: () => terminal, + doneSeen: () => done, + dispose() { + if (disposed) return; + disposed = true; + decoder = null; + buffer = ""; + }, + }; +} + /** * Relay a passthrough SSE body like relayWithAbort, but convert a MID-STREAM failure (upstream * reset after headers) into a clean terminal: any partial block is closed off, then a synthetic @@ -110,18 +183,57 @@ export function relaySseWithFailedTail( ): ReadableStream { const reader = body.getReader(); const encoder = new TextEncoder(); + const terminalBoundary = createSseTerminalOutputBoundary(); + let closed = false; + const relayChunk = ( + controller: ReadableStreamDefaultController, + value: Uint8Array, + ): "terminal" | "output" | "buffered" => { + const outbound = terminalBoundary.feed(value); + if (outbound.byteLength > 0) controller.enqueue(outbound); + if (!terminalBoundary.terminalSeen()) return outbound.byteLength > 0 ? "output" : "buffered"; + + // A Responses terminal frame is the protocol boundary. Some compatible + // gateways leave the HTTP connection open after response.completed, which + // otherwise leaves Codex waiting forever even though the model turn is done. + // Preserve through the terminal block only, add the conventional sentinel + // when there was no real [DONE] data event, then stop reading upstream. + if (!terminalBoundary.doneSeen()) { + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + } + closed = true; + controller.close(); + const reason = "Responses terminal event received"; + // Notify the tee inspection branch as well. It has already received the + // same terminal-bearing upstream chunk, so its bounded drain records the + // real terminal and then releases the turn/upstream keep-alive connection. + onClientGone?.(reason); + reader.cancel(reason).catch(() => {}); + terminalBoundary.dispose(); + return "terminal"; + }; return new ReadableStream({ async pull(controller) { try { - const { done, value } = await reader.read(); - if (done) { - controller.close(); - return; + for (;;) { + const { done, value } = await reader.read(); + if (done) { + const tail = terminalBoundary.finish(); + if (tail.byteLength > 0) controller.enqueue(tail); + terminalBoundary.dispose(); + controller.close(); + return; + } + const result = relayChunk(controller, value); + if (result !== "buffered") return; } - controller.enqueue(value); } catch (err) { + const partial = terminalBoundary.finish(); + terminalBoundary.dispose(); + if (closed) return; const payload = buildFailedTailPayload(err); try { + if (partial.byteLength > 0) controller.enqueue(partial); // Leading blank line terminates a partial SSE block so the failed frame parses cleanly. controller.enqueue(encoder.encode(`\n\nevent: response.failed\ndata: ${payload}\n\ndata: [DONE]\n\n`)); controller.close(); @@ -130,6 +242,7 @@ export function relaySseWithFailedTail( } }, cancel(reason) { + terminalBoundary.dispose(); if (onClientGone) onClientGone(reason); else upstream.abort(reason); reader.cancel(reason).catch(() => {}); @@ -137,11 +250,12 @@ export function relaySseWithFailedTail( }); } -export function nextSseBlock(buffer: string): { block: string; rest: string } | null { +export function nextSseBlock(buffer: string): { block: string; delimiter: string; rest: string } | null { const match = buffer.match(/\r?\n\r?\n/); if (!match || match.index === undefined) return null; return { block: buffer.slice(0, match.index), + delimiter: match[0], rest: buffer.slice(match.index + match[0].length), }; } diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 505c5c397..87dc4fe8f 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -491,6 +491,10 @@ export async function handleResponsesCompact( // Always record the real upstream status: a local buffering failure after a // 200 upstream response must not soft-avoid a healthy account or rotate a thread. recordCompactPoolOutcome(outcomeCtx, upstream.status, { retryAfter, resetAt }); + // Lift usage and response metadata from the buffered upstream JSON into the + // request log; the routed branch gets the same through handleResponses. The + // synthetic buffer errors are not upstream bodies and stay uninspected. + if (buffered.ok) inspectResponseLogJson(logCtx, await buffered.clone().text()); return buffered; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 2271b04fd..8975a65c4 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1903,7 +1903,9 @@ async function handleResponsesInner( inspectChunk: chunk => inspector.feed(chunk), finishInspection: () => inspector.finish(), disposeInspection: () => inspector.dispose(), - sawTerminal: () => inspector.reported(), + // Stream lifetime follows the protocol terminal even when this request + // has no outcome callback configured (reported() would stay false). + sawTerminal: () => inspector.terminalSeen(), ...(win32EagerRewrite ? { rewritePayload: composeSsePayloadRewrites(...payloadRewrites) } : {}), @@ -1921,9 +1923,10 @@ async function handleResponsesInner( onClientCancel: () => options.onNativePassthroughCancel?.(), onDone: () => unregisterTurn(turnAc), }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined); - // selectEagerPath admits only no-rewrite traffic on both eligible platforms; - // win32 rewrite traffic reaches this relay too, but with the payload rewrite - // applied inline — never via an image/item-id JS pull wrapper (#32111, #864). + // When selected, this relay closes response.completed even if upstream + // keeps the connection alive. Windows rewrite traffic applies its + // payload transform inline — never via the Bun#32111-unsafe + // tee()+JS-pull chain (#864). if (!headers.has("content-type")) headers.set("content-type", "text/event-stream"); return markEagerRelaySseResponse( markNativePassthroughSseResponse(new Response(eagerBody, { @@ -1990,15 +1993,13 @@ async function handleResponsesInner( ); } if (!headers.has("content-type")) headers.set("content-type", "text/event-stream"); - // win32 must keep the pure native relay (Bun#32111 JS-sink segfault); elsewhere a JS pull - // relay is established practice (relayWithAbort, relaySseWithHeartbeat) and lets a - // mid-stream reset end with a clean response.failed terminal instead of a raw socket error. + // Windows was handled by the eager terminal-aware branch above. Remaining + // tee traffic can use the JS relay to close on a protocol terminal and to + // convert a mid-stream reset into a clean response.failed event. const rewrittenBody = payloadRewrites.length > 0 ? relaySseWithPayloadRewrite(nativeBody, composeSsePayloadRewrites(...payloadRewrites), translatorBudget) : nativeBody; - const clientBody = process.platform === "win32" && !needsClientRewrite - ? nativeBody - : relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); + const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); return markNativePassthroughSseResponse(new Response(clientBody, { status: upstreamResponse.status, headers, diff --git a/src/types.ts b/src/types.ts index 0bbdc7e41..faceb9968 100644 --- a/src/types.ts +++ b/src/types.ts @@ -85,6 +85,12 @@ export interface OcxAssistantMessage { phase?: OcxMessagePhase; model?: string; timestamp: number; + /** + * Kiro `reasoningContent.redactedContent` for THIS assistant turn — an opaque encrypted blob + * Kiro replays to preserve model reasoning across turns. Provider-specific and unrenderable, so + * it rides the message rather than a content part: any other adapter simply ignores it. + */ + kiroRedactedReasoning?: string; } export interface OcxDeveloperMessage { @@ -254,6 +260,9 @@ export type AdapterEvent = // opaque redacted_thinking blocks. Both must be replayed verbatim or tool-use turns 400. | { type: "thinking_signature"; signature: string } | { type: "redacted_thinking"; data: string } + // Kiro reasoning round-trip: the encrypted `redactedContent` blob for the CURRENT assistant turn. + // Never rendered — it only rides the reasoning item's envelope so the next request can replay it. + | { type: "kiro_redacted_reasoning"; data: string } | { type: "reasoning_raw_delta"; text: string } | { type: "tool_call_start"; id: string; name: string } | { type: "tool_call_delta"; arguments: string } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 78f2cc90f..00b36e03e 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -38,19 +38,19 @@ to GUI static serving. Native passthrough SSE has TWO shapes, selected per request in `src/server/responses/core.ts`: -- **Default: tee + background inspection.** `upstreamResponse.body.tee()` sends - branch[0] to the client (pure native relay on win32 without any client-facing - rewrite — the Bun#32111 crash workaround; a JS relay elsewhere) while branch[1] is +- **Default outside Windows: tee + background inspection.** `upstreamResponse.body.tee()` sends + branch[0] through a terminal-aware client relay while branch[1] is drained eagerly by `consumeForInspection`/`consumeForResponseLogMetadata` for terminal-outcome recording, quota, the passthrough continuation cache, and request logs. This remains the default shape on bundled Bun 1.3.14. -- **Gated: eager bounded relay** (`src/server/relay-eager.ts`). win32 and darwin - no-client-rewrite traffic only (neither image-gen aliases nor item-id repair), - selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`. Windows `auto` - becomes eager only on runtimes proven to carry the Bun#32111 fix - (`MIN_FIXED_BUN_VERSION`, null until a bundle bump), while explicit - `streamMode: "eager-relay"` opts in today. Darwin is explicit-only: `auto` - stays tee even after a future threshold bump. One eager reader + byte-bounded +- **Terminal-aware eager bounded relay** (`src/server/relay-eager.ts`). Windows + uses this single-reader shape for rewrite traffic and for no-rewrite traffic + selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`; the latter keeps + `legacy-tee` and known-bad-runtime `auto` on tee as documented. When selected, + `response.completed` closes the client stream even if upstream keeps HTTP/SSE + alive. Darwin uses it for no-client-rewrite traffic only (neither image-gen + aliases nor item-id repair) and is explicit-only: `auto` stays tee even after + a future threshold bump. One eager reader + byte-bounded client queue + post-cancel bounded discard-drain replaces the tee and goes directly to the response without a JS rewrite wrapper, preserving the full inspection side-effect set (shared `createSseInspector` factory in `relay.ts`) @@ -399,6 +399,45 @@ Grounded in the open-sourced official client (xai-org/grok-build); unit + eviden `fetchWithHeaderTimeout` takes an executor so provider fetch wrappers stay inside the timeout race. +## Kiro reasoning round-trip (`redactedContent`) + +Kiro never returns plaintext reasoning for its **GPT-5.6 family** (`gpt-5.6-sol`, `-terra`, +`-luna`): `reasoningContentEvent` carries a KMS-encrypted `redactedContent` blob, never `text`. +Their `additionalModelRequestFieldsSchema` (`ListAvailableModels`) accepts only `reasoning.effort` +with `additionalProperties: false` — there is no display/summary opt-in, so this is the only +reasoning these models can return. Kiro's own CLI replays the blob on the matching +`assistantResponseMessage.reasoningContent` to preserve model reasoning across turns; dropping it +makes every turn restart without the previous turn's reasoning. Verified on kiro-cli 2.14.1 and +2.16.0, all three models. + +The Claude 4.6+/5 entries advertise a different, richer contract (`thinking.type` adaptive/disabled, +`thinking.display` summarized/omitted, `output_config.effort`, `max_tokens`) and are not covered by +that measurement; older Claude, deepseek, minimax, glm, and qwen entries advertise no additional +fields at all. The handling below keys off the wire field, not the model id, so any model that +sends `redactedContent` round-trips. + +- The blob rides the existing `ocxr1:` envelope as `krc` (`src/responses/reasoning-envelope.ts`) on + an envelope-only reasoning item — `summary: []`, no text deltas — so it stays invisible in the + Codex app while round-tripping, exactly like the hidden-thinking path. +- **Pairing is backwards.** Kiro emits `reasoningContentEvent` at the END of an assistant turn, + after content AND tool calls. A `krc`-only item therefore belongs to the turn that already + closed, so the parser attaches it to the PRECEDING assistant message rather than folding it into + the following turn like ordinary reasoning (`src/responses/parser.ts`). With no assistant turn to + own it, the blob is dropped rather than mis-paired. +- The blob lives on `OcxAssistantMessage.kiroRedactedReasoning`, not on a thinking content part, so + no other adapter replays provider-private state if the conversation switches providers. + +Kiro reports context pressure in its own `contextUsageEvent`, which is the authoritative source. On +every capture taken (2.14.1 and 2.16.0) `metadataEvent` carried only `stopReason` — which is why +reading the percentage from `metadataEvent` alone never saw a value — but the parser still accepts a +finite `contextUsagePercentage` (and a `tokenUsage` block) there as a fallback, so a value parsed +from `metadataEvent` is legitimate rather than impossible. Both feed the same field, and any +positive value overwrites an earlier one. + +Spend arrives in `meteringEvent` as **credits, not tokens**. No captured response carried +`tokenUsage` on any event, which is why Kiro usage stays estimated; `meteringEvent` is currently +ignored because a credit is not a token count. + ## Parallel tool calls (default-on for chat providers) The openai-chat adapter buffers ALL streamed `tool_calls` deltas (keyed by `index`, falling back to diff --git a/tests/anthropic-thinking-signature.test.ts b/tests/anthropic-thinking-signature.test.ts index 14180680a..e1227e33a 100644 --- a/tests/anthropic-thinking-signature.test.ts +++ b/tests/anthropic-thinking-signature.test.ts @@ -166,6 +166,39 @@ describe("parser ocxr1 decode + anthropic replay", () => { expect(thinking?.redacted).toEqual(["RED1"]); }); + // Kiro emits its reasoningContentEvent at the END of an assistant turn (after content AND tool + // calls), so a krc-only envelope belongs to the turn BEFORE it. Folding it forward like ordinary + // reasoning would attach turn N's blob to turn N+1 and hand Kiro a mismatched blob. + test("krc-only reasoning attaches to the preceding assistant turn", async () => { + const parsed = parseRequest({ + model: "kiro/gpt-5.6-sol", + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "first" }] }, + { type: "reasoning", id: "rs_1", summary: [], encrypted_content: encodeReasoningEnvelope({ krc: "BLOB1" }) }, + { type: "message", role: "user", content: [{ type: "input_text", text: "more" }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "second" }] }, + ], + }); + const assistants = parsed.context.messages.filter(m => m.role === "assistant"); + expect(assistants).toHaveLength(2); + expect((assistants[0] as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe("BLOB1"); + expect((assistants[1] as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBeUndefined(); + }); + + test("krc-only reasoning with no preceding assistant turn is dropped, not mis-paired", async () => { + const parsed = parseRequest({ + model: "kiro/gpt-5.6-sol", + input: [ + { type: "reasoning", id: "rs_1", summary: [], encrypted_content: encodeReasoningEnvelope({ krc: "ORPHAN" }) }, + { type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "answer" }] }, + ], + }); + const assistant = parsed.context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBeUndefined(); + }); + test("hidden signed text (txt) is restored as the thinking body", async () => { const encrypted = encodeReasoningEnvelope({ sig: "RealSig1234567890==", txt: "the hidden signed text" }); const parsed = parseRequest({ diff --git a/tests/claude-inbound.test.ts b/tests/claude-inbound.test.ts index 2065169dc..2edb35ca1 100644 --- a/tests/claude-inbound.test.ts +++ b/tests/claude-inbound.test.ts @@ -175,6 +175,41 @@ describe("claude inbound translation", () => { expect(() => parseRequest(body)).not.toThrow(); }); + test("tool_result document blocks surface the attachment marker", () => { + const body = anthropicToResponsesBody({ + model: "m", max_tokens: 10, + messages: [ + { role: "assistant", content: [{ type: "tool_use", id: "t1", name: "Read", input: {} }] }, + { + role: "user", + content: [{ + type: "tool_result", tool_use_id: "t1", + content: [ + { type: "text", text: "3 pages" }, + { type: "document", source: { type: "base64", media_type: "application/pdf", data: "aWc=" }, title: "report.pdf" }, + ], + }], + }, + { role: "assistant", content: [{ type: "tool_use", id: "t2", name: "Read", input: {} }] }, + { + role: "user", + content: [{ + type: "tool_result", tool_use_id: "t2", + content: [{ type: "document", source: { type: "base64", media_type: "application/pdf", data: "aWc=" } }], + }], + }, + ], + }) as any; + expect(body.input[1].output).toEqual([ + { type: "input_text", text: "3 pages" }, + { type: "input_text", text: "[document: report.pdf]" }, + ]); + // An untitled document still leaves a marker rather than the empty output that + // read as "the tool returned nothing". + expect(body.input[3].output).toEqual([{ type: "input_text", text: "[document]" }]); + expect(() => parseRequest(body)).not.toThrow(); + }); + test("modelMap: exact, date-stripped, passthrough", () => { const cc = { modelMap: { "claude-sonnet-4-5": "gemini/gemini-3-flash", "claude-opus-4": "xai/grok-4" } }; expect(resolveInboundModel("claude-sonnet-4-5", cc)).toBe("gemini/gemini-3-flash"); diff --git a/tests/claude-outbound.test.ts b/tests/claude-outbound.test.ts index ab997823b..8651f779d 100644 --- a/tests/claude-outbound.test.ts +++ b/tests/claude-outbound.test.ts @@ -207,6 +207,58 @@ describe("claude outbound SSE", () => { expect(startIndexes).toEqual([0, 1, 2]); }); + test("multi-part reasoning summaries keep the JSON path's part separator", async () => { + const upstream = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.output_item.added", { output_index: 0, item: { type: "reasoning", id: "rs_1" } }), + sse("response.reasoning_summary_part.added", { item_id: "rs_1", output_index: 0, summary_index: 0, part: { type: "summary_text", text: "" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 0, delta: "**A**\n\nOne." }), + sse("response.reasoning_summary_part.added", { item_id: "rs_1", output_index: 0, summary_index: 1, part: { type: "summary_text", text: "" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 1, delta: "**B**\n\nTwo." }), + sse("response.output_item.done", { output_index: 0, item: { type: "reasoning", id: "rs_1" } }), + sse("response.output_item.added", { output_index: 1, item: { type: "reasoning", id: "rs_2" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_2", output_index: 1, summary_index: 0, delta: "Three." }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } }), + ].join(""); + const msg = await collectAnthropicMessage(responsesSseToAnthropicSse(streamFrom(upstream), "m"), "m") as Record; + // Parts within an item are separated; a new reasoning item opens its own block. + const thinkingBlocks = msg.content.filter((b: Record) => b.type === "thinking"); + expect(thinkingBlocks.map((b: Record) => b.thinking)).toEqual([ + "**A**\n\nOne.\n\n**B**\n\nTwo.", + "Three.", + ]); + + // Parity: the non-streaming translator joins the same summary parts identically. + const json = responsesJsonToAnthropicMessage({ + id: "resp_1", + status: "completed", + output: [{ type: "reasoning", id: "rs_1", summary: [{ type: "summary_text", text: "**A**\n\nOne." }, { type: "summary_text", text: "**B**\n\nTwo." }] }], + usage: { input_tokens: 1, output_tokens: 1 }, + }, "m") as Record; + const jsonThinking = json.content.find((b: Record) => b.type === "thinking"); + expect(jsonThinking.thinking).toBe("**A**\n\nOne.\n\n**B**\n\nTwo."); + }); + + test("same-part deltas and index-free reasoning frames never get a separator", async () => { + const samePart = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 0, delta: "Hel" }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 0, delta: "lo" }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } }), + ].join(""); + const msg1 = await collectAnthropicMessage(responsesSseToAnthropicSse(streamFrom(samePart), "m"), "m") as Record; + expect(msg1.content.find((b: Record) => b.type === "thinking").thinking).toBe("Hello"); + + const indexFree = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.reasoning_text.delta", { delta: "A" }), + sse("response.reasoning_text.delta", { delta: "B" }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } }), + ].join(""); + const msg2 = await collectAnthropicMessage(responsesSseToAnthropicSse(streamFrom(indexFree), "m"), "m") as Record; + expect(msg2.content.find((b: Record) => b.type === "thinking").thinking).toBe("AB"); + }); + test("data-only Responses frames infer event names from payload types", async () => { const upstream = [ dataOnlySse({ type: "response.created", response: { id: "resp_data_only", status: "in_progress" } }), diff --git a/tests/google-vertex-stream.test.ts b/tests/google-vertex-stream.test.ts index 120883c16..996c9ba73 100644 --- a/tests/google-vertex-stream.test.ts +++ b/tests/google-vertex-stream.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test"; import { createGoogleAdapter as createGoogleAdapterProduction } from "../src/adapters/google"; -import { isVertexTruncationReason, vertexTruncationErrorMessage } from "../src/adapters/google-truncation"; +import { isVertexTruncatedTurn, isVertexTruncationReason, vertexTruncationErrorMessage } from "../src/adapters/google-truncation"; import { bridgeToResponsesSSE } from "../src/bridge"; import type { AdapterEvent, OcxProviderConfig } from "../src/types"; import { withTestTranslatorBudget } from "./helpers/translator-budget"; @@ -30,6 +30,14 @@ describe("vertex truncation helpers", () => { expect(isVertexTruncationReason(undefined)).toBe(false); expect(vertexTruncationErrorMessage("MAX_TOKENS")).toContain("truncated upstream"); }); + + test("MALFORMED_FUNCTION_CALL fails closed with zero started calls; MAX_TOKENS does not", () => { + expect(isVertexTruncatedTurn("MALFORMED_FUNCTION_CALL", 0)).toBe(true); + expect(isVertexTruncatedTurn("MAX_TOKENS", 0)).toBe(false); + expect(isVertexTruncatedTurn("MAX_TOKENS", 1)).toBe(true); + expect(isVertexTruncatedTurn("STOP", 5)).toBe(false); + expect(isVertexTruncatedTurn(undefined, 5)).toBe(false); + }); }); describe("vertex parseStream fail-closed truncation", () => { @@ -77,6 +85,17 @@ describe("vertex parseStream fail-closed truncation", () => { expect(text).toContain('"incomplete_details":{"reason":"max_output_tokens"}'); }); + test("MALFORMED_FUNCTION_CALL with NO emitted call part yields a terminal error, not done", async () => { + // The malformed call is dropped upstream, so the final chunk usually carries only the + // finishReason. Without the guard this surfaced as a clean empty completion. + const events = await collect(vertexProvider, [ + { candidates: [{ finishReason: "MALFORMED_FUNCTION_CALL" }], usageMetadata: { promptTokenCount: 5, candidatesTokenCount: 0 } }, + ]); + const last = events[events.length - 1]; + expect(last.type).toBe("error"); + expect(events.some(e => e.type === "done")).toBe(false); + }); + test("usage-only final chunk (no candidates) is not dropped", async () => { const events = await collect(vertexProvider, [ { candidates: [{ content: { parts: [{ text: "hi" }] } }] }, @@ -98,6 +117,14 @@ describe("vertex parseResponse fail-closed truncation (non-streaming)", () => { expect(events.some(e => e.type === "done")).toBe(false); }); + test("MALFORMED_FUNCTION_CALL with no call part yields a terminal error, not done", async () => { + const adapter = createGoogleAdapter(vertexProvider); + const body = JSON.stringify({ candidates: [{ finishReason: "MALFORMED_FUNCTION_CALL" }], usageMetadata: { promptTokenCount: 5, candidatesTokenCount: 0 } }); + const events = await adapter.parseResponse!(new Response(body, { status: 200 })); + expect(events[events.length - 1].type).toBe("error"); + expect(events.some(e => e.type === "done")).toBe(false); + }); + test("clean STOP non-stream response yields done", async () => { const adapter = createGoogleAdapter(vertexProvider); const body = JSON.stringify({ candidates: [{ content: { parts: [{ text: "ok" }] }, finishReason: "STOP" }], usageMetadata: { promptTokenCount: 2, candidatesTokenCount: 1 } }); diff --git a/tests/kiro-adapter.test.ts b/tests/kiro-adapter.test.ts index 2288567eb..2cc49fb38 100644 --- a/tests/kiro-adapter.test.ts +++ b/tests/kiro-adapter.test.ts @@ -268,6 +268,32 @@ describe("kiro adapter — buildRequest", () => { expect(results[0].status).toBe("success"); }); + // Kiro's own client replays the encrypted reasoning blob on the assistant turn it belongs to; + // dropping it makes every turn start without the previous turn's reasoning. + test("assistant history replays the Kiro redacted reasoning blob", async () => { + const messages = [ + { role: "user", content: "think" }, + { role: "assistant", content: [{ type: "text", text: "answer" }], kiroRedactedReasoning: "LktUUn5+blob" }, + { role: "user", content: "again" }, + ]; + const { body } = await createKiroAdapter(provider).buildRequest(parsedWith(messages)); + const arm = JSON.parse(body).conversationState.history + .find((h: { assistantResponseMessage?: unknown }) => h.assistantResponseMessage)?.assistantResponseMessage; + expect(arm.reasoningContent).toEqual({ redactedContent: "LktUUn5+blob" }); + }); + + test("assistant history omits reasoningContent when no blob was captured", async () => { + const messages = [ + { role: "user", content: "think" }, + { role: "assistant", content: [{ type: "text", text: "answer" }] }, + { role: "user", content: "again" }, + ]; + const { body } = await createKiroAdapter(provider).buildRequest(parsedWith(messages)); + const arm = JSON.parse(body).conversationState.history + .find((h: { assistantResponseMessage?: unknown }) => h.assistantResponseMessage)?.assistantResponseMessage; + expect(arm).not.toHaveProperty("reasoningContent"); + }); + test("empty tool output is normalized to a non-empty Kiro result block", async () => { const messages = [ { role: "user", content: "run it" }, diff --git a/tests/kiro-reasoning-roundtrip.test.ts b/tests/kiro-reasoning-roundtrip.test.ts new file mode 100644 index 000000000..a38121d59 --- /dev/null +++ b/tests/kiro-reasoning-roundtrip.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, test } from "bun:test"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../src/bridge"; +import { parseRequest } from "../src/responses/parser"; +import { decodeReasoningEnvelope } from "../src/responses/reasoning-envelope"; +import type { AdapterEvent } from "../src/types"; +import { createTranslatorBudget } from "../src/lib/translator-budget"; + +const BLOB = "LktUUn5+ZXlKbGJtTnllWEIwYVc5dVVtVm5hVzl1SWpvaQ=="; + +async function* replay(events: AdapterEvent[]): AsyncGenerator { + for (const event of events) yield event; +} + +async function drain(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let out = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + out += decoder.decode(value, { stream: true }); + } + return out; +} + +/** Output items in emission order, as Codex reconstructs them from the SSE stream. */ +function doneItems(sse: string): Record[] { + const items: Record[] = []; + for (const line of sse.split("\n")) { + if (!line.startsWith("data: ")) continue; + try { + const json = JSON.parse(line.slice(6)) as { type?: string; item?: Record; output_index?: number }; + if (json.type === "response.output_item.done" && json.item) { + items.push({ ...json.item, __index: json.output_index }); + } + } catch { /* partial frame */ } + } + return items; +} + +/** Feed emitted items back as Responses input, the way Codex replays history next turn. */ +function reparse(items: Record[]) { + return parseRequest({ + model: "kiro/gpt-5.6-sol", + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }, + ...items.map(({ __index, status, ...item }) => item), + ], + }); +} + +// Kiro emits its reasoning blob at the END of a turn, while the assistant message is still open. +// Emitting the envelope item on arrival reused the open message's output_index AND placed the blob +// before the message, where the parser's backwards pairing drops it as orphaned — silently +// defeating the round-trip. Both paths must defer it until the message has closed. +describe("kiro redacted-reasoning round-trip (bridge → parse)", () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "the answer" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, + ]; + + test("SSE: the blob lands after the assistant message, on its own output index", async () => { + const items = doneItems(await drain(bridgeToResponsesSSE(replay(events), "kiro/gpt-5.6-sol"))); + const types = items.map(i => i.type); + expect(types).toEqual(["message", "reasoning"]); + + const indexes = items.map(i => i.__index); + expect(new Set(indexes).size).toBe(indexes.length); // no output_index collision + + const envelope = decodeReasoningEnvelope(items[1].encrypted_content as string); + expect(envelope?.krc).toBe(BLOB); + expect(items[1].summary).toEqual([]); + }); + + test("SSE: replayed history attaches the blob to the assistant turn that produced it", async () => { + const items = doneItems(await drain(bridgeToResponsesSSE(replay(events), "kiro/gpt-5.6-sol"))); + const assistant = reparse(items).context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); + }); + + test("batch: the blob lands after the assistant message and survives replay", () => { + const response = buildResponseJSON( + [ + { type: "text_delta", text: "the answer" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, + ], + "kiro/gpt-5.6-sol", + ); + const output = response.output as Record[]; + expect(output.map(i => i.type)).toEqual(["message", "reasoning"]); + + const assistant = reparse(output).context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); + }); + + test("batch: the raw blob is retained then released, leaving only the finalized items", () => { + // A big blob makes the accounting unambiguous: the raw string is an allocation distinct from + // the finalized item that embeds its base64. + const bigBlob = "X".repeat(4000); + const budget = createTranslatorBudget(); + const response = buildResponseJSON( + [ + { type: "text_delta", text: "the answer" }, + { type: "kiro_redacted_reasoning", data: bigBlob }, + { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, + ], + "kiro/gpt-5.6-sol", + { translatorBudget: budget }, + ); + const items = response.output as Record[]; + const finalizedBytes = items.reduce((sum, item) => sum + Buffer.byteLength(JSON.stringify(item)), 0); + const { currentBytes, highWaterBytes, overflows } = budget.snapshot(); + + // EXACTLY the finalized output items remain retained. A raw blob still held would show up as + // ~4000 extra bytes here; releasing bytes that were never charged would show up as a shortfall. + expect(currentBytes).toBe(finalizedBytes); + // ...and it really was charged while held, rather than never accounted for at all. + expect(highWaterBytes).toBeGreaterThanOrEqual(finalizedBytes + bigBlob.length); + expect(overflows).toBe(0); + }); + + test("a turn ending in a tool call still pairs the blob with that assistant turn", async () => { + const items = doneItems(await drain(bridgeToResponsesSSE(replay([ + { type: "tool_call_start", id: "call_1", name: "bash" }, + { type: "tool_call_delta", arguments: "{\"command\":\"ls\"}" }, + { type: "tool_call_end" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 1 }, endTurn: false }, + ]), "kiro/gpt-5.6-sol"))); + expect(items.map(i => i.type)).toEqual(["function_call", "reasoning"]); + + const assistant = reparse(items).context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); + }); +}); diff --git a/tests/kiro-stream.test.ts b/tests/kiro-stream.test.ts index 4de16927f..b2a92bdf1 100644 --- a/tests/kiro-stream.test.ts +++ b/tests/kiro-stream.test.ts @@ -1306,6 +1306,40 @@ describe("kiro adapter — parseStream", () => { ]); }); + // Kiro's Sol-family models never return plaintext reasoning: reasoningContentEvent carries an + // encrypted `redactedContent` blob (verified against kiro-cli 2.14.1 and 2.16.0), which the + // official client replays on the matching assistantResponseMessage to preserve reasoning across + // turns. Reading only `text` dropped it entirely. + test("reasoningContentEvent redactedContent is captured for round-trip", async () => { + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(streamOf( + eventFrame({ content: "visible answer" }), + eventFrame({ redactedContent: "LktUUn5+encrypted" }, "reasoningContentEvent"), + )))); + expect(events).toEqual([ + { type: "text_delta", text: "visible answer" }, + { type: "kiro_redacted_reasoning", data: "LktUUn5+encrypted" }, + expect.objectContaining({ type: "done", endTurn: true }), + ]); + }); + + test("reasoningContentEvent carrying both text and redactedContent emits both", async () => { + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(streamOf( + eventFrame({ text: "plain", redactedContent: "blob" }, "reasoningContentEvent"), + )))); + expect(events).toEqual([ + { type: "reasoning_raw_delta", text: "plain" }, + { type: "kiro_redacted_reasoning", data: "blob" }, + expect.objectContaining({ type: "done" }), + ]); + }); + + // Kiro reports context pressure in its own event type; metadataEvent carries only stopReason, so + // reading contextUsagePercentage from metadataEvent alone never saw a value. + test("contextUsageEvent supplies the absolute context usage percentage", () => { + const parsed = parseKiroEvent("contextUsageEvent", enc.encode(JSON.stringify({ contextUsagePercentage: 42.5 }))); + expect(parsed).toEqual({ type: "context_usage", contextUsagePercentage: 42.5 }); + }); + test("thinking tags split across chunks are parsed as reasoning", async () => { const frames = [ eventFrame({ content: " { ); expect(sseBranch).toContain("upstreamResponse.body.tee()"); - // win32 must receive the tee'd body untouched when no client rewrite is required — no JS pull - // wrapper on the default path (Bun#32111 segfault). + // Windows no-rewrite traffic must honor the stream-mode/runtime gate so + // legacy-tee remains a safety escape hatch for Bun#32111. expect(sseBranch).toContain("const repairConfig = route.provider.responsesItemIdRepair;"); expect(sseBranch).toContain("const needsClientRewrite = imageGenCallAliases.size > 0"); expect(sseBranch).toContain("new Response(eagerBody"); expect(sseBranch).toContain("const rewrittenBody = payloadRewrites.length > 0"); - expect(sseBranch).toContain('process.platform === "win32"'); - expect(sseBranch).toContain("&& !needsClientRewrite"); - expect(sseBranch).toContain("? nativeBody"); + expect(sseBranch).toContain("eagerPath?.useEagerRelay || win32EagerRewrite"); + expect(sseBranch).not.toContain("win32TerminalRelay"); // #864: win32 traffic that DOES need a client rewrite takes the eager single // reader with the payload rewrite applied inline — never the tee()+JS-pull // chain that loses the terminal block on Windows (Bun#32111). diff --git a/tests/relay-eager.test.ts b/tests/relay-eager.test.ts index a66b759d1..9840ead4b 100644 --- a/tests/relay-eager.test.ts +++ b/tests/relay-eager.test.ts @@ -176,8 +176,10 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { expect(text).toContain("RESTORED"); expect(text).not.toContain("image_gen__gen"); expect(text).toContain("response.completed"); - // A partial trailing block reaches the client verbatim at EOF. - expect(text).toContain("trailing-partial"); + // The protocol terminal ends the client stream; bytes produced after it + // belong to the gateway's retained connection and must not hold Codex open. + expect(text).not.toContain("trailing-partial"); + expect(text.endsWith("data: [DONE]\n\n")).toBe(true); }); test("identity rewrite preserves framing byte-for-byte", async () => { @@ -198,11 +200,37 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { up.close(); const text = await reading; - expect(text).toBe(new TextDecoder().decode(joinBytes([first, enc.encode(second)]))); + expect(text).toBe( + new TextDecoder().decode(joinBytes([first, enc.encode(second)])) + "data: [DONE]\n\n", + ); // The rewrite actually ran — this is what makes the test red pre-fix. expect(rewriteCalls).toBeGreaterThan(0); }); + test("drops coalesced post-terminal frames and detects only a real DONE event", async () => { + for (const realDone of [false, true]) { + const up = controlledUpstream(); + const { hooks } = makeHooks(); + const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks); + const reading = readAll(relayed); + const completed = JSON.stringify({ + type: "response.completed", + response: { status: "completed", note: "data: [DONE]" }, + }); + up.push(enc.encode( + `event: response.completed\ndata: ${completed}\n\n` + + (realDone ? "data: [DONE]\n\n" : "") + + `data: {"type":"response.output_text.delta","delta":"must not leak"}\n\n`, + )); + up.close(); + + const text = await reading; + expect(text).not.toContain("must not leak"); + expect(countOccurrences(text, "\ndata: [DONE]\n\n")).toBe(1); + expect(text.endsWith("data: [DONE]\n\n")).toBe(true); + } + }); + test("unchanged multi-data-line events keep their original framing", async () => { const up = controlledUpstream(); const { hooks } = makeHooks(); @@ -238,7 +266,7 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { expect(text).toBe("data: �"); }); - test("retained rewrite-budget bytes are released on upstream abort", async () => { + test("terminal framing keeps partial blocks out of the rewrite budget", async () => { const budget = createTranslatorBudget(); const up = controlledUpstream(); const ac = new AbortController(); @@ -248,13 +276,15 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { up.push(enc.encode(`data: {"type":"unterminated"`)); await settle(); - expect(budget.snapshot().currentBytes).toBeGreaterThan(0); + // The shared terminal boundary now owns incomplete SSE framing, so the + // downstream rewrite stage never retains an unterminated block. + expect(budget.snapshot().currentBytes).toBe(0); ac.abort(new Error("test abort")); await settle(); expect(budget.snapshot().currentBytes).toBe(0); }); - test("blocks without a data field pass through untouched", async () => { + test("blocks without a data field pass through untouched before the terminal", async () => { const up = controlledUpstream(); const { hooks } = makeHooks(); let rewriteCalls = 0; @@ -265,8 +295,8 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks); const reading = readAll(relayed); - up.push(enc.encode(`event: response.completed\ndata: ${COMPLETED}\n\n`)); up.push(enc.encode(`: keepalive comment\n\n`)); + up.push(enc.encode(`event: response.completed\ndata: ${COMPLETED}\n\n`)); up.close(); const text = await reading; @@ -362,7 +392,7 @@ describe("relaySseEagerBounded — side-effect parity", () => { const clientBytes = await readAllBytes(relayed); await settle(); - expect(clientBytes).toEqual(joinBytes(frames)); + expect(clientBytes).toEqual(joinBytes([...frames, enc.encode("data: [DONE]\n\n")])); const wireText = new TextDecoder().decode(clientBytes); expect(wireText).not.toContain('"output":'); expect(rec.completed).toHaveLength(1); @@ -470,6 +500,45 @@ describe("relaySseEagerBounded — #44 cancel semantics", () => { expect(rec.dones).toBe(1); }); + test("post-cancel terminal ends metadata-only drain without waiting for timeout", async () => { + const inspector = createSseInspector({}); + const up = controlledUpstream(); + const rec = { cancels: 0, dones: 0, synthetics: [] as string[] }; + let resolveDone!: () => void; + const relayDone = new Promise(resolve => { resolveDone = resolve; }); + const relayed = relaySseEagerBounded(up.stream, new AbortController(), { + inspectChunk: chunk => inspector.feed(chunk), + finishInspection: () => inspector.finish(), + disposeInspection: () => inspector.dispose(), + // Mirrors the no-onTerminal wiring in responses/core.ts. + sawTerminal: () => inspector.terminalSeen(), + onSynthetic: kind => { rec.synthetics.push(kind); }, + onClientCancel: () => { rec.cancels += 1; }, + onDone: () => { rec.dones += 1; resolveDone(); }, + }, { postCancelDrainMs: 5_000 }); + const reader = relayed.getReader(); + up.push(sse(DELTA)); + await settle(5); + await reader.cancel(); + + // Keep upstream open after delivering the terminal. The protocol terminal, + // not EOF or the five-second drain timer, must finish the relay lifecycle. + up.push(sse(COMPLETED)); + await Promise.race([ + relayDone, + new Promise((_, reject) => setTimeout( + () => reject(new Error("metadata-only terminal drain waited for timeout")), + 200, + )), + ]); + + expect(inspector.reported()).toBe(false); + expect(inspector.terminalSeen()).toBe(true); + expect(rec.cancels).toBe(0); + expect(rec.synthetics).toEqual([]); + expect(rec.dones).toBe(1); + }); + test("(d) post-cancel drain timeout → onClientCancel fired, upstream aborted", async () => { const { hooks, rec } = makeHooks(); const up = controlledUpstream(); diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index 346aed2d5..b1d350510 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -23,6 +23,7 @@ import { resolveCodexAuthContext, } from "../src/codex/auth-context"; import { supportsNativeResponsesCompactEndpoint } from "../src/providers/openai-tiers"; +import type { RequestLogContext } from "../src/server/request-log"; import type { OcxConfig, OcxProviderConfig } from "../src/types"; const originalFetch = globalThis.fetch; @@ -164,6 +165,33 @@ describe("supportsNativeResponsesCompactEndpoint (#422)", () => { }); }); +describe("native compact usage reporting", () => { + test("the buffered upstream body fills the request log usage and stays intact for the client", async () => { + const config = { + defaultProvider: "openai-apikey", + providers: { + "openai-apikey": { + adapter: "openai-responses", + baseUrl: "https://api.openai.com/v1", + authMode: "key", + apiKey: "sk-test", + }, + }, + } as unknown as OcxConfig; + globalThis.fetch = (async () => jsonResponse(completedPayload("native summary"))) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponsesCompact( + compactionRequest(baseCompactionBody({ model: "openai-apikey/gpt-5.5" })), + config, + logCtx, + ); + expect(response.status).toBe(200); + const body = await response.json() as { usage?: Record }; + expect(body.usage).toMatchObject({ input_tokens: 10, output_tokens: 5, total_tokens: 15 }); + expect(logCtx.usage).toMatchObject({ inputTokens: 10, outputTokens: 5, totalTokens: 15 }); + }); +}); + describe("native Codex pool compaction", () => { test("keeps a Spark reset cooldown separate from a Terra compact request (#590)", async () => { const testDir = mkdtempSync(join(tmpdir(), "ocx-compact-scope-")); diff --git a/tests/sse-failed-tail.test.ts b/tests/sse-failed-tail.test.ts index 1faddcca1..70876dafd 100644 --- a/tests/sse-failed-tail.test.ts +++ b/tests/sse-failed-tail.test.ts @@ -59,6 +59,61 @@ describe("relaySseWithFailedTail", () => { expect(upstream.signal.aborted).toBe(false); }); + test("closes at response.completed when the upstream keeps its SSE connection open", async () => { + const upstream = new AbortController(); + let sourceCancelled = false; + let sentTerminal = false; + const src = new ReadableStream({ + pull(controller) { + if (!sentTerminal) { + sentTerminal = true; + controller.enqueue(encoder.encode( + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n', + )); + } + // Deliberately never close: several Responses-compatible gateways keep + // this connection alive after the protocol terminal event. + }, + cancel() { sourceCancelled = true; }, + }); + + const out = await Promise.race([ + drain(relaySseWithFailedTail(src, upstream)), + new Promise((_, reject) => setTimeout(() => reject(new Error("relay did not close at terminal")), 200)), + ]); + + expect(out).toContain("response.completed"); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + expect(sourceCancelled).toBe(true); + expect(upstream.signal.aborted).toBe(false); + }); + + test("drops frames coalesced after the terminal block", async () => { + const upstream = new AbortController(); + const src = sourceStream([ + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n' + + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"must not leak"}\n\n', + ]); + + const out = await drain(relaySseWithFailedTail(src, upstream)); + + expect(out).toContain("response.completed"); + expect(out).not.toContain("must not leak"); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + }); + + test("recognizes only a real DONE data event", async () => { + const ordinaryText = 'data: {"type":"response.completed","response":{"status":"completed","note":"data: [DONE]"}}\n\n'; + const withRealDone = ordinaryText + "data: [DONE]\n\n"; + + const ordinaryOut = await drain(relaySseWithFailedTail(sourceStream([ordinaryText]), new AbortController())); + const realOut = await drain(relaySseWithFailedTail(sourceStream([withRealDone]), new AbortController())); + + expect(ordinaryOut.endsWith("data: [DONE]\n\n")).toBe(true); + expect(ordinaryOut.split("\ndata: [DONE]\n\n").length - 1).toBe(1); + expect(realOut).toBe(withRealDone); + }); + test("mid-stream error keeps prior bytes and appends a clean failed terminal", async () => { const upstream = new AbortController(); const src = sourceStream(['data: {"type":"response.output_text.delta","delta":"hel', ""], { failAfter: true });