diff --git a/scripts/test-layout/layout-clients.json b/scripts/test-layout/layout-clients.json index fb8a72fe2bf..21cd4d5f083 100644 --- a/scripts/test-layout/layout-clients.json +++ b/scripts/test-layout/layout-clients.json @@ -10,6 +10,7 @@ "cli-export-model-names.test.ts": "cli", "management-export-model-names.test.ts": "server", "claude-native-fast-1m-ingress.test.ts": "claude-integration", + "messages-native-clean-eof.test.ts": "claude-integration", "stop-restoration-preflight.test.ts": "service", "claude-intercept-acl-barrier.test.ts": "claude-integration", "claude-config-root-writer-barrier.test.ts": "claude-integration", diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index b8071a94921..89103065327 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -1190,28 +1190,48 @@ export async function collectAnthropicMessage( } }; - try { - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - buffer = replaceRetained(buffer, buffer + decoder.decode(value, { stream: true }), "live_transient"); - let sep: number; - while ((sep = buffer.indexOf("\n\n")) !== -1) { + const drainFrames = () => { + let sep: number; + while ((sep = buffer.indexOf("\n\n")) !== -1) { const rawFrame = buffer.slice(0, sep); buffer = replaceRetained(buffer, buffer.slice(sep + 2), "live_transient"); + // Event name and data as SSE dispatches them: the last event field, untrimmed, and the + // data fields joined by newlines. let eventName = ""; - let dataLine = ""; + const dataLines: string[] = []; for (const line of rawFrame.split("\n")) { const eventValue = sseFieldValue(line, "event"); - if (eventValue !== null) { eventName = eventValue.trim(); continue; } + if (eventValue !== null) { eventName = eventValue; continue; } const dataValue = sseFieldValue(line, "data"); - if (dataValue !== null) dataLine += dataValue; + if (dataValue !== null) dataLines.push(dataValue); } - if (!eventName || !dataLine) continue; + const dataLine = dataLines.join("\n"); + if (!dataLine) continue; let data: unknown; try { data = JSON.parse(dataLine); } catch { continue; } - if (isRec(data)) handle(eventName, data); - } + if (!isRec(data)) continue; + // A data-only frame is named by its payload type, the same rule the stream tap applies. + const name = eventName || (typeof data.type === "string" ? data.type : ""); + if (name) handle(name, data); + } + }; + + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + // Frame on LF only after normalizing CRLF/CR line endings, holding a trailing CR until the + // next chunk shows whether its LF follows, so a CRLF stream folds like an LF one. + const incoming = buffer + decoder.decode(value, { stream: true }); + const heldCr = incoming.endsWith("\r"); + const normalized = (heldCr ? incoming.slice(0, -1) : incoming).replace(/\r\n?/g, "\n") + (heldCr ? "\r" : ""); + buffer = replaceRetained(buffer, normalized, "live_transient"); + drainFrames(); + } + // At EOF a held CR is a line ending: it can complete the final frame. + if (buffer.endsWith("\r")) { + buffer = replaceRetained(buffer, buffer.slice(0, -1) + "\n", "live_transient"); + drainFrames(); } } finally { reader.releaseLock(); diff --git a/src/server/messages-native.ts b/src/server/messages-native.ts index 0baffa8e020..c15763f9398 100644 --- a/src/server/messages-native.ts +++ b/src/server/messages-native.ts @@ -888,7 +888,10 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions) if (contentType.includes("text/event-stream") && response.body) { // `upstream` follows req.signal and is also what shutdown and turn release abort, so both // the client leaving and this lane's own abort read as a cancel, not as a failed stream. - const bodyGuard = resolvePassthroughBodyGuard(config, upstream.signal); + const bodyGuard = { + ...resolvePassthroughBodyGuard(config, upstream.signal), + ...(activeProvider.anthropicEofTolerance === true ? { eofTolerant: true } : {}), + }; const observed = logIds ? observeFirstChunk(response.body, () => recordFirstOutput(logCtx, logIds.start)) : response.body; const renamed = activeRequest.oauthToolNames ? restoreOAuthToolNamesInSse(observed, activeRequest.oauthToolNames, translatorBudget) diff --git a/src/server/messages-shared.ts b/src/server/messages-shared.ts index 8444fd73737..582e061f0bc 100644 --- a/src/server/messages-shared.ts +++ b/src/server/messages-shared.ts @@ -52,6 +52,31 @@ export function anthropicUsageToOcx(usage: Rec | undefined): { inputTokens: numb } /** Body-occupancy guard for the native passthrough (devlog 260716_passthrough_followups/010). */ +/** + * True when an undelimited SSE tail (LF-normalized) holds nothing a client would act on: only + * whitespace, comments, or a ping. Anything else is a message event cut off at EOF. + */ +export function isHarmlessSseTail(tail: string): boolean { + const lines = tail.split("\n").filter(line => line.trim() !== ""); + let event: string | null = null; + const data: string[] = []; + for (const line of lines) { + if (line.startsWith(":")) continue; + const eventValue = sseFieldValue(line, "event"); + if (eventValue !== null) { event = eventValue; continue; } + const dataValue = sseFieldValue(line, "data"); + if (dataValue !== null) { data.push(dataValue); continue; } + return false; + } + if (data.length === 0) return event === null || event === "ping"; + try { + const payload: unknown = JSON.parse(data.join("\n")); + return isRec(payload) && payload.type === "ping" && (event === null || event === "ping"); + } catch { + return false; + } +} + export interface PassthroughBodyGuard { /** Idle window in ms — raw upstream-byte inactivity while a read is pending. 0 disables. */ stallMs: number; @@ -59,6 +84,11 @@ export interface PassthroughBodyGuard { maxBytes: number; /** Client request signal for deterministic cancel classification. */ reqSignal?: AbortSignal; + /** + * The routed provider opted into anthropicEofTolerance: a content-only clean EOF (no + * message_stop, no stop_reason, nothing cut) stays a completed turn, as it always was. + */ + eofTolerant?: boolean; } export type PassthroughCloseReason = "terminal" | "client_cancel" | "body_stall" | "body_overflow"; @@ -86,23 +116,43 @@ export function tapAnthropicSseForLog( let usageAcc: Rec = {}; // message_stop or an upstream error event: the turn's own terminal already went through. let terminalSeen = false; - const inspectFrame = (frame: string) => { - const dataLine = frame - .split("\n") + // stop_reason of the latest message event the client received as a complete SSE event. Any + // later message event (resumed content) clears it; pings and comments do not. + let lastDeliveredStop: string | null = null; + // A frame is evidence of an event only when its payload type and its event name agree (or the + // event line is absent): clients and the non-stream fold dispatch on the event name. + const named = (event: string | null, data: Rec, type: string) => data.type === type && (event === null || event === type); + const inspectFrame = (frame: string, delivered = true) => { + const lines = frame.split("\n"); + // SSE dispatches on the last event field of a frame. + const eventLine = lines.map(l => sseFieldValue(l, "event")).findLast((v): v is string => v !== null); + // The value is used as SSE delivers it (one leading space stripped, nothing trimmed). + const event = eventLine === undefined ? null : eventLine; + const dataLine = lines .map(l => sseFieldValue(l, "data")) .filter((v): v is string => v !== null) - .join(""); + .join("\n"); if (!dataLine) return; let data: unknown; - try { data = JSON.parse(dataLine); } catch { return; } - if (!isRec(data)) return; + try { data = JSON.parse(dataLine); } catch { data = null; } + if (!isRec(data)) { + // A malformed event after a stop reason means the turn kept going and was cut. + if (delivered) lastDeliveredStop = null; + return; + } recordGenerationEvent(logCtx, data.type); if (data.type === "message_start" && isRec(data.message) && isRec(data.message.usage)) { usageAcc = { ...usageAcc, ...data.message.usage }; } else if (data.type === "message_delta" && isRec(data.usage)) { usageAcc = { ...usageAcc, ...data.usage }; - } else if (data.type === "message_stop" || data.type === "error") { + } + if (named(event, data, "message_stop") || named(event, data, "error")) { terminalSeen = true; + } else if (delivered && !named(event, data, "ping")) { + const delta = isRec(data.delta) ? data.delta : undefined; + lastDeliveredStop = named(event, data, "message_delta") && typeof delta?.stop_reason === "string" + ? delta.stop_reason + : null; } }; const inspect = (chunk: Uint8Array) => { @@ -121,14 +171,24 @@ export function tapAnthropicSseForLog( }; // The last block can sit in the buffer without its blank line: an upstream that stopped after // it, or a held trailing CR. Count it before deciding how the turn ended (the Responses relay - // flushes the same candidate). Returns true when this flush found the terminal in a block the - // client has not seen a blank line after, so the caller restores one. - const flushTail = (): boolean => { - const terminalBefore = terminalSeen; - const tail = (buffer + decoder.decode()).replace(/\r\n?/g, "\n"); + // flushes the same candidate). terminalInTail: this flush found the terminal in a block the + // client has not seen a blank line after, so the caller restores one. tailClean: nothing but + // comments or a ping is left undelimited, so no message event was cut off. + const flushTail = (): { terminalInTail: boolean; tailClean: boolean } => { + // At EOF a held CR is a line ending, so it can complete a frame the client already has. + let tail = (buffer + decoder.decode()).replace(/\r\n?/g, "\n"); buffer = ""; - if (tail) inspectFrame(tail); - return terminalSeen && !terminalBefore && !tail.endsWith("\n\n"); + let sep: number; + while ((sep = tail.indexOf("\n\n")) !== -1) { + inspectFrame(tail.slice(0, sep)); + tail = tail.slice(sep + 2); + } + const terminalBefore = terminalSeen; + if (tail) inspectFrame(tail, false); + return { + terminalInTail: terminalSeen && !terminalBefore, + tailClean: isHarmlessSseTail(tail), + }; }; const reader = upstream.getReader(); let settled = false; @@ -152,7 +212,7 @@ export function tapAnthropicSseForLog( settled = true; idle.cancel(); detachAbort(); - const terminalInTail = flushTail(); + const { terminalInTail } = flushTail(); recordUsage(); if (terminalSeen) { // The turn already ended (message_stop or an upstream error event): an upstream that then @@ -216,12 +276,44 @@ export function tapAnthropicSseForLog( idle.pause(); if (settled) return; // stall/overflow/abort won the race while we awaited if (done) { + // Bun can resolve a cancelled read as done before the abort listener runs: a client + // that left is a cancel, not a cut upstream. + if (guard?.reqSignal?.aborted) { + onClientAbort(); + return; + } settled = true; idle.cancel(); detachAbort(); + const { terminalInTail, tailClean } = flushTail(); recordUsage(); - finalize(200, { closeReason: "terminal" }); - controller.close(); + // A clean EOF ends the turn only when the client already has its terminal, or a + // complete non-error stop_reason with nothing cut after it (compatible providers may + // omit message_stop; the translated adapter accepts the same shape). The opt-in + // anthropicEofTolerance profile keeps a content-only EOF as it always was. + const failure = terminalSeen + ? null + : lastDeliveredStop === "error" + ? 'upstream ended the turn with stop_reason "error"' + : !tailClean + ? "anthropic passthrough upstream stream ended before message_stop" + : lastDeliveredStop !== null || guard?.eofTolerant === true + ? null + : "anthropic passthrough upstream stream ended before message_stop"; + if (failure === null) { + finalize(200, { closeReason: "terminal" }); + if (terminalInTail) controller.enqueue(encoder.encode("\n\n")); + controller.close(); + return; + } + // A body that stopped short is a failed turn, logged and surfaced like a mid-stream + // reset (below), so the client never reads a partial answer as a finished message. + logCtx.transportPhase = "mid_stream"; + logCtx.terminalSource = "synthetic"; + logCtx.upstreamError = failure; + if (logCtx.activeAttempt) logCtx.activeAttempt.streamAborted = true; + finalize(502, { terminalStatus: "failed", closeReason: "terminal" }); + closeWithErrorFrame("api_error", failure); return; } if (value.byteLength > 0) { @@ -249,7 +341,7 @@ export function tapAnthropicSseForLog( settled = true; idle.cancel(); detachAbort(); - const terminalInTail = flushTail(); + const { terminalInTail } = flushTail(); recordUsage(); if (isTranslatorBudgetExceededError(err)) { // A local cap, not an upstream failure: the non-streaming native Messages fold diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 4c259199390..46dac508c2f 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -593,7 +593,7 @@ Native Anthropic passthrough in `src/server/claude-messages.ts` forwards the cal Native passthrough retains upstream `anthropic-ratelimit-*` response headers for Claude Code quota/statusLine consumers on SSE, JSON and upstream errors. `src/server/anthropic-rate-limit-headers.ts` selects only that family instead of copying all upstream headers, so cookies and unrelated metadata are not relayed. Missing rate-limit headers are not fabricated; body, status, content type and existing non-stream `Retry-After` behavior stay unchanged. `tests/claude-integration/claude-native-rate-limit-headers.test.ts` exercises the production ingress against a synthetic upstream for all four response shapes (SSE, JSON, upstream error and count_tokens). -`tapAnthropicSseForLog` in `src/server/claude-messages.ts` relays the streamed body of both the native passthrough and the managed native Messages lane (`src/server/messages-native.ts`). The response headers are already sent, so a stall, a byte-cap overflow, or an upstream read failure ends the body with an Anthropic `event: error` frame after a blank-line boundary and a clean close: `timeout_error` for an idle stall, `api_error` for the byte cap, and `api_error` when an upstream read fails mid-stream (a socket reset). The mid-stream reset is logged like the Responses relay's read error: status 502, `terminalStatus: "failed"`, `closeReason: "terminal"`, `transportPhase: "mid_stream"`, a synthetic terminal source, the attempt marked `streamAborted`, the redacted reason in `upstreamError`, and the usage seen before the reset. The non-streaming fold in the managed lane closes its row with the tap's meta for a reset, a stall or the byte cap, so its row matches the streaming lane's. The request is not replayed. Every one of these frames logs status 502. A stall is `terminalStatus: "incomplete"` with `closeReason: "body_stall"`, the status the Responses relay gives a stall-timeout incomplete. The byte cap is `incomplete` with `closeReason: "body_overflow"`, the same 502 the non-stream passthrough answers for it. Both carry the proxy's message in `upstreamError`, and a reset is `terminalStatus: "failed"`. Either way the row keeps its failure diagnostics in usage.jsonl. A stall or overflow after the turn's own terminal (`message_stop` or an upstream `error` event) is a finished turn: it logs 200 and closes without an error frame, as the read-error branch does. The tap finds frames in a copy normalized to LF, because SSE lines may end in CRLF, LF or CR; it holds a trailing CR until the next chunk so a split CRLF stays one line ending, and forwards the original bytes unchanged. Some read failures are not upstream failures. When the cancel signal is already aborted, the rejection is a `499` client cancel, because Bun can reject the read before it dispatches the abort listener. The managed lane passes its upstream controller's signal, so shutdown and turn release count as cancels too. When `message_stop` or an upstream `error` event has been seen, including one still in the buffer without its blank-line delimiter, the turn is complete: it logs 200 and closes with no error frame. A terminal found only in that unterminated tail gets its blank line restored, because an SSE parser drops an event that EOF cuts off. A translator budget overflow is a local cap, so it still errors the stream, and the non-streaming fold answers it with 413. `tests/claude-integration/claude-native-passthrough.test.ts` and `tests/claude-integration/messages-native.test.ts` cover both lanes against an upstream that resets after a partial or a complete body, plus both cancel paths. +`tapAnthropicSseForLog` in `src/server/messages-shared.ts` (re-exported by `src/server/claude-messages.ts`) relays the streamed body of both the native passthrough and the managed native Messages lane (`src/server/messages-native.ts`). The response headers are already sent, so a stall, a byte-cap overflow, or an upstream read failure ends the body with an Anthropic `event: error` frame after a blank-line boundary and a clean close: `timeout_error` for an idle stall, `api_error` for the byte cap, and `api_error` when an upstream read fails mid-stream (a socket reset). The mid-stream reset is logged like the Responses relay's read error: status 502, `terminalStatus: "failed"`, `closeReason: "terminal"`, `transportPhase: "mid_stream"`, a synthetic terminal source, the attempt marked `streamAborted`, the redacted reason in `upstreamError`, and the usage seen before the reset. The non-streaming fold in the managed lane closes its row with the tap's meta for a reset, a stall or the byte cap, so its row matches the streaming lane's. The request is not replayed. Every one of these frames logs status 502. A stall is `terminalStatus: "incomplete"` with `closeReason: "body_stall"`, the status the Responses relay gives a stall-timeout incomplete. The byte cap is `incomplete` with `closeReason: "body_overflow"`, the same 502 the non-stream passthrough answers for it. Both carry the proxy's message in `upstreamError`, and a reset is `terminalStatus: "failed"`. Either way the row keeps its failure diagnostics in usage.jsonl. A stall or overflow after the turn's own terminal (`message_stop` or an upstream `error` event) is a finished turn: it logs 200 and closes without an error frame, as the read-error branch does. The tap finds frames in a copy normalized to LF, because SSE lines may end in CRLF, LF or CR; it holds a trailing CR until the next chunk so a split CRLF stays one line ending, and forwards the original bytes unchanged. Some read failures are not upstream failures. When the cancel signal is already aborted, the rejection is a `499` client cancel, because Bun can reject the read before it dispatches the abort listener. The managed lane passes its upstream controller's signal, so shutdown and turn release count as cancels too. When `message_stop` or an upstream `error` event has been seen, including one still in the buffer without its blank-line delimiter, the turn is complete: it logs 200 and closes with no error frame. A terminal found only in that unterminated tail gets its blank line restored, because an SSE parser drops an event that EOF cuts off. A translator budget overflow is a local cap, so it still errors the stream, and the non-streaming fold answers it with 413. `tests/claude-integration/claude-native-passthrough.test.ts` and `tests/claude-integration/messages-native.test.ts` cover both lanes against an upstream that resets after a partial or a complete body, plus both cancel paths. A clean upstream EOF (no reset) ends the turn only when the client already has its terminal (`message_stop` or an upstream `error` event), or when the last message event it received as a complete SSE event was a `message_delta` with a non-error `stop_reason` and nothing but whitespace, comments or a ping follows it undelimited: compatible providers may omit `message_stop`, and the translated adapter accepts the same shape. Those bytes are relayed unchanged and log 200. Every other clean EOF is a cut turn and fails exactly like a mid-stream reset (502, `terminalStatus: "failed"`, `transportPhase: "mid_stream"`, `streamAborted`, an `api_error` frame): a partial body, an EOF after `stop_reason: "error"`, a stop reason left undelimited in the tail, or a message event cut off after an earlier stop reason (later content clears an earlier stop reason). A frame counts as evidence only when its payload `type` and its `event` name agree or the event line is absent, so a `ping`-named frame carrying a `message_stop` payload is not a terminal. In the managed lane, a routed provider with `anthropicEofTolerance` keeps a content-only EOF (no stop reason, nothing cut) as a finished turn, while caller-forward passthrough treats it as a failed turn; an error stop reason or a cut tail fails on both lanes. A done read after the cancel signal aborted is a 499 client cancel, the same rule as a rejected read. The non-streaming fold (`collectAnthropicMessage` in `src/claude/outbound.ts`) frames on a CRLF/CR-normalized copy and names a data-only frame by its payload type, so CRLF and data-only streams fold with their content. `tests/claude-integration/messages-native-clean-eof.test.ts` covers the tap's verdict for each EOF shape, the cancel race, the fold, and the managed lane end to end. A native passthrough answered without a stream records its reason in `upstreamError`, so the row and the failure diagnostics in usage.jsonl name the cause. An upstream error response (status 400 or above) is relayed verbatim; the stored diagnostic is `Provider error : ` only for a valid Anthropic error envelope with one of the closed types `invalid_request_error`, `authentication_error`, `permission_error`, `not_found_error`, `rate_limit_error`, `api_error`, `overloaded_error` or `request_too_large`. Upstream messages and arbitrary type strings never enter this diagnostic. Unknown or malformed envelopes, non-JSON bodies and bodies over 64 Ki characters log `Provider error `. Local header timeout, body stall, byte-cap overflow and cancel diagnostics contain fixed text plus validated guard limits. Fetch failures log the fixed reason `anthropic passthrough failed: upstream connection error`; the existing redacted client response is preserved. Classification uses these stored reasons and HTTP status, so a `permission_error` at 403 is `permission_denied` regardless of upstream message wording. `tests/claude-integration/claude-native-passthrough.test.ts` covers these cases and checks that echoed account identifiers and request content are absent from both history sinks. diff --git a/tests/claude-integration/claude-messages-endpoint.test.ts b/tests/claude-integration/claude-messages-endpoint.test.ts index 50152c6d0a3..3878e5de061 100644 --- a/tests/claude-integration/claude-messages-endpoint.test.ts +++ b/tests/claude-integration/claude-messages-endpoint.test.ts @@ -451,6 +451,7 @@ const MESSAGE_START_FRAME = 'event: message_start\ndata: {"type":"message_start" const UNSPACED_USAGE_FRAMES = [ 'event:message_start\ndata:{"type":"message_start","message":{"usage":{"input_tokens":11}}}\n\n', 'event:message_delta\ndata:{"type":"message_delta","usage":{"output_tokens":7}}\n\n', + 'event:message_stop\ndata:{"type":"message_stop"}\n\n', ].join(""); test("A0: usage extraction accepts unspaced data fields (#1170)", async () => { @@ -467,8 +468,8 @@ test("A0: usage extraction accepts unspaced data fields (#1170)", async () => { // The bytes pass through untouched either way; what the strict prefix broke was the inspection. expect(text).toContain("message_start"); - // "terminal" rather than "eof" is itself part of the fix: recognizing the unspaced - // `message_delta` is what lets the tap classify the close as a real terminal frame. + // A 200 terminal close depends on recognizing the unspaced frames: an unparsed + // `message_stop` would leave the turn looking cut short. expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]); expect(ctx.usage).toEqual(expect.objectContaining({ inputTokens: 11, outputTokens: 7 })); }); diff --git a/tests/claude-integration/messages-native-clean-eof.test.ts b/tests/claude-integration/messages-native-clean-eof.test.ts new file mode 100644 index 00000000000..c4bc01212d8 --- /dev/null +++ b/tests/claude-integration/messages-native-clean-eof.test.ts @@ -0,0 +1,283 @@ +/** + * A native Messages stream whose upstream ends cleanly (no reset) before the turn is over must + * reach the client as a failure it can see, never as a finished partial answer. Covers the shared + * stream tap (tapAnthropicSseForLog), the managed native lane end to end, and the non-stream fold. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { saveConfig } from "../../src/config"; +import { collectAnthropicMessage } from "../../src/claude/outbound"; +import { clearKeyCooldowns } from "../../src/providers/key-failover"; +import { handleClaudeMessages } from "../../src/server/claude-messages"; +import { tapAnthropicSseForLog } from "../../src/server/messages-shared"; +import { getRequestLogEntries, type RequestLogContext } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; + +const encoder = new TextEncoder(); +const ev = (event: string, data: object) => "event: " + event + "\ndata: " + JSON.stringify(data) + "\n\n"; +const start = ev("message_start", { type: "message_start", message: { id: "msg_eof", type: "message", role: "assistant", content: [], model: "claude-x", stop_reason: null, usage: { input_tokens: 5, output_tokens: 0 } } }); +const blockStart = ev("content_block_start", { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }); +const delta = ev("content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "half an ans" } }); +const blockStop = ev("content_block_stop", { type: "content_block_stop", index: 0 }); +const md = (stopReason: string) => ev("message_delta", { type: "message_delta", delta: { stop_reason: stopReason, stop_sequence: null }, usage: { output_tokens: 7 } }); +const stop = ev("message_stop", { type: "message_stop" }); +const ping = ev("ping", { type: "ping" }); +const partial = start + blockStart + delta; +const ERROR_FRAME = "\n\nevent: error\ndata: "; + +function streamOf(chunks: readonly string[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) controller.enqueue(encoder.encode(chunks[i++]!)); + else controller.close(); + }, + }); +} + +async function tap(raw: string | readonly string[], eofTolerant = false) { + const calls: Array<{ status: number; [key: string]: unknown }> = []; + const logCtx: RequestLogContext = { model: "claude-x", provider: "anth" }; + const out = tapAnthropicSseForLog(streamOf(typeof raw === "string" ? [raw] : raw), logCtx, + (status, meta) => calls.push({ status, ...meta }), { stallMs: 5_000, maxBytes: 0, ...(eofTolerant ? { eofTolerant } : {}) }); + return { text: await new Response(out).text(), calls, logCtx }; +} + +// Expected verdict for each clean-EOF shape: "ok" = bytes relayed unchanged and a 200 terminal; +// "fail" = an Anthropic error event after the relayed bytes and a failed 502 row. +const CASES: Array<[string, string, boolean, "ok" | "fail"]> = [ + ["complete turn", partial + blockStop + md("end_turn") + stop, false, "ok"], + ["upstream error event", partial + ev("error", { type: "error", error: { type: "overloaded_error", message: "busy" } }), false, "ok"], + ["partial content", partial, false, "fail"], + ["partial content with CRLF", partial.replace(/\n/g, "\r\n"), false, "fail"], + ["delivered end_turn without message_stop", partial + blockStop + md("end_turn"), false, "ok"], + ["delivered tool_use", start + md("tool_use"), false, "ok"], + ["delivered refusal", start + md("refusal"), false, "ok"], + ["delivered content_filter", start + md("content_filter"), false, "ok"], + ["delivered max_tokens", partial + md("max_tokens"), false, "ok"], + ["delivered stop_reason error", partial + md("error"), false, "fail"], + ["tolerant + delivered stop_reason error", partial + md("error"), true, "fail"], + ["tolerant + partial content", partial, true, "ok"], + ["tolerant + delivered end_turn", start + md("end_turn"), true, "ok"], + ["undelimited end_turn tail", partial + md("end_turn").slice(0, -2), false, "fail"], + ["tolerant + undelimited stop_reason error tail", partial + md("error").slice(0, -2), true, "fail"], + ["tolerant + undelimited end_turn tail", partial + md("end_turn").slice(0, -2), true, "fail"], + ["delivered end_turn then a cut event", start + md("end_turn") + 'event: content_block_delta\ndata: {"type":"content_bl', false, "fail"], + ["delivered end_turn then an undelimited keepalive comment", start + md("end_turn") + ": keepalive\n", false, "ok"], + ["delivered end_turn then an undelimited ping", start + md("end_turn") + ping.slice(0, -2), false, "ok"], + ["delivered end_turn then a ping", start + md("end_turn") + ping, false, "ok"], + ["ping-named message_delta is not a stop reason", partial + ev("ping", { type: "message_delta", delta: { stop_reason: "end_turn" } }), false, "fail"], + ["ping-named message_stop is not a terminal", partial + ev("ping", { type: "message_stop" }), false, "fail"], + ["ping-named error is not a terminal", partial + ev("ping", { type: "error", error: {} }), false, "fail"], + ["data-only stop reason", start + 'data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}\n\n', false, "ok"], + ["stop reason then resumed content", start + md("tool_use") + delta, false, "fail"], + ["tolerant + stop reason then resumed content", start + md("tool_use") + delta, true, "ok"], + ["delivered end_turn with CRLF", (partial + md("end_turn")).replace(/\n/g, "\r\n"), false, "ok"], + ["delivered end_turn with CR-only line endings", (partial + md("end_turn")).replace(/\n/g, "\r"), false, "ok"], + ["message_stop with CR-only line endings", (partial + md("end_turn") + stop).replace(/\n/g, "\r"), false, "ok"], + ["last event field wins: message_stop then ping", partial + 'event: message_stop\nevent: ping\ndata: {"type":"message_stop"}\n\n', false, "fail"], + ["last event field wins: ping then message_stop", partial + 'event: ping\nevent: message_stop\ndata: {"type":"message_stop"}\n\n', false, "ok"], + ["event name with a trailing space is another event", partial + 'event: message_stop \ndata: {"type":"message_stop"}\n\n', false, "fail"], + ["event name with an extra leading space is another event", partial + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', false, "fail"], + ["stop reason under a trailing-space event name does not count", partial + 'event: message_delta \ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n', false, "fail"], + ["data lines split inside a JSON string are invalid JSON", partial + 'event: message_stop\ndata: {"type":"message_\ndata: stop"}\n\n', false, "fail"], + ["data lines split at JSON whitespace still parse", partial + 'event: message_stop\ndata: {"type":\ndata: "message_stop"}\n\n', false, "ok"], + ["empty body", "", false, "fail"], +]; + +describe("clean upstream EOF in the native Messages stream tap", () => { + for (const [name, raw, tolerant, verdict] of CASES) { + test(name + " -> " + verdict, async () => { + const { text, calls, logCtx } = await tap(raw, tolerant); + if (verdict === "ok") { + expect(text).toBe(raw); + expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]); + return; + } + expect(text.startsWith(raw)).toBe(true); + const frame = text.slice(raw.length); + expect(frame.startsWith(ERROR_FRAME)).toBe(true); + expect(JSON.parse(frame.slice(ERROR_FRAME.length).trim())).toMatchObject({ type: "error", error: { type: "api_error" } }); + expect(calls).toEqual([{ status: 502, terminalStatus: "failed", closeReason: "terminal" }]); + expect(logCtx.transportPhase).toBe("mid_stream"); + expect(logCtx.terminalSource).toBe("synthetic"); + expect(logCtx.upstreamError).toBeTruthy(); + }); + } + + test("message_stop left undelimited at EOF still ends the turn and gets its blank line back", async () => { + const raw = partial + blockStop + md("end_turn") + stop.slice(0, -2); + const { text, calls } = await tap(raw); + expect(text).toBe(raw + "\n\n"); + expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]); + }); + + test("a CRLF split across reads is still one line ending", async () => { + const crlf = (partial + md("end_turn")).replace(/\n/g, "\r\n"); + const at = crlf.lastIndexOf("\r\n\r\n") + 1; + const { calls } = await tap([crlf.slice(0, at), crlf.slice(at)]); + expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]); + }); + + test("usage observed before the cut is kept on the failed turn", async () => { + const { logCtx, calls } = await tap(partial); + expect(calls[0]!.status).toBe(502); + expect(logCtx.usage).toEqual(expect.objectContaining({ inputTokens: 5 })); + }); + + test("a read that resolves done after the client aborted is a cancel, not a cut upstream", async () => { + const abort = new AbortController(); + let sent = false; + const source = new ReadableStream({ + pull(controller) { + if (sent) { abort.abort(); controller.close(); return; } + sent = true; + controller.enqueue(encoder.encode(partial)); + }, + }); + const calls: Array<{ status: number; [key: string]: unknown }> = []; + const logCtx: RequestLogContext = { model: "claude-x", provider: "anth" }; + // Keep the abort listener from winning: the done read must classify the cancel itself. + const reqSignal = { get aborted() { return abort.signal.aborted; }, reason: undefined, addEventListener() {}, removeEventListener() {} } as unknown as AbortSignal; + const text = await new Response(tapAnthropicSseForLog(source, logCtx, (status, meta) => calls.push({ status, ...meta }), { stallMs: 0, maxBytes: 0, reqSignal })).text(); + expect(text).toBe(partial); + expect(calls).toEqual([{ status: 499, closeReason: "client_cancel" }]); + }); +}); + +describe("non-stream fold of Messages SSE", () => { + const fold = (chunks: readonly string[]) => collectAnthropicMessage(streamOf(chunks), "claude-x", createTestTranslatorBudget()); + + test("CRLF frames fold with their content and stop reason", async () => { + const message = await fold([(partial + blockStop + md("tool_use") + stop).replace(/\n/g, "\r\n")]); + expect(message).toMatchObject({ type: "message", stop_reason: "tool_use", content: [{ type: "text", text: "half an ans" }] }); + }); + + test("a CRLF split across reads folds the same way", async () => { + const crlf = (partial + blockStop + md("end_turn") + stop).replace(/\n/g, "\r\n"); + const at = crlf.indexOf("\r\n\r\n") + 1; + const message = await fold([crlf.slice(0, at), crlf.slice(at)]); + expect(message).toMatchObject({ type: "message", content: [{ type: "text", text: "half an ans" }] }); + }); + + test("data-only frames are named by their payload type", async () => { + const dataOnly = (partial + blockStop + md("tool_use") + stop).replace(/^event: .*\n/gm, ""); + const message = await fold([dataOnly]); + expect(message).toMatchObject({ type: "message", stop_reason: "tool_use", content: [{ type: "text", text: "half an ans" }] }); + }); + + test("a final frame delimited by CR alone still folds", async () => { + const message = await fold([(partial + blockStop + md("tool_use")).replace(/\n/g, "\r")]); + expect(message).toMatchObject({ type: "message", stop_reason: "tool_use", content: [{ type: "text", text: "half an ans" }] }); + }); +}); + +// --- Managed native lane, end to end against a local fake upstream --- + +let upstream: ReturnType | undefined; +let releaseSpendHome: (() => void) | undefined; +let testDir = ""; +let previousHome: string | undefined; + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-messages-eof-")); + process.env.OPENCODEX_HOME = testDir; + clearKeyCooldowns(); +}); + +afterEach(async () => { + releaseSpendHome?.(); + releaseSpendHome = undefined; + await upstream?.stop(true); + upstream = undefined; + clearKeyCooldowns(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (testDir) removeTreeWithRetry(testDir); +}); + +/** An upstream that answers with the given SSE body and then ends it cleanly. */ +function cleanEofUpstream(body: string, eofTolerance = false): OcxConfig { + upstream = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) { + await req.json(); + return new Response(body, { headers: { "content-type": "text/event-stream" } }); + } }); + releaseSpendHome ??= acquireOwnedSpendHome(); + const config = { + port: 0, + defaultProvider: "anth", + providers: { anth: { + adapter: "anthropic", baseUrl: "http://127.0.0.1:" + upstream.port, authMode: "key", apiKey: "fixture-key-alpha", + allowPrivateNetwork: true, models: ["claude-x"], ...(eofTolerance ? { anthropicEofTolerance: true } : {}), + } }, + protocols: { rollout: { managedMessagesNative: true } }, + } as OcxConfig; + saveConfig(config); + return config; +} + +async function send(config: OcxConfig, stream: boolean) { + const requestId = "eof-" + crypto.randomUUID(); + const request = new Request("http://localhost/v1/messages", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "anth/claude-x", max_tokens: 64, stream, messages: [{ role: "user", content: "fixture question" }] }), + }); + const response = await handleClaudeMessages(request, config, { model: "", provider: "" }, { requestId, start: Date.now() }); + const text = await response.text(); + for (let i = 0; i < 100 && !getRequestLogEntries().some(entry => entry.requestId === requestId); i++) await Bun.sleep(10); + const rows = getRequestLogEntries().filter(entry => entry.requestId === requestId); + expect(rows).toHaveLength(1); + return { response, text, row: rows[0]! }; +} + +describe("managed native Messages lane with a clean upstream EOF", () => { + test("a streamed turn cut before message_stop ends with an Anthropic error event and a failed row", async () => { + const { response, text, row } = await send(cleanEofUpstream(partial), true); + expect(response.status).toBe(200); + expect(text).toContain("half an ans"); + expect(text).toContain(ERROR_FRAME); + expect(text).toContain("anthropic passthrough upstream stream ended before message_stop"); + expect(row.status).toBe(502); + expect(row.terminalStatus).toBe("failed"); + expect(row.transportPhase).toBe("mid_stream"); + expect(row.attempts?.at(-1)?.streamAborted).toBe(true); + expect(JSON.stringify(row)).not.toContain("fixture-key-alpha"); + }); + + test("a non-streaming caller gets an Anthropic 502, never the partial message", async () => { + const { response, text, row } = await send(cleanEofUpstream(partial), false); + expect(response.status).toBe(502); + expect(JSON.parse(text)).toMatchObject({ type: "error", error: { type: "api_error" } }); + expect(text).not.toContain("half an ans"); + expect(row.status).toBe(502); + }); + + test("a complete stop reason without message_stop is relayed as the finished turn", async () => { + const body = partial + blockStop + md("end_turn"); + const { response, text, row } = await send(cleanEofUpstream(body), true); + expect(response.status).toBe(200); + expect(text).not.toContain("event: error"); + expect(row.status).toBe(200); + }); + + test("an upstream error event stays the only terminal", async () => { + const body = partial + ev("error", { type: "error", error: { type: "overloaded_error", message: "busy" } }); + const { text } = await send(cleanEofUpstream(body), true); + expect(text.split("event: error").length - 1).toBe(1); + expect(text).toContain("overloaded_error"); + }); + + test("a provider with anthropicEofTolerance keeps a content-only EOF as it was", async () => { + const { response, text, row } = await send(cleanEofUpstream(partial, true), true); + expect(response.status).toBe(200); + expect(text).not.toContain("event: error"); + expect(row.status).toBe(200); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index c6fca6464eb..2723641562a 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -468,7 +468,7 @@ "claude-intercept-settings.test.ts": "claude-integration", "claude-intercept-no-proxy.test.ts": "claude-integration", "claude-management-api.test.ts": "claude-integration", "claude-manual-env.test.ts": "gui", "claude-messages-endpoint.test.ts": "claude-integration", "claude-native-fast-1m-ingress.test.ts": "claude-integration", - "claude-messages-thread.test.ts": "claude-integration", "messages-native.test.ts": "claude-integration", + "claude-messages-thread.test.ts": "claude-integration", "messages-native.test.ts": "claude-integration", "messages-native-clean-eof.test.ts": "claude-integration", "messages-native-decline-trace.test.ts": "claude-integration", "messages-native-oauth.test.ts": "claude-integration", "messages-native-opaque-state.test.ts": "claude-integration", "messages-surface-matrix.test.ts": "claude-integration", "claude-model-info.test.ts": "claude-integration", "claude-models-discovery.test.ts": "claude-integration",