diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts
index 7f4d0dbd994..45eb39e9585 100644
--- a/src/server/claude-messages.ts
+++ b/src/server/claude-messages.ts
@@ -14,6 +14,7 @@ import {
   resolveAdmissionModelScope,
 } from "./admission-model-scope";
 import { jsonUtf8Bytes } from "../lib/json-byte-size";
+import { redactSecretString } from "../lib/redact";
 import { sseFieldValue } from "../lib/sse-decoder";
 import { enforceAnthropicImageLimits, sniffImageDimensions } from "../adapters/anthropic-image-guard";
 import { normalizeAnthropicImages } from "../adapters/anthropic-image-normalize";
@@ -295,46 +296,55 @@ export interface PassthroughBodyGuard {
 }
 
 type PassthroughCloseReason = "terminal" | "client_cancel" | "body_stall" | "body_overflow";
+type PassthroughFinalizeMeta = { closeReason: PassthroughCloseReason; terminalStatus?: "failed" };
 
 /**
  * Tap an Anthropic-vocabulary SSE stream for the request log (usage + terminal),
  * bounding body occupancy: idle (silence-only, timed ONLY while a reader.read() is
  * pending so downstream backpressure never counts as upstream inactivity) and a
- * cumulative byte cap. On stall/overflow it appends a protocol-compatible Anthropic
- * `event: error` terminal frame after a blank-line boundary, closes, and cancels the
- * upstream reader — never a total-wall-clock bound (slow-but-alive streams live).
+ * cumulative byte cap. On stall/overflow, or when the upstream read fails after the
+ * headers went out, it appends a protocol-compatible Anthropic `event: error` terminal
+ * frame after a blank-line boundary, closes, and cancels the upstream reader — never a
+ * total-wall-clock bound (slow-but-alive streams live).
  * Exported for deterministic unit tests.
  */
 export function tapAnthropicSseForLog(
   upstream: ReadableStream<Uint8Array>,
   logCtx: RequestLogContext,
-  finalize: (status: number, meta: { closeReason: PassthroughCloseReason }) => void,
+  finalize: (status: number, meta: PassthroughFinalizeMeta) => void,
   guard?: PassthroughBodyGuard,
 ): ReadableStream<Uint8Array> {
   const decoder = new TextDecoder();
   const encoder = new TextEncoder();
   let buffer = "";
   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")
+      .map(l => sseFieldValue(l, "data"))
+      .filter((v): v is string => v !== null)
+      .join("");
+    if (!dataLine) return;
+    let data: unknown;
+    try { data = JSON.parse(dataLine); } catch { return; }
+    if (!isRec(data)) return;
+    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") {
+      terminalSeen = true;
+    }
+  };
   const inspect = (chunk: Uint8Array) => {
     buffer += decoder.decode(chunk, { stream: true });
     let sep: number;
     while ((sep = buffer.indexOf("\n\n")) !== -1) {
       const frame = buffer.slice(0, sep);
       buffer = buffer.slice(sep + 2);
-      const dataLine = frame
-        .split("\n")
-        .map(l => sseFieldValue(l, "data"))
-        .filter((v): v is string => v !== null)
-        .join("");
-      if (!dataLine) continue;
-      let data: unknown;
-      try { data = JSON.parse(dataLine); } catch { continue; }
-      if (!isRec(data)) continue;
-      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 };
-      }
+      inspectFrame(frame);
     }
   };
   const reader = upstream.getReader();
@@ -345,13 +355,7 @@ export function tapAnthropicSseForLog(
   const recordUsage = () => {
     logCtx.usage = anthropicUsageToOcx(Object.keys(usageAcc).length > 0 ? usageAcc : undefined);
   };
-  const failBody = (closeReason: "body_stall" | "body_overflow", errType: string, message: string) => {
-    if (settled) return;
-    settled = true;
-    idle.cancel();
-    detachAbort();
-    recordUsage();
-    finalize(200, { closeReason });
+  const closeWithErrorFrame = (errType: string, message: string) => {
     const payload = JSON.stringify({ type: "error", error: { type: errType, message } });
     try {
       // Leading blank line terminates any partial SSE block so the frame parses cleanly
@@ -359,6 +363,15 @@ export function tapAnthropicSseForLog(
       tapController?.enqueue(encoder.encode(`\n\nevent: error\ndata: ${payload}\n\n`));
       tapController?.close();
     } catch { /* client already torn down */ }
+  };
+  const failBody = (closeReason: "body_stall" | "body_overflow", errType: string, message: string) => {
+    if (settled) return;
+    settled = true;
+    idle.cancel();
+    detachAbort();
+    recordUsage();
+    finalize(200, { closeReason });
+    closeWithErrorFrame(errType, message);
     reader.cancel(new DOMException(message, closeReason === "body_stall" ? "TimeoutError" : "QuotaExceededError")).catch(() => {});
   };
   const idle = idleDeadline(guard?.stallMs ?? 0, () => {
@@ -427,12 +440,56 @@ export function tapAnthropicSseForLog(
         controller.enqueue(value);
       } catch (err) {
         if (settled) return;
+        // Bun can settle a fetch body read before it dispatches the abort listeners
+        // (consumeForInspection in relay.ts): read the signal itself before calling this
+        // rejection an upstream failure.
+        if (guard?.reqSignal?.aborted) {
+          onClientAbort();
+          return;
+        }
         settled = true;
         idle.cancel();
         detachAbort();
+        // A read error can follow the last SSE block before its blank-line delimiter. Count
+        // that block before deciding how the turn ended, as the Responses relay does.
+        const terminalBeforeTail = terminalSeen;
+        const tail = buffer + decoder.decode();
+        buffer = "";
+        if (tail) inspectFrame(tail);
         recordUsage();
-        finalize(200, { closeReason: "terminal" });
-        try { controller.error(err); } catch { /* torn down */ }
+        if (isTranslatorBudgetExceededError(err)) {
+          // A local cap, not an upstream failure: the non-streaming native Messages fold
+          // maps this error to a 413 itself, so it still errors the stream.
+          finalize(200, { closeReason: "terminal" });
+          try { controller.error(err); } catch { /* torn down */ }
+          return;
+        }
+        if (terminalSeen) {
+          // Only the transport trailer was lost; the client already has the turn's terminal.
+          // The Responses relay likewise reports a read error only without a seen terminal.
+          finalize(200, { closeReason: "terminal" });
+          try {
+            // An SSE parser drops an event that EOF cuts off before its blank line, so restore
+            // the delimiter when the terminal was only found in that unterminated tail.
+            if (!terminalBeforeTail) controller.enqueue(encoder.encode("\n\n"));
+            controller.close();
+          } catch { /* torn down */ }
+          reader.cancel(err).catch(() => {});
+          return;
+        }
+        // The upstream read failed after the 200 went out (a mid-stream socket reset). Log it
+        // the way the Responses relay does (onReadError): a truncated body is a failed turn,
+        // not a completed one. The client gets the same Anthropic error terminal as a stall,
+        // instead of a connection reset — or, on some Bun releases, a bare EOF that reads as
+        // a finished message.
+        const message = redactSecretString(`anthropic passthrough upstream stream failed: ${err instanceof Error ? err.message : String(err)}`);
+        logCtx.transportPhase = "mid_stream";
+        logCtx.terminalSource = "synthetic";
+        logCtx.upstreamError = message.slice(0, 500);
+        if (logCtx.activeAttempt) logCtx.activeAttempt.streamAborted = true;
+        finalize(502, { terminalStatus: "failed", closeReason: "terminal" });
+        closeWithErrorFrame("api_error", message);
+        reader.cancel(err).catch(() => {});
       }
     },
     cancel(reason) {
diff --git a/src/server/messages-native.ts b/src/server/messages-native.ts
index 5a88dc50aa3..d03b6a33299 100644
--- a/src/server/messages-native.ts
+++ b/src/server/messages-native.ts
@@ -150,7 +150,7 @@ export interface HandleNativeMessagesOptions {
   callerAnthropicBeta?: string | null;
 }
 
-type FinishLog = (status: number, message?: string, closeReason?: FinalRequestLogMeta["closeReason"]) => void;
+type FinishLog = (status: number, message?: string, meta?: FinalRequestLogMeta) => void;
 
 /** Relay an upstream body unchanged, recording first output on its first non-empty chunk. */
 function observeFirstChunk(body: ReadableStream<Uint8Array>, onFirst: () => void): ReadableStream<Uint8Array> {
@@ -311,10 +311,10 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
   attemptHandle.seal(logCtx.accountLogLabel);
   const { attempt } = attemptHandle;
   const finalLog = createFinalRequestLog(logIds, logCtx);
-  const finishLog: FinishLog = (status, message, closeReason = "non_stream") => {
+  const finishLog: FinishLog = (status, message, meta = { closeReason: "non_stream" }) => {
     if (finalLog.finished()) return;
     if (message) logCtx.upstreamError = redactSecretString(message).slice(0, 500);
-    finalLog.finish(status, { closeReason });
+    finalLog.finish(status, meta);
   };
   const bindUsage = (usage: OcxUsage | undefined) => {
     if (!usage) return;
@@ -597,7 +597,9 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
 
   const contentType = response.headers.get("content-type")?.toLowerCase() ?? "";
   if (contentType.includes("text/event-stream") && response.body) {
-    const bodyGuard = resolvePassthroughBodyGuard(config, req.signal);
+    // `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 observed = logIds ? observeFirstChunk(response.body, () => recordFirstOutput(logCtx, logIds.start)) : response.body;
     const renamed = activeRequest.oauthToolNames
       ? restoreOAuthToolNamesInSse(observed, activeRequest.oauthToolNames, translatorBudget)
@@ -609,7 +611,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
         try {
           cleanupAbort();
           bindUsage(logCtx.usage);
-          finishLog(status, undefined, meta.closeReason);
+          finishLog(status, undefined, meta);
           if (meta.closeReason !== "terminal") upstream.abort();
         } finally {
           releaseStreamTurn();
@@ -621,9 +623,10 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
       });
     }
     // A non-streaming caller whose upstream streamed anyway: fold the stream into one message.
-    const tapState: { closeReason?: FinalRequestLogMeta["closeReason"] } = {};
+    const tapState: { closeReason?: FinalRequestLogMeta["closeReason"]; meta?: FinalRequestLogMeta } = {};
     const tapped = tapAnthropicSseForLog(source, logCtx, (_status, meta) => {
       tapState.closeReason = meta.closeReason;
+      tapState.meta = meta;
     }, bodyGuard);
     try {
       const message = await collectAnthropicMessage(tapped, requestedModel, translatorBudget);
@@ -634,7 +637,12 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
       bindUsage(logCtx.usage);
       if (message.type === "error") {
         const error = isRec(message.error) ? message.error : {};
-        return fail(502, typeof error.message === "string" ? error.message : "upstream stream failed", "api_error");
+        const text = typeof error.message === "string" ? error.message : "upstream stream failed";
+        // The tap's own verdict (a reset, a stall, the byte cap) closes this row with the same meta
+        // as the streaming lane's row; a plain upstream error event keeps the non_stream row.
+        const tapMeta = tapState.meta;
+        if (tapMeta && (tapMeta.terminalStatus || tapMeta.closeReason !== "terminal")) finishLog(502, text, tapMeta);
+        return fail(502, text, "api_error");
       }
       finishLog(200);
       return Response.json(message);
diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md
index a01466ab96b..e256089a2a8 100644
--- a/structure/clients/claude-desktop.md
+++ b/structure/clients/claude-desktop.md
@@ -462,4 +462,8 @@ The [compaction routing override](../transports/responses-failover.md#compaction
 
 Native Anthropic passthrough in `src/server/claude-messages.ts` forwards the caller's body except for tool-call ids: `sanitizePassthroughToolCallIds` runs the request-scoped allocator from `src/adapters/tool-call-id.ts` over every `*tool_use` id and `*tool_result` `tool_use_id`. Conforming ids are reserved first and stay byte-identical, a non-conforming or overlength id is rewritten to a conforming id of at most 64 characters with call/result pairing kept, and an empty id throws `AnthropicRequestError`, so the request fails with a local 400 before the upstream fetch. `tests/claude-integration/claude-native-passthrough.test.ts` covers rewriting, pairing, the empty id, the overlength id and collision with an existing valid id.
 
+## Native passthrough stream terminals
+
+`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. The logged status is not uniform across these frames: a stall and the byte cap keep status 200 with `closeReason` `body_stall` or `body_overflow`, which classify as `incomplete`, while a reset is a 502 that classifies as `failed`. 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.
+
 Linked-machine data uses the [connection-bound relay contract](../remote-link.md#connection-bound-relay-authentication); client-local credentials and routing policy remain unchanged.
diff --git a/tests/claude-integration/claude-native-passthrough.test.ts b/tests/claude-integration/claude-native-passthrough.test.ts
index fe7b1ccb20f..c01425f744e 100644
--- a/tests/claude-integration/claude-native-passthrough.test.ts
+++ b/tests/claude-integration/claude-native-passthrough.test.ts
@@ -11,6 +11,10 @@ import { startServer } from "../../src/server";
 import type { OcxConfig } from "../../src/types";
 import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home";
 import { removeTreeWithRetry } from "../helpers/remove-tree";
+import { startTruncatedSseUpstream } from "../helpers/truncated-sse-upstream";
+import { tapAnthropicSseForLog } from "../../src/server/claude-messages";
+import type { RequestLogContext } from "../../src/server/request-log";
+import { TranslatorBudgetExceededError } from "../../src/lib/translator-budget";
 
 let testDir = "";
 let previousHome: string | undefined;
@@ -788,3 +792,179 @@ test("an overlength id is rewritten within 64 characters and a colliding valid i
     upstream.stop(true);
   }
 });
+
+// --- Mid-stream upstream reset: the stream had started, then the upstream socket went away ---
+
+const PARTIAL_TURN_SSE = [
+  `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_up", type: "message", role: "assistant", content: [], model: "claude-fable-5", stop_reason: null, usage: { input_tokens: 12, output_tokens: 1 } } })}\n\n`,
+  `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } })}\n\n`,
+  `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "half an ans" } })}\n\n`,
+].join("");
+
+test("a mid-stream upstream reset ends the native stream with an Anthropic error event and logs a failed turn", async () => {
+  const { clearRequestLogsForTests } = await import("../../src/server/request-log");
+  clearRequestLogsForTests();
+  const upstream = startTruncatedSseUpstream(PARTIAL_TURN_SSE);
+  saveConfig(cfg(`http://127.0.0.1:${upstream.port}`));
+  const server = startServer(0);
+  try {
+    const res = await fetch(new URL("/v1/messages?beta=true", server.url), {
+      method: "POST",
+      headers: OAUTH_HEADERS,
+      body: JSON.stringify(claudeBody()),
+    });
+    expect(res.status).toBe(200);
+    // The body ends cleanly with a protocol terminal the client can act on, instead of a
+    // connection reset (or, on some Bun releases, a bare chunked EOF) after "half an ans".
+    const text = await res.text();
+    expect(text).toContain("half an ans");
+    expect(text).toContain("\n\nevent: error\ndata: ");
+    const errorFrame = JSON.parse(text.slice(text.lastIndexOf("data: ") + 6).trim()) as { type: string; error: { type: string; message: string } };
+    expect(errorFrame.type).toBe("error");
+    expect(errorFrame.error.type).toBe("api_error");
+    expect(errorFrame.error.message).toContain("anthropic passthrough upstream stream failed: ");
+    // The committed request is not replayed.
+    expect(upstream.requests()).toBe(1);
+
+    const logs = logsFromApiBody<{
+      status?: number;
+      terminalStatus?: string;
+      closeReason?: string;
+      transportPhase?: string;
+      terminalSource?: string;
+      failureCause?: string;
+      upstreamError?: string;
+      usage?: { inputTokens?: number };
+    }>(await (await fetch(new URL("/api/logs?tail=1", server.url))).json());
+    expect(logs).toHaveLength(1);
+    const row = logs[0]!;
+    // Same row the Responses relay writes for a mid-stream reset: a truncated 200 body is not a
+    // completed turn.
+    expect(row.status).toBe(502);
+    expect(row.terminalStatus).toBe("failed");
+    expect(row.closeReason).toBe("terminal");
+    expect(row.transportPhase).toBe("mid_stream");
+    expect(row.terminalSource).toBe("synthetic");
+    expect(row.failureCause).toBe("transport-ambiguous");
+    expect(row.upstreamError).toContain("anthropic passthrough upstream stream failed: ");
+    // Usage seen before the reset is still recorded.
+    expect(row.usage?.inputTokens).toBe(12);
+  } finally {
+    await server.stop(true);
+    upstream.stop();
+  }
+});
+
+test("a translator budget overflow still errors the tapped stream for callers that map it", async () => {
+  const overflow = new TranslatorBudgetExceededError("live_transient", 1024);
+  let sent = false;
+  const source = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (sent) {
+        controller.error(overflow);
+        return;
+      }
+      sent = true;
+      controller.enqueue(new TextEncoder().encode(PARTIAL_TURN_SSE));
+    },
+  });
+  const calls: unknown[] = [];
+  const logCtx: RequestLogContext = { model: "claude-fable-5", provider: "anthropic-native" };
+  const tapped = tapAnthropicSseForLog(source, logCtx, (status, meta) => calls.push({ status, ...meta }), { stallMs: 5_000, maxBytes: 0 });
+  // The non-streaming native Messages fold turns this error into a 413; an error frame would
+  // have reached it as a generic 502 instead.
+  await expect(new Response(tapped).text()).rejects.toBe(overflow);
+  expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]);
+  expect(logCtx.transportPhase).toBeUndefined();
+});
+
+test("a read rejection that lands before the client abort listener still finalizes as a client cancel", async () => {
+  // Bun can settle a fetch body read before it dispatches the abort listeners (see
+  // consumeForInspection in src/server/relay.ts). Model that order: the signal is already
+  // aborted when the read rejects, and its listener has not run.
+  const signal = { aborted: false, reason: undefined as unknown, addEventListener() {}, removeEventListener() {} };
+  let sent = false;
+  const source = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (sent) {
+        signal.aborted = true;
+        signal.reason = new DOMException("client went away", "AbortError");
+        controller.error(signal.reason);
+        return;
+      }
+      sent = true;
+      controller.enqueue(new TextEncoder().encode(PARTIAL_TURN_SSE));
+    },
+  });
+  const calls: unknown[] = [];
+  const logCtx: RequestLogContext = { model: "claude-fable-5", provider: "anthropic-native" };
+  const tapped = tapAnthropicSseForLog(source, logCtx, (status, meta) => calls.push({ status, ...meta }), {
+    stallMs: 5_000,
+    maxBytes: 0,
+    reqSignal: signal as unknown as AbortSignal,
+  });
+  const text = await new Response(tapped).text();
+  expect(text).not.toContain("event: error");
+  expect(calls).toEqual([{ status: 499, closeReason: "client_cancel" }]);
+  expect(logCtx.transportPhase).toBeUndefined();
+  expect(logCtx.upstreamError).toBeUndefined();
+});
+
+const COMPLETE_TURN_SSE = PARTIAL_TURN_SSE + [
+  `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n`,
+  `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 5 } })}\n\n`,
+  `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`,
+].join("");
+
+test("a reset after message_stop is a finished turn: no error event and a completed row", async () => {
+  const { clearRequestLogsForTests } = await import("../../src/server/request-log");
+  clearRequestLogsForTests();
+  // Only the chunked-encoding trailer is lost; the turn itself arrived whole.
+  const upstream = startTruncatedSseUpstream(COMPLETE_TURN_SSE);
+  saveConfig(cfg(`http://127.0.0.1:${upstream.port}`));
+  const server = startServer(0);
+  try {
+    const res = await fetch(new URL("/v1/messages?beta=true", server.url), {
+      method: "POST",
+      headers: OAUTH_HEADERS,
+      body: JSON.stringify(claudeBody()),
+    });
+    const text = await res.text();
+    expect(text.endsWith(`event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`)).toBe(true);
+    expect(text).not.toContain("event: error");
+    const logs = logsFromApiBody<{ status?: number; closeReason?: string; transportPhase?: string; upstreamError?: string }>(
+      await (await fetch(new URL("/api/logs?tail=1", server.url))).json(),
+    );
+    expect(logs).toHaveLength(1);
+    expect(logs[0]).toMatchObject({ status: 200, closeReason: "terminal" });
+    expect(logs[0]!.transportPhase).toBeUndefined();
+    expect(logs[0]!.upstreamError).toBeUndefined();
+  } finally {
+    await server.stop(true);
+    upstream.stop();
+  }
+});
+
+test("a terminal frame still in the buffer when the read fails counts as seen", async () => {
+  // The reset can land after message_stop but before its blank-line delimiter.
+  const withoutDelimiter = COMPLETE_TURN_SSE.slice(0, -2);
+  let sent = false;
+  const source = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (sent) {
+        controller.error(new Error("The socket connection was closed unexpectedly."));
+        return;
+      }
+      sent = true;
+      controller.enqueue(new TextEncoder().encode(withoutDelimiter));
+    },
+  });
+  const calls: unknown[] = [];
+  const logCtx: RequestLogContext = { model: "claude-fable-5", provider: "anthropic-native" };
+  const tapped = tapAnthropicSseForLog(source, logCtx, (status, meta) => calls.push({ status, ...meta }), { stallMs: 5_000, maxBytes: 0 });
+  const text = await new Response(tapped).text();
+  // The delimiter is restored: an SSE parser drops an event that EOF cuts off before its blank line.
+  expect(text).toBe(`${withoutDelimiter}\n\n`);
+  expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]);
+  expect(logCtx.usage).toEqual(expect.objectContaining({ inputTokens: 12, outputTokens: 5 }));
+});
diff --git a/tests/claude-integration/messages-native.test.ts b/tests/claude-integration/messages-native.test.ts
index 37b19deddaf..8deaabc5e68 100644
--- a/tests/claude-integration/messages-native.test.ts
+++ b/tests/claude-integration/messages-native.test.ts
@@ -14,8 +14,10 @@ import { clearKeyCooldowns } from "../../src/providers/key-failover";
 import { estimateClaudeRequestTokens, handleClaudeCountTokens, handleClaudeMessages } from "../../src/server/claude-messages";
 import { getRequestLogEntries } from "../../src/server/request-log";
 import type { OcxConfig } from "../../src/types";
+import type { AdmissionLease } from "../../src/lib/admission";
 import { acquireOwnedSpendHome } from "../helpers/owned-spend-home";
 import { removeTreeWithRetry } from "../helpers/remove-tree";
+import { startTruncatedSseUpstream } from "../helpers/truncated-sse-upstream";
 
 interface Seen {
   path: string;
@@ -158,6 +160,36 @@ async function send(config: OcxConfig, body: Record<string, unknown>, headers?:
   return { requestId, response, text };
 }
 
+/** Streams the first frames of a turn, then stays open until the request is aborted. */
+function hangingAfterPartialTurn(): Response {
+  return new Response(new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(new TextEncoder().encode(sseText(SSE_FRAMES.slice(0, 3))));
+    },
+  }), { headers: { "content-type": "text/event-stream" } });
+}
+
+async function drain(reader: ReadableStreamDefaultReader<Uint8Array>): Promise<string> {
+  const decoder = new TextDecoder();
+  let text = "";
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) return text;
+      text += decoder.decode(value, { stream: true });
+    }
+  } catch {
+    return text;
+  }
+}
+
+async function settledRowFor(requestId: string) {
+  for (let i = 0; i < 100 && !getRequestLogEntries().some(entry => entry.requestId === requestId); i++) {
+    await Bun.sleep(10);
+  }
+  return rowFor(requestId);
+}
+
 describe("managed native Messages", () => {
   test("sends exactly the allowlisted source body with the provider key and no caller credential", async () => {
     const config = fixtureConfig(startUpstream());
@@ -254,6 +286,102 @@ describe("managed native Messages", () => {
     expect(callerForwardSeen[0]!.headers.get("x-api-key")).toBe("sk-ant-fixture-caller");
   });
 
+  test("a mid-stream upstream reset ends the relayed stream with an Anthropic error event and a failed row", async () => {
+    const truncated = startTruncatedSseUpstream(sseText(SSE_FRAMES.slice(0, 3)));
+    try {
+      const config = fixtureConfig(truncated.port);
+      const { requestId, response, text } = await send(config, { ...SOURCE_BODY, stream: true });
+      expect(response.status).toBe(200);
+      expect(text).toContain("streamed");
+      expect(text).toContain("\n\nevent: error\ndata: ");
+      expect(text).toContain('"type":"api_error"');
+      expect(text).toContain("anthropic passthrough upstream stream failed: ");
+      expect(truncated.requests()).toBe(1);
+      const row = rowFor(requestId);
+      expect(row.status).toBe(502);
+      expect(row.terminalStatus).toBe("failed");
+      expect(row.closeReason).toBe("terminal");
+      expect(row.transportPhase).toBe("mid_stream");
+      expect(row.terminalSource).toBe("synthetic");
+      expect(row.failureCause).toBe("transport-ambiguous");
+      expect(row.attempts?.at(-1)?.streamAborted).toBe(true);
+      expect(row.upstreamError).toContain("anthropic passthrough upstream stream failed: ");
+      expect(JSON.stringify(row)).not.toContain("fixture-key-alpha");
+    } finally {
+      truncated.stop();
+    }
+  });
+
+  test("a non-streaming caller whose upstream stream resets still gets an Anthropic 502", async () => {
+    const truncated = startTruncatedSseUpstream(sseText(SSE_FRAMES.slice(0, 3)));
+    try {
+      const config = fixtureConfig(truncated.port);
+      const { requestId, response, text } = await send(config, { ...SOURCE_BODY, stream: false });
+      expect(response.status).toBe(502);
+      expect(JSON.parse(text)).toMatchObject({ type: "error", error: { type: "api_error" } });
+      expect(text).toContain("anthropic passthrough upstream stream failed: ");
+      const row = rowFor(requestId);
+      // Same row as the streaming lane for the same reset.
+      expect(row.status).toBe(502);
+      expect(row.terminalStatus).toBe("failed");
+      expect(row.closeReason).toBe("terminal");
+      expect(row.transportPhase).toBe("mid_stream");
+      expect(row.failureCause).toBe("transport-ambiguous");
+    } finally {
+      truncated.stop();
+    }
+  });
+
+  test("a non-streaming fold that hits the body byte cap keeps the tap's close reason", async () => {
+    const config = fixtureConfig(startUpstream(() => new Response(SSE_TEXT, { headers: { "content-type": "text/event-stream" } })),
+      { claudeCode: { bodyMaxBytes: 64 } as OcxConfig["claudeCode"] });
+    const { requestId, response, text } = await send(config, { ...SOURCE_BODY, stream: false });
+    expect(response.status).toBe(502);
+    expect(text).toContain("exceeded 64 bytes");
+    const row = rowFor(requestId);
+    expect(row.status).toBe(502);
+    expect(row.closeReason).toBe("body_overflow");
+  });
+
+  test("a client that disconnects mid-stream is logged as a cancel, not an upstream failure", async () => {
+    const config = fixtureConfig(startUpstream(hangingAfterPartialTurn));
+    const client = new AbortController();
+    const requestId = `pf08-${crypto.randomUUID()}`;
+    const response = await handleClaudeMessages(new Request("http://localhost/v1/messages", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ ...SOURCE_BODY, stream: true }),
+      signal: client.signal,
+    }), config, { model: "", provider: "" }, { requestId, start: Date.now() });
+    const reader = response.body!.getReader();
+    expect((await reader.read()).done).toBe(false);
+    client.abort(new DOMException("client went away", "AbortError"));
+    await drain(reader);
+    const row = await settledRowFor(requestId);
+    expect(row.status).toBe(499);
+    expect(row.closeReason).toBe("client_cancel");
+    expect(row.transportPhase).toBeUndefined();
+  });
+
+  test("the lane aborting its own upstream (shutdown, turn release) is logged as a cancel", async () => {
+    const config = fixtureConfig(startUpstream(hangingAfterPartialTurn));
+    let turn: AbortController | undefined;
+    const lease = { bindAbortController(controller: AbortController) { turn = controller; }, release() {} };
+    const requestId = `pf08-${crypto.randomUUID()}`;
+    const response = await handleClaudeMessages(messagesRequest({ ...SOURCE_BODY, stream: true }), config,
+      { model: "", provider: "" }, { requestId, start: Date.now(), turnAdmissionLease: lease as unknown as AdmissionLease });
+    const reader = response.body!.getReader();
+    expect((await reader.read()).done).toBe(false);
+    expect(turn).toBeDefined();
+    turn!.abort(new Error("server shutdown"));
+    const text = await drain(reader);
+    expect(text).not.toContain("event: error");
+    const row = await settledRowFor(requestId);
+    expect(row.status).toBe(499);
+    expect(row.closeReason).toBe("client_cancel");
+    expect(row.upstreamError).toBeUndefined();
+  });
+
   test("count_tokens counts the body the native lane sends", async () => {
     const config = fixtureConfig(startUpstream());
     await send(config, { ...SOURCE_BODY, stream: false });
diff --git a/tests/helpers/truncated-sse-upstream.ts b/tests/helpers/truncated-sse-upstream.ts
new file mode 100644
index 00000000000..d4ca914035b
--- /dev/null
+++ b/tests/helpers/truncated-sse-upstream.ts
@@ -0,0 +1,47 @@
+/**
+ * A fake upstream that answers any request with HTTP 200 `text/event-stream`, writes the given SSE
+ * text as one chunk, and then closes the socket without the terminating zero-length chunk.
+ *
+ * That is a mid-stream reset as the proxy's `fetch` sees it: the response headers arrived, part of
+ * the body arrived, and the next `reader.read()` rejects with ECONNRESET ("The socket connection
+ * was closed unexpectedly"). `Bun.serve` cannot produce it deterministically — an errored response
+ * body is sent as a connection reset by one Bun release and as a clean chunked EOF by another —
+ * so the fixture speaks HTTP/1.1 over a raw socket instead.
+ */
+export function startTruncatedSseUpstream(sse: string): { port: number; requests: () => number; stop: () => void } {
+  const body = new TextEncoder().encode(sse);
+  const head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n";
+  const answered = new WeakSet<object>();
+  const timers = new Set<ReturnType<typeof setTimeout>>();
+  let requests = 0;
+  const listener = Bun.listen({
+    hostname: "127.0.0.1",
+    port: 0,
+    socket: {
+      data(socket) {
+        if (answered.has(socket)) return;
+        answered.add(socket);
+        requests += 1;
+        socket.write(`${head}${body.byteLength.toString(16)}\r\n`);
+        socket.write(body);
+        socket.write("\r\n");
+        // Give the partial body time to reach the reader before the reset: ending the socket
+        // right after the write lets Bun 1.3 fail the fetch before the headers are read.
+        const timer = setTimeout(() => {
+          timers.delete(timer);
+          socket.end();
+        }, 20);
+        timers.add(timer);
+      },
+    },
+  });
+  return {
+    port: listener.port,
+    requests: () => requests,
+    stop: () => {
+      for (const timer of timers) clearTimeout(timer);
+      timers.clear();
+      listener.stop(true);
+    },
+  };
+}