feat(cohere): stream v2 tool arguments

API.md

@@ -332,6 +332,7 @@ composable integration value:
 | `streamfold/vercel-ai` | `vercelAI` |
 | `streamfold/openai` | `openAI` |
 | `streamfold/anthropic` | `anthropic` |
+| `streamfold/cohere` | `cohere` |
 | `streamfold/bedrock` | `bedrock` |
 | `streamfold/gemini` | `gemini` |
 | `streamfold/langchain` | `langchain` |
@@ -380,6 +381,14 @@ boundaries, `tool-input-end` continues to complete a call immediately.
 `tool-call` and UI `tool-input-available` events also work without preceding
 deltas. Repeated completion events do not emit duplicate results.
 
+For Cohere v2, pass `client.chatStream(...)` from `CohereClientV2` to
+`readStructured` with `integration: cohere`. Both the SDK's camelCase
+`toolCalls` payloads and decoded wire `tool_calls` payloads are accepted.
+Calls are correlated by index and expose the tool-call ID. They finish at
+`tool-call-end` or EOF. Text and tool-plan events are ignored; malformed JSON
+and generation error messages fail the integration. The Cohere SDK is not
+a runtime dependency of Streamfold.
+
 For Amazon Bedrock, pass the `stream` returned by `ConverseStreamCommand` to
 `readStructured` with `integration: bedrock`. The adapter correlates tool-use
 blocks by `contentBlockIndex` and exposes `toolUseId` as the call ID. Text,

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Add a Cohere v2 adapter for SDK and decoded wire tool argument streams.
 - Add a Bedrock ConverseStream adapter for streaming tool-use arguments.
 - Accept Anthropic stable and beta SDK message event interfaces directly in
   the Anthropic adapter and managed reader.

README.md

@@ -79,6 +79,7 @@ Streamfold includes isolated adapters for decoded SDK events:
 | `streamfold/vercel-ai` | Vercel AI SDK |
 | `streamfold/openai` | OpenAI Responses |
 | `streamfold/anthropic` | Anthropic Messages |
+| `streamfold/cohere` | Cohere v2 Chat |
 | `streamfold/bedrock` | Amazon Bedrock ConverseStream |
 | `streamfold/gemini` | Gemini Interactions |
 | `streamfold/langchain` | LangChain message chunks |

benchmarks/cohere-conformance.test.mjs

@@ -0,0 +1,28 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { CohereClientV2 } from "cohere-ai";
+import { readStructured } from "../packages/core/src/index.js";
+import { cohere } from "../packages/core/src/cohere.js";
+
+test("reads interleaved arguments from Cohere SDK camelCase SSE events", async () => {
+  const events = [
+    { type: "message-start", id: "fixture", delta: { message: { role: "assistant" } } },
+    { type: "tool-plan-delta", delta: { message: { tool_plan: "Check both cities" } } },
+    ...["paris", "tokyo"].map((id, index) => ({ type: "tool-call-start", index, delta: { message: { tool_calls: { id, type: "function", function: { name: "weather", arguments: "" } } } } })),
+    { type: "tool-call-delta", index: 0, delta: { message: { tool_calls: { function: { arguments: '{"city":"Pa' } } } } },
+    { type: "tool-call-delta", index: 1, delta: { message: { tool_calls: { function: { arguments: '{"city":"Tokyo"}' } } } } },
+    { type: "tool-call-delta", index: 0, delta: { message: { tool_calls: { function: { arguments: 'ris"}' } } } } },
+    { type: "tool-call-end", index: 1 }, { type: "tool-call-end", index: 0 },
+    { type: "message-end", delta: { finish_reason: "TOOL_CALL" } },
+  ];
+  const client = new CohereClientV2({
+    token: "test-token",
+    fetch: async () => new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), { headers: { "Content-Type": "text/event-stream" } }),
+  });
+  const source = await client.chatStream({ model: "command-a", messages: [{ role: "user", content: "Weather?" }] });
+  const updates = [];
+  for await (const update of readStructured(source, { integration: cohere, limits: { snapshots: "immutable" } })) updates.push(update);
+  assert.deepEqual(updates.map(({ type, id }) => [type, id]), [["start", "paris"], ["start", "tokyo"], ["update", "paris"], ["update", "tokyo"], ["update", "paris"], ["complete", "tokyo"], ["complete", "paris"]]);
+  assert.deepEqual(updates[2].partialValue, { city: "Pa" });
+  assert.deepEqual(updates.slice(-2).map(({ value }) => value), [{ city: "Tokyo" }, { city: "Paris" }]);
+});

examples/tsconfig.json

@@ -5,5 +5,5 @@
     "checkJs": true,
     "skipLibCheck": true
   },
-  "files": ["assistant-ui.mjs", "vercel-ai.mjs", "custom.mjs", "../tests/langchain-sdk.ts", "../tests/gemini-sdk.ts", "../tests/openai-sdk.ts", "../tests/anthropic-sdk.ts", "../tests/bedrock-sdk.ts"]
+  "files": ["assistant-ui.mjs", "vercel-ai.mjs", "custom.mjs", "../tests/langchain-sdk.ts", "../tests/gemini-sdk.ts", "../tests/openai-sdk.ts", "../tests/anthropic-sdk.ts", "../tests/bedrock-sdk.ts", "../tests/cohere-sdk.ts"]
 }

package.json

@@ -38,6 +38,7 @@
     "ai": "7.0.152",
     "assistant-stream": "0.3.42",
     "bumpp": "^12.3.0",
+    "cohere-ai": "8.3.0",
     "fast-check": "4.10.2",
     "openai": "7.45.1",
     "playwright": "^1.63.0",

packages/core/package.json

@@ -51,6 +51,10 @@
     "./vercel-ai": {
       "types": "./src/vercel-ai.d.ts",
       "default": "./src/vercel-ai.js"
+    },
+    "./cohere": {
+      "types": "./src/cohere.d.ts",
+      "default": "./src/cohere.js"
     }
   },
   "engines": {

packages/core/src/cohere.d.ts

@@ -0,0 +1,30 @@
+import type {
+  BatchEventStructuredStream,
+  BatchStructuredStreamIntegration,
+  StructuredStreamIntegrationOptions,
+  StructuredStreamPool,
+} from "./index.js";
+
+type ToolCall = {
+  readonly id?: string;
+  readonly function?: { readonly arguments?: string };
+};
+
+export type CohereToolCallEvent =
+  | {
+      readonly type: "tool-call-start" | "tool-call-delta";
+      readonly index?: number;
+      readonly delta?: {
+        readonly message?: { readonly toolCalls?: ToolCall; readonly tool_calls?: ToolCall };
+      };
+    }
+  | { readonly type: "tool-call-end"; readonly index?: number }
+  | { readonly type: string }
+  | { readonly type: string; readonly [key: string]: unknown };
+
+export const cohere: BatchStructuredStreamIntegration<CohereToolCallEvent, string>;
+
+export function createStructuredStream(
+  pool?: StructuredStreamPool<string>,
+  options?: StructuredStreamIntegrationOptions,
+): BatchEventStructuredStream<CohereToolCallEvent, string>;

packages/core/src/cohere.js

@@ -0,0 +1,47 @@
+import { createStructuredStreamPool } from "./index.js";
+import { createIntegration } from "./internal/integration.js";
+
+export const cohere = (pool = createStructuredStreamPool(), options) => {
+  const idsByIndex = new Map();
+  return createIntegration(
+    pool,
+    (event, calls) => {
+      if (event.type === "message-end" && event.delta?.error) throw new Error(event.delta.error);
+      const message = event.delta?.message;
+      const tool = message?.toolCalls ?? message?.tool_calls;
+      if (event.type === "tool-call-start") {
+        if (event.index === undefined) {
+          calls.unmatched("Tool call start has no index");
+          return undefined;
+        }
+        const id = tool?.id;
+        if (id !== undefined) idsByIndex.set(event.index, id);
+        return calls.start(id, tool?.function?.arguments ?? "");
+      }
+      if (event.type === "tool-call-delta") {
+        const id = idsByIndex.get(event.index);
+        if (id === undefined) {
+          calls.unmatched("Arguments delta has no matching tool call index");
+          return undefined;
+        }
+        return calls.push(id, tool?.function?.arguments ?? "");
+      }
+      if (event.type === "tool-call-end") {
+        const id = idsByIndex.get(event.index);
+        if (id !== undefined) {
+          idsByIndex.delete(event.index);
+          return calls.complete(id);
+        }
+      }
+      return undefined;
+    },
+    (complete) => {
+      for (const id of idsByIndex.values()) complete(id);
+      idsByIndex.clear();
+    },
+    "cohere",
+    options,
+  );
+};
+
+export { cohere as createStructuredStream };

packages/core/src/cohere.test.js

@@ -0,0 +1,44 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { createStructuredStreamPool } from "./index.js";
+import { cohere } from "./cohere.js";
+
+const start = (index, id, args = "") => ({ type: "tool-call-start", index, delta: { message: { tool_calls: { id, function: { arguments: args } } } } });
+const delta = (index, args) => ({ type: "tool-call-delta", index, delta: { message: { tool_calls: { function: { arguments: args } } } } });
+const end = (index) => ({ type: "tool-call-end", index });
+
+test("Cohere handles decoded wire arguments and finalizes each call once", () => {
+  const stream = cohere();
+  stream.push(start(0, "first", '{"city":'));
+  stream.push(delta(0, '"Paris"}'));
+  assert.deepEqual(stream.pushAll(end(0)).map(({ id, value }) => ({ id, value })), [{ id: "first", value: { city: "Paris" } }]);
+  assert.deepEqual(stream.pushAll(end(0)), []);
+  stream.push(start(0, "second", "{}"));
+  assert.deepEqual(stream.finish().map(({ id, value }) => ({ id, value })), [{ id: "first", value: { city: "Paris" } }, { id: "second", value: {} }]);
+});
+
+test("Cohere releases active calls on truncated arguments, limits, or provider errors", () => {
+  for (const [options, input, ending, expected] of [
+    [{}, '{"city":', end(0), /complete|incomplete/i],
+    [{ maxBytes: 4 }, '{"city":"Paris"}', end(0), /maxBytes/],
+    [{}, '{}', { type: "message-end", delta: { error: "generation failed" } }, /generation failed/],
+  ]) {
+    const pool = createStructuredStreamPool(options);
+    const stream = cohere(pool);
+    stream.push(start(0, "first"));
+    stream.push(start(1, "second"));
+    assert.throws(() => { stream.push(delta(0, input)); stream.push(ending); }, expected);
+    assert.equal(pool.size, 0);
+    assert.throws(() => stream.finish(), expected);
+  }
+});
+
+test("Cohere ignores text and diagnoses uncorrelated argument events", () => {
+  const diagnostics = [];
+  const stream = cohere(undefined, { onDiagnostic: (value) => diagnostics.push(value) });
+  assert.deepEqual(stream.pushAll({ type: "content-delta", delta: { message: { content: { text: "Hello" } } } }), []);
+  stream.push(delta(9, "{}"));
+  stream.push({ type: "tool-call-start", delta: { message: { tool_calls: { id: "missing-index" } } } });
+  assert.deepEqual(stream.finish(), []);
+  assert.equal(diagnostics.filter(({ code }) => code === "UNMATCHED_TOOL_EVENT").length, 2);
+});