fix(anthropic): preserve terminal SSE frames

docs-site/src/content/docs/reference/adapters.md

@@ -62,7 +62,8 @@ streams the response back **untranslated**.
   output headroom, and **drops `temperature`/`top_p`** when thinking is enabled (Anthropic forbids
   them there).
 - Always sends `anthropic-version: 2023-06-01`. Streams `content_block_delta` (`text_delta`,
-  `thinking_delta`, `input_json_delta`).
+  `thinking_delta`, compatible `reasoning_delta`, `input_json_delta`). The SSE decoder preserves
+  event state across fetch chunks and accepts a terminal `message_stop` without a trailing newline.
 
 ## `google`
 

src/adapters/anthropic.ts

@@ -21,6 +21,7 @@ import { normalizeAnthropicImages } from "./anthropic-image-normalize";
 import { neutralizeIdentity } from "./identity";
 import { CLAUDE_CODE_HEADERS, claudeCodeSessionId } from "./client-fingerprint";
 import { buildNonOpenAIToolCatalogNudgeForTools } from "./tool-catalog-nudge";
+import { decodeServerSentEvents } from "../lib/sse-decoder";
 
 /** Map a user content part to an Anthropic content block (text or image source). */
 function toAnthropicContentPart(p: OcxContentPart): unknown {
@@ -719,9 +720,6 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
         return;
       }
 
-      const reader = response.body.getReader();
-      const decoder = new TextDecoder();
-      let buffer = "";
       let currentBlockType = "";
       let currentToolCallId = "";
       let currentToolCallName = "";
@@ -739,34 +737,19 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
         };
       };
 
-      try {
-        while (true) {
-          const { done, value } = await reader.read();
-          if (done) break;
-          buffer += decoder.decode(value, { stream: true });
+      for await (const record of decodeServerSentEvents(response.body)) {
+        const payload = record.data.trim();
+        if (!payload) continue;
 
-          const lines = buffer.split("\n");
-          buffer = lines.pop() ?? "";
-
-          let currentEventType = "";
-          for (const line of lines) {
-            if (line.startsWith("event: ")) {
-              currentEventType = line.slice(7).trim();
-              continue;
-            }
-            if (!line.startsWith("data: ")) continue;
-            const payload = line.slice(6).trim();
-            if (!payload) continue;
-
-            let data: Record<string, unknown>;
-            try {
-              data = JSON.parse(payload) as Record<string, unknown>;
-            } catch {
-              debugDroppedFrame("anthropic", payload);
-              continue;
-            }
+        let data: Record<string, unknown>;
+        try {
+          data = JSON.parse(payload) as Record<string, unknown>;
+        } catch {
+          debugDroppedFrame("anthropic", payload);
+          continue;
+        }
 
-            switch (currentEventType || data.type) {
+        switch (record.event || data.type) {
               case "message_start": {
                 const message = data.message as { usage?: Record<string, number> } | undefined;
                 pendingUsage = mergeAnthropicUsage(pendingUsage, message?.usage);
@@ -794,7 +777,12 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
                   yield { type: "text_delta", text: delta.text };
                 } else if (delta.type === "thinking_delta" && typeof delta.thinking === "string") {
                   yield { type: "thinking_delta", thinking: delta.thinking };
-                } else if (delta.type === "signature_delta" && typeof delta.signature === "string" && currentBlockType === "thinking") {
+                } else if (delta.type === "reasoning_delta" && typeof delta.reasoning === "string") {
+                  // Some Anthropic-compatible reasoning models use `reasoning` names for the
+                  // otherwise equivalent thinking block. Preserve it as raw reasoning and keep
+                  // later text blocks independent.
+                  yield { type: "thinking_delta", thinking: delta.reasoning };
+                } else if (delta.type === "signature_delta" && typeof delta.signature === "string" && (currentBlockType === "thinking" || currentBlockType === "reasoning")) {
                   // Arrives once, just before the thinking block's content_block_stop; block-scoped
                   // so a stray signature on a non-thinking block can never be captured.
                   yield { type: "thinking_signature", signature: delta.signature };
@@ -827,20 +815,15 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
                 yield { type: "error", message: err?.message ?? "Anthropic error" };
                 return;
               }
-            }
-            currentEventType = "";
-          }
         }
-        if (pendingUsage && !emittedDone) yield* emitDone();
-      } finally {
-        reader.releaseLock();
       }
+      if (pendingUsage && !emittedDone) yield* emitDone();
     },
 
     async parseResponse(response: Response): Promise<AdapterEvent[]> {
       const json = await response.json() as Record<string, unknown>;
       const events: AdapterEvent[] = [];
-      const content = json.content as { type: string; text?: string; id?: string; name?: string; input?: unknown; thinking?: string; signature?: string; data?: string }[] | undefined;
+      const content = json.content as { type: string; text?: string; id?: string; name?: string; input?: unknown; thinking?: string; reasoning?: string; signature?: string; data?: string }[] | undefined;
       if (content) {
         for (const block of content) {
           if (block.type === "text" && block.text) {
@@ -850,6 +833,8 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
             if (typeof block.signature === "string" && block.signature) {
               events.push({ type: "thinking_signature", signature: block.signature });
             }
+          } else if (block.type === "reasoning" && typeof block.reasoning === "string") {
+            events.push({ type: "thinking_delta", thinking: block.reasoning });
           } else if (block.type === "redacted_thinking" && typeof block.data === "string") {
             events.push({ type: "redacted_thinking", data: block.data });
           } else if (block.type === "tool_use") {

src/lib/sse-decoder.ts

@@ -0,0 +1,75 @@
+export interface ServerSentEvent {
+  event?: string;
+  data: string;
+}
+
+/**
+ * Decode text/event-stream records across arbitrary fetch chunk boundaries.
+ *
+ * The final record is dispatched at EOF even when the upstream omits the trailing blank line or
+ * final newline. That matters for compatible APIs that place a terminal event in the last bytes of
+ * the body: dropping that record turns a successful response into an adapter_eof failure.
+ */
+export async function* decodeServerSentEvents(
+  source: ReadableStream<Uint8Array>,
+): AsyncGenerator<ServerSentEvent> {
+  const reader = source.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  let event: string | undefined;
+  let dataLines: string[] = [];
+
+  const dispatch = (): ServerSentEvent | undefined => {
+    if (dataLines.length === 0) {
+      event = undefined;
+      return undefined;
+    }
+    const record = { ...(event ? { event } : {}), data: dataLines.join("\n") };
+    event = undefined;
+    dataLines = [];
+    return record;
+  };
+
+  const acceptLine = (rawLine: string): ServerSentEvent | undefined => {
+    const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
+    if (line === "") return dispatch();
+    if (line.startsWith(":")) return undefined;
+
+    const colon = line.indexOf(":");
+    const field = colon < 0 ? line : line.slice(0, colon);
+    let value = colon < 0 ? "" : line.slice(colon + 1);
+    if (value.startsWith(" ")) value = value.slice(1);
+
+    if (field === "event") event = value;
+    else if (field === "data") dataLines.push(value);
+    return undefined;
+  };
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (value) buffer += decoder.decode(value, { stream: !done });
+
+      let newline: number;
+      while ((newline = buffer.indexOf("\n")) >= 0) {
+        const record = acceptLine(buffer.slice(0, newline));
+        buffer = buffer.slice(newline + 1);
+        if (record) yield record;
+      }
+
+      if (!done) continue;
+      buffer += decoder.decode();
+      if (buffer.length > 0) {
+        const record = acceptLine(buffer);
+        buffer = "";
+        if (record) yield record;
+      }
+      const finalRecord = dispatch();
+      if (finalRecord) yield finalRecord;
+      break;
+    }
+  } finally {
+    try { await reader.cancel(); } catch { /* already closed/errored */ }
+    try { reader.releaseLock(); } catch { /* already released */ }
+  }
+}

tests/anthropic-compatible-stream.test.ts

@@ -0,0 +1,112 @@
+import { describe, expect, test } from "bun:test";
+import { createAnthropicAdapter } from "../src/adapters/anthropic";
+import { bridgeToResponsesSSE } from "../src/bridge";
+import { responsesSseToAnthropicSse } from "../src/claude/outbound";
+import type { AdapterEvent, OcxProviderConfig } from "../src/types";
+
+const provider = {
+  adapter: "anthropic",
+  baseUrl: "https://api.kimi.com/coding",
+  apiKey: "test-key",
+  authMode: "key",
+} as OcxProviderConfig;
+
+const kimiCompatibleSse = [
+  'event: message_start\ndata: {"type":"message_start","message":{}}',
+  'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"reasoning","reasoning":""}}',
+  'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"reasoning_delta","reasoning":"think"}}',
+  'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
+  'event: content_block_start\ndata: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}',
+  'event: content_block_delta\ndata: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"visible"}}',
+  'event: content_block_stop\ndata: {"type":"content_block_stop","index":1}',
+  // Deliberately no usage snapshot and no newline after the terminal data line.
+  'event: message_stop\ndata: {"type":"message_stop"}',
+].join("\n\n");
+
+function arbitrarilyChunkedResponse(text = kimiCompatibleSse): Response {
+  const encoder = new TextEncoder();
+  const bytes = encoder.encode(text);
+  const cuts = [3, 19, 47, 101, 173, 251, 337, 419, bytes.length];
+  return new Response(new ReadableStream<Uint8Array>({
+    start(controller) {
+      let offset = 0;
+      for (const end of cuts) {
+        if (end <= offset) continue;
+        controller.enqueue(bytes.slice(offset, Math.min(end, bytes.length)));
+        offset = end;
+        if (offset >= bytes.length) break;
+      }
+      controller.close();
+    },
+  }));
+}
+
+async function collectAdapterEvents(response: Response): Promise<AdapterEvent[]> {
+  const events: AdapterEvent[] = [];
+  for await (const event of createAnthropicAdapter(provider).parseStream(response)) events.push(event);
+  return events;
+}
+
+describe("Anthropic-compatible reasoning stream termination (#312)", () => {
+  test("preserves reasoning and visible text, then emits done from final message_stop", async () => {
+    const events = await collectAdapterEvents(arbitrarilyChunkedResponse());
+
+    expect(events).toContainEqual({ type: "thinking_delta", thinking: "think" });
+    expect(events).toContainEqual({ type: "text_delta", text: "visible" });
+    expect(events.at(-1)).toEqual({ type: "done", usage: undefined });
+  });
+
+  test("Responses bridge completes instead of reporting adapter_eof", async () => {
+    const responses = bridgeToResponsesSSE(
+      createAnthropicAdapter(provider).parseStream(arbitrarilyChunkedResponse()),
+      "kimi/k3",
+      undefined,
+      undefined,
+      undefined,
+      undefined,
+      0,
+    );
+    const text = await new Response(responses).text();
+
+    expect(text).toContain("response.reasoning_summary_text.delta");
+    expect(text).toContain("response.output_text.delta");
+    expect(text).toContain("response.completed");
+    expect(text).not.toContain("response.incomplete");
+    expect(text).toContain("visible");
+  });
+
+  test("Claude Messages translation keeps content blocks and message_stop", async () => {
+    const responses = bridgeToResponsesSSE(
+      createAnthropicAdapter(provider).parseStream(arbitrarilyChunkedResponse()),
+      "kimi/k3",
+      undefined,
+      undefined,
+      undefined,
+      undefined,
+      0,
+    );
+    const anthropic = responsesSseToAnthropicSse(responses, "k3", { pingIntervalMs: 0 });
+    const text = await new Response(anthropic).text();
+
+    expect(text).toContain('"type":"thinking_delta","thinking":"think"');
+    expect(text).toContain('"type":"text_delta","text":"visible"');
+    expect(text).toContain("event: message_stop");
+  });
+
+  test("non-streaming compatible reasoning blocks map without hiding later text", async () => {
+    const events = await createAnthropicAdapter(provider).parseResponse(new Response(JSON.stringify({
+      content: [
+        { type: "reasoning", reasoning: "think" },
+        { type: "text", text: "visible" },
+      ],
+      usage: { input_tokens: 2, output_tokens: 3 },
+      stop_reason: "end_turn",
+    })));
+
+    expect(events).toEqual([
+      { type: "thinking_delta", thinking: "think" },
+      { type: "text_delta", text: "visible" },
+      { type: "done", usage: { inputTokens: 2, outputTokens: 3 }, stopReason: "end_turn" },
+    ]);
+  });
+});

tests/sse-decoder.test.ts

@@ -0,0 +1,48 @@
+import { describe, expect, test } from "bun:test";
+import { decodeServerSentEvents } from "../src/lib/sse-decoder";
+
+function chunkedStream(chunks: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    start(controller) {
+      for (const chunk of chunks) controller.enqueue(encoder.encode(chunk));
+      controller.close();
+    },
+  });
+}
+
+async function collect(chunks: string[]) {
+  const records = [];
+  for await (const record of decodeServerSentEvents(chunkedStream(chunks))) records.push(record);
+  return records;
+}
+
+describe("text/event-stream decoder", () => {
+  test("preserves event state across arbitrary reader chunks", async () => {
+    expect(await collect([
+      "event: content_",
+      "block_delta\n",
+      "data: {\"type\":\"content_block_delta\"}\n\n",
+    ])).toEqual([{
+      event: "content_block_delta",
+      data: '{"type":"content_block_delta"}',
+    }]);
+  });
+
+  test("dispatches the terminal record without a final newline or blank delimiter", async () => {
+    expect(await collect([
+      "event: message_stop\n",
+      'data: {"type":"message_stop"}',
+    ])).toEqual([{
+      event: "message_stop",
+      data: '{"type":"message_stop"}',
+    }]);
+  });
+
+  test("joins multiline data and accepts CRLF framing", async () => {
+    expect(await collect([
+      "event: custom\r\ndata: first\r\n",
+      "data: second\r\n\r\n",
+    ])).toEqual([{ event: "custom", data: "first\nsecond" }]);
+  });
+});