refactor(channels): move message-format projections into Channels

.changeset/channels-ai-sdk-connectors.md

@@ -2,4 +2,4 @@
 "agents": minor
 ---
 
-Add `AiSdkHarness` from the new `agents/harness/ai-sdk` entry point, which runs each message with `streamText` and keeps sessions and transcripts in the Durable Object, so `Channels.forHarness` can serve an AI SDK model. The same entry point exports the AI SDK message conversions it is built on and `createSendMessageTool`, an AI SDK tool that sends a message to a surface through the gateway.
+Add `AiSdkHarness` from the new `agents/harness/ai-sdk` entry point, which runs each message with `streamText` and keeps sessions and transcripts in the Durable Object, so `Channels.forHarness` can serve an AI SDK model. The same entry point exports `createSendMessageTool`, an AI SDK tool that sends a message to a surface through the gateway.

.changeset/channels-projections.md

@@ -0,0 +1,5 @@
+---
+"agents": minor
+---
+
+Add projections, which map a harness vendor's message format to and from Channels transcript messages and response chunks, one entry point per vendor. `agents/experimental/channels/projections/ai-sdk` holds the AI SDK conversions (`toResponseChunks`, `toTranscriptMessage`, `toUIMessageChunk`, `toUIMessage`). `agents/experimental/channels/projections/pi` projects pi-durable entries and live events. `agents/harness/pi` gains `piChannelsHarness`, which puts `PiHarness` behind the shared harness interface so `Channels.forHarness` can serve it.

examples/next/channels/README.md

@@ -71,11 +71,12 @@ conversation is also reachable directly at
 
 ## The pi agent
 
-`PiHarness` keeps pi's own shape. `src/pi/channels-harness.ts`,
-`piChannelsHarness(harness, { kv })`, is the harness adapter that puts it
-behind the shared harness interface. It only translates shapes: pi entries
-become transcript messages, pi events become session events, and pi
-submission ids become the caller's operation ids. `Channels.forHarness`
+`PiHarness` keeps pi's own shape. `piChannelsHarness(harness, { kv })`,
+also from `agents/harness/pi`, is the harness adapter that puts it behind
+the shared harness interface: pi events become session events, and pi
+submission ids become the caller's operation ids. pi entries and messages
+become transcript messages and response chunks through the pi projection
+(`agents/experimental/channels/projections/pi`). `Channels.forHarness`
 knows nothing about pi:
 
 ```ts

examples/next/channels/src/pi/agent.ts

@@ -13,9 +13,8 @@ import {
 import { WebChannel } from "agents/experimental/channels/web";
 import { Lifecycle } from "agents/lifecycle";
 import { createModels } from "@earendil-works/pi-ai/models";
-import { PiHarness } from "agents/harness/pi";
+import { PiHarness, piChannelsHarness } from "agents/harness/pi";
 import { createAI } from "agents/models/pi-ai";
-import { piChannelsHarness } from "./channels-harness";
 
 const MODEL_ID = "@cf/moonshotai/kimi-k2.7-code";
 

packages/agents/package.json

@@ -422,6 +422,16 @@
       "import": "./dist/experimental/channels/web/ai-sdk.js",
       "require": "./dist/experimental/channels/web/ai-sdk.js"
     },
+    "./experimental/channels/projections/ai-sdk": {
+      "types": "./dist/experimental/channels/projections/ai-sdk.d.ts",
+      "import": "./dist/experimental/channels/projections/ai-sdk.js",
+      "require": "./dist/experimental/channels/projections/ai-sdk.js"
+    },
+    "./experimental/channels/projections/pi": {
+      "types": "./dist/experimental/channels/projections/pi.d.ts",
+      "import": "./dist/experimental/channels/projections/pi.js",
+      "require": "./dist/experimental/channels/projections/pi.js"
+    },
     "./tsconfig": "./agents.tsconfig.json",
     "./vite": {
       "types": "./dist/vite.d.ts",

packages/agents/scripts/build.ts

@@ -55,7 +55,9 @@ const entries = [
   "src/experimental/channels/telegram/index.ts",
   "src/experimental/channels/web/index.ts",
   "src/experimental/channels/web/client.ts",
-  "src/experimental/channels/web/ai-sdk.ts"
+  "src/experimental/channels/web/ai-sdk.ts",
+  "src/experimental/channels/projections/ai-sdk.ts",
+  "src/experimental/channels/projections/pi.ts"
 ];
 
 for (const entry of entries) {

packages/agents/src/experimental/channels/__tests__/ai-sdk-projection.test.ts

@@ -0,0 +1,132 @@
+import type { UIMessage, UIMessageChunk } from "ai";
+import { describe, expect, it } from "vitest";
+import {
+  toResponseChunks,
+  toTranscriptMessage,
+  toUIMessage,
+  toUIMessageChunk
+} from "../projections/ai-sdk";
+import type { ResponseChunk } from "../protocol";
+import { responseWriter } from "../response";
+
+/** Append every chunk to a response, which throws on a grammar error. */
+function record(chunks: readonly ResponseChunk[]): ResponseChunk[] {
+  const recorded: ResponseChunk[] = [];
+  const response = responseWriter({
+    streamId: "response",
+    cursor: 0,
+    append: (chunk) => recorded.push(chunk as unknown as ResponseChunk),
+    close: () => {},
+    error: () => {}
+  });
+  for (const chunk of chunks) response.append(chunk);
+  response.end();
+  return recorded;
+}
+
+async function collect(
+  chunks: readonly UIMessageChunk[]
+): Promise<ResponseChunk[]> {
+  async function* stream(): AsyncGenerator<UIMessageChunk> {
+    yield* chunks;
+  }
+  const out: ResponseChunk[] = [];
+  for await (const chunk of toResponseChunks(stream())) out.push(chunk);
+  return out;
+}
+
+const generation: UIMessageChunk[] = [
+  { type: "start" },
+  { type: "start-step" },
+  { type: "reasoning-start", id: "r" },
+  { type: "reasoning-delta", id: "r", delta: "Look it up" },
+  { type: "reasoning-end", id: "r" },
+  { type: "text-start", id: "t" },
+  { type: "text-delta", id: "t", delta: "Checking" },
+  { type: "text-end", id: "t" },
+  { type: "tool-input-start", toolCallId: "c", toolName: "weather" },
+  { type: "tool-input-delta", toolCallId: "c", inputTextDelta: '{"city":' },
+  {
+    type: "tool-input-available",
+    toolCallId: "c",
+    toolName: "weather",
+    input: { city: "Lisbon" }
+  },
+  { type: "tool-output-available", toolCallId: "c", output: { temp: 21 } },
+  { type: "data-progress", id: "p", data: { done: 1 } },
+  { type: "finish-step" },
+  { type: "finish", messageMetadata: { tokens: 12 } }
+];
+
+describe("AI SDK projection", () => {
+  it("projects a generation into grammatical response chunks", async () => {
+    const chunks = await collect(generation);
+    expect(record(chunks)).toEqual(chunks);
+    expect(chunks.map((chunk) => chunk.type)).toEqual([
+      "reasoning-start",
+      "reasoning-delta",
+      "reasoning-end",
+      "text-start",
+      "text-delta",
+      "text-end",
+      "tool-input-start",
+      "tool-input-delta",
+      "tool-input-available",
+      "tool-output-available",
+      "data",
+      "metadata"
+    ]);
+  });
+
+  it("projects response chunks back into the UI chunks they came from", async () => {
+    const chunks = await collect(generation);
+    expect(chunks.map(toUIMessageChunk)).toEqual([
+      ...generation.slice(2, 13),
+      { type: "message-metadata", messageMetadata: { tokens: 12 } }
+    ]);
+  });
+
+  it("round-trips a saved UI message through a transcript message", () => {
+    const message = {
+      id: "m1",
+      role: "assistant",
+      metadata: { model: "test" },
+      parts: [
+        { type: "reasoning", text: "Look it up" },
+        { type: "text", text: "Checking" },
+        {
+          type: "tool-weather",
+          toolCallId: "c",
+          state: "output-available",
+          input: { city: "Lisbon" },
+          output: { temp: 21 }
+        },
+        {
+          type: "dynamic-tool",
+          toolName: "search",
+          toolCallId: "d",
+          state: "output-error",
+          input: { q: "x" },
+          errorText: "offline"
+        },
+        { type: "data-progress", id: "p", data: { done: 1 } }
+      ]
+    } as UIMessage;
+    expect(toUIMessage(toTranscriptMessage(message))).toEqual(message);
+  });
+
+  it("throws after an aborted stream, keeping what it produced", async () => {
+    async function* stream(): AsyncGenerator<UIMessageChunk> {
+      yield { type: "text-start", id: "a" };
+      yield { type: "text-delta", id: "a", delta: "Half" };
+      yield { type: "abort" };
+    }
+    const seen: string[] = [];
+    await expect(async () => {
+      for await (const chunk of toResponseChunks(stream())) {
+        seen.push(chunk.type);
+      }
+    }).rejects.toThrow("aborted");
+    expect(seen).toEqual(["text-start", "text-delta"]);
+  });
+});

packages/agents/src/experimental/channels/__tests__/ai-sdk.test.ts

@@ -1,11 +1,7 @@
-import { asSchema, type UIMessage, type UIMessageChunk } from "ai";
+import { asSchema, type UIMessage } from "ai";
 import { describe, expect, it, vi } from "vitest";
 import { ChannelGateway, type ChannelMessage, type DeliveryResult } from "..";
-import {
-  answerToolCall,
-  createSendMessageTool,
-  toResponseChunks
-} from "../../../harness/ai-sdk";
+import { answerToolCall, createSendMessageTool } from "../../../harness/ai-sdk";
 
 function executable(tool: ReturnType<typeof createSendMessageTool>) {
   return tool.execute as unknown as (
@@ -36,7 +32,7 @@ function host(
   };
 }
 
-describe("AI SDK message adapter", () => {
+describe("AI SDK send-message tool", () => {
   it("adapts a Host-resolved surface to a caller-described tool", async () => {
     const deliver = vi.fn(
       async (): Promise<DeliveryResult> => ({
@@ -84,7 +80,7 @@ describe("AI SDK message adapter", () => {
   });
 });
 
-describe("AI SDK turn conversions", () => {
+describe("AI SDK turns", () => {
   const asked = (state: string, extra: Record<string, unknown> = {}) =>
     ({
       id: "m1",
@@ -131,19 +127,4 @@ describe("AI SDK turn conversions", () => {
       output: "here"
     });
   });
-
-  it("throws after an aborted stream, keeping what it produced", async () => {
-    async function* stream(): AsyncGenerator<UIMessageChunk> {
-      yield { type: "text-start", id: "a" };
-      yield { type: "text-delta", id: "a", delta: "Half" };
-      yield { type: "abort" };
-    }
-    const seen: string[] = [];
-    await expect(async () => {
-      for await (const chunk of toResponseChunks(stream())) {
-        seen.push(chunk.type);
-      }
-    }).rejects.toThrow("aborted");
-    expect(seen).toEqual(["text-start", "text-delta"]);
-  });
 });