fix(claude): fail native Messages streams that end before message_stop

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",

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();

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)

src/server/messages-shared.ts

@@ -52,13 +52,43 @@ 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;
   /** Cumulative body byte cap. 0 disables. */
   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