codec/anthropic: add Anthropic Agent SDK codec implementation by JoaoDiasAbly · Pull Request #17 · ably/ably-ai-transport-js

JoaoDiasAbly · 2026-03-30T13:32:40Z

Implement AgentCodec — a Codec<AgentCodecEvent, AgentMessage> that maps Anthropic Agent SDK streaming events and messages to/from Ably message primitives.

Components:

Encoder: converts SDKPartialAssistantMessage stream events into Ably publish/append operations, tracking open content blocks by index
Decoder: reconstructs SDKPartialAssistantMessage events from Ably messages, with lifecycle tracker for mid-stream join handling
Accumulator: builds SDKAssistantMessage state from decoder outputs, tracking concurrent in-progress messages with tool input buffering
AgentCodec: wires the three factories together with isTerminal (SDKResultMessage)
Transport factories: createClientTransport and createServerTransport pre-bound to AgentCodec

Adds @anthropic-ai/sdk as a peer dependency for proper type resolution of transitive types from @anthropic-ai/claude-agent-sdk.

Unit tests cover all code paths including content block streaming (text, tool_use, thinking), abort handling, lifecycle tracker phases, lazy message creation for mid-stream joins, and concurrent message tracking. Integration tests validate encode/decode roundtrips over real Ably channels for 6 scenarios.

Key Differences from Vercel Codec

Concern	Vercel	Anthropic Agent SDK
TEvent source	`UIMessageChunk` -- Vercel's own streaming type	`SDKMessage` subset -- wraps Anthropic API stream events
TMessage	`UIMessage` -- single type with `role` field	`SDKAssistantMessage \| SDKUserMessage` -- union of two structurally different types
Event self-identification	Every chunk carries its own identity (e.g. `chunk.id`, `chunk.toolCallId`)	`content_block_stop` has only `index`, no type. Encoder must track open blocks by index.
Encoder state	Stateless aside from `_aborted` flag	Needs `Map<number, { name, streamId }>` to track open content blocks by index
Streaming opt-in	Always streaming	Requires `includePartialMessages: true`. Extended thinking disables streaming entirely.
Event union size	`UIMessageChunk` has ~20 variants, all conversation-relevant	`SDKMessage` has ~20 variants, only ~5 are conversation-relevant. Need a filtered subset type.
Lifecycle events	`start`, `start-step`, `finish-step`, `finish` (Vercel-specific)	`message_start`, `message_stop`, `message_delta` (Anthropic API events)
Content blocks	Parts array on UIMessage (text, file, reasoning, dynamic-tool, data-*)	Content blocks on BetaMessage (text, tool_use, thinking)
Tool input streaming	`tool-input-start/delta/available` events with `toolCallId`	`content_block_start/delta/stop` with `index` + `input_json_delta`
Non-streaming fallback	`tool-input-available` without prior start -> discrete publish	Complete `AssistantMessage` when `includePartialMessages` is false
Peer dependency	`ai` package	`@anthropic-ai/claude-agent-sdk` package
Client-side SDK	Yes (`useChat`, React hooks from `@ai-sdk/react`)	No -- Agent SDK is server-only. Generic React hooks from `@ably/ai-transport/react` used directly.

Known Issues & Limitations

#	Type	Location	Description
1	Codec authoring tension	Lifecycle tracker + complex SDK types	The lifecycle tracker synthesizes events for mid-stream joins. For Vercel, synthetic events are simple (`{ type: 'start', messageId }` -- flat objects). For Anthropic, synthetic events require constructing nested `BetaMessage` objects with many required fields. This is fragile -- if the Anthropic SDK adds a required field, the synthetic construction breaks at runtime (the `as unknown as` cast bypasses compile-time checks). Mitigated with block-level `eslint-disable` for `unicorn/no-null` and clear CAST comments. The fundamental friction remains: the lifecycle tracker forces codecs to construct full `TEvent` objects.
2	CI / type resolution	Agent SDK broken `.d.ts`	See detailed writeup below.
3	Codec limitation	`accumulator.updateMessage()`	Uses `uuid ?? session_id` as identity key. `SDKUserMessage.uuid` is optional and `session_id` can be empty, so the key can produce false-positive matches. Not currently called by the core transport. If it becomes load-bearing, the interface should be extended to pass `x-ably-msg-id`.

Anthropic Agent SDK broken type declarations

The @anthropic-ai/claude-agent-sdk package ships a sdk.d.ts with two problems:

Missing transitive dependency types: It imports from @modelcontextprotocol/sdk/types.js but doesn't declare it as a dependency. Without @modelcontextprotocol/sdk installed, these imports fail.
Broken internal references: sdk.d.ts defines a SDKControlRequestInner union type that references 12+ type names that are never defined in the .d.ts (e.g. SDKControlChannelEnableRequest, SDKControlEndSessionRequest). No dependency can fix this -- the types simply don't exist.

With skipLibCheck: true, TypeScript ignores both problems. But ESLint's @typescript-eslint rules create their own TypeScript program instance, and depending on the environment (Node version, npm resolution strategy), the broken .d.ts may cause some or all exported types to resolve as error.

What we did:

Added @anthropic-ai/sdk as a peer + dev dependency (fixes BetaMessage/MessageParam resolution)
Added @modelcontextprotocol/sdk as a dev dependency (fixes MCP type resolution)
This reduced broken refs from "all types unresolvable" to 15 internal SDKControl*Request types on one line -- types we never use
Added per-line eslint-disable on the 3 test lines that CI flags

Why it passes locally but fails in CI: Locally (macOS, Node 24), the 15 broken internal refs don't cascade. In CI (Ubuntu, Node 20/22/24, npm ci), they may cause the module's exports to be treated as unresolvable. This is a difference in TypeScript's error recovery behavior across environments.

Conclusions

How easy was it to add a new codec?

The architecture delivered on its promise. The two-layer split (generic transport + pluggable codec) meant we never touched a single line of transport code. The EncoderCore and DecoderCore handled all Ably wire protocol concerns. We wrote ~2,100 lines of codec source code and ~3,800 lines of tests, with zero changes to the existing codebase (aside from package.json for deps and build config).

The custom-codec example was an effective starting template. The Vercel codec served as a comprehensive reference for every edge case. The documentation in docs/internals/ was accurate and thorough enough to build a mental model before reading any source code.

Overall verdict: moderately easy for the codec itself, but significantly harder than expected due to SDK type friction.

What went well

EncoderCore and DecoderCore are excellent abstractions. The domain encoder is just a switch statement mapping events to four core operations. The domain decoder provides four hooks. All Ably-specific complexity (serial tracking, append batching, flush/recovery, prefix-match, first-contact) is handled by the cores. A codec author never needs to understand Ably message actions.
The header utilities (headerWriter/headerReader) are clean and ergonomic. The x-domain- prefix is handled automatically. The fluent builder pattern makes header construction readable.
The lifecycle tracker is a good generic solution for the mid-stream join problem. Configuring it with phases is straightforward.
The test infrastructure is reusable. Sandbox provisioning, unique channel names, client lifecycle helpers -- all worked immediately for our codec's integration tests.
The transport factory pattern (Omit codec from options) is trivially implementable. The Anthropic transport factories are ~50 lines.

What was challenging

Transitive SDK type resolution. The Anthropic Agent SDK (@anthropic-ai/claude-agent-sdk) depends on @anthropic-ai/sdk for types like BetaRawMessageStreamEvent and BetaMessage. Without @anthropic-ai/sdk installed, TypeScript resolves these as error types, causing every property access to require eslint-disable comments. Adding @anthropic-ai/sdk as an explicit peer dependency fixed the cascading failures. This is not a problem with the AI Transport SDK's design, but it is a real pain point for codec authors whose framework SDK has deep type dependency chains.
Complex nested SDK types require as unknown as casts. The Anthropic codec has ~30 as unknown as casts in source files (vs 0 in Vercel). These exist because the decoder and accumulator construct SDK types (BetaRawMessageStreamEvent, BetaMessage, content blocks) from decoded wire data. The object literals don't structurally satisfy the full union types -- for example, BetaContentBlock is a union of 14 variants, and constructing any one of them requires all fields for that variant. The root cause is using complex SDK types as TEvent/TMessage. If we defined our own simpler types (like the custom-codec example's flat AgentMessage { id, role, text, toolCalls }), there'd be zero casts -- but consumers couldn't work with familiar SDK types. The casts are at well-defined trust boundaries and validated by the switch-based type narrowing, so they're correct -- just not ideal for maintainability. Additionally, the Anthropic SDK uses null where the project linter prefers undefined, requiring eslint-disable for unicorn/no-null. These are consolidated into block-level disables around shell object constructions to minimize noise.
Events that don't self-identify. Vercel's UIMessageChunk types carry their own identity (e.g. text-delta has chunk.id). Anthropic's content_block_delta and content_block_stop only carry an index, so the encoder must track open blocks to map index to stream ID. This is a minor annoyance, not a fundamental problem, but it adds state that the Vercel encoder doesn't need.
Union TMessage types. The Vercel codec has a single UIMessage type for both roles. The Anthropic codec has SDKAssistantMessage | SDKUserMessage -- structurally different types. This complicates updateMessage, writeMessages, and any code that switches on the union. It works, but it's less ergonomic.
Preserving SDK metadata through the wire. Fields like uuid, session_id, parent_tool_use_id need to survive encode -> Ably -> decode. Each requires a domain header. The initial implementation missed uuid and session_id, which were caught during the deep audit. The Vercel codec doesn't have this problem because UIMessage has fewer metadata fields.

Known limitation: session_id on streaming events. Discrete messages carry session_id via the x-domain-sessionId header and are reconstructed correctly. But streaming events (SDKPartialAssistantMessage reconstructed by the decoder from streamed content blocks) get session_id: '' because session_id is not included in the persistent headers passed to startStream. The accumulated SDKAssistantMessage has the correct session_id (from the message_start event data). Only the transient wrappers produced by the decoder have session_id: '', and nothing currently reads that field from streaming events. Fix path: thread the outer SDKPartialAssistantMessage through _handleStreamEvent so session_id can be added as a persistent header.

Suggestions to improve the base primitives

These are improvements to the core SDK identified while building this codec. None are blockers.

Lifecycle tracker should not force codecs to construct full TEvent objects.

The lifecycle tracker synthesizes missing "startup" events for mid-stream joins. Each phase has a build() function that must return TEvent[].

For Vercel, this is trivial:
```
// Vercel: 1 line, no casts, no eslint-disable
build: (ctx) => [{ type: 'start', messageId: ctx.messageId }]
```
For Anthropic, TEvent includes SDKPartialAssistantMessage -- a wrapper around deeply nested Anthropic SDK types. The build() function must construct a ~25-line object with multiple casts and block-level eslint-disable. The as unknown as cast bypasses compile-time safety, so if the SDK adds a required field, the synthetic construction breaks at runtime.

A simpler alternative: the lifecycle tracker could just report which phases are missing and let the decoder's own hooks handle construction. The hooks already know how to build events -- they do it for every stream start. The tracker doesn't need to build events itself; it just needs to track what's been emitted and what hasn't.
The accumulator contract should specify that messages returns a stable reference. The MessageAccumulator interface says messages returns TMessage[] but doesn't say whether it's the same array every time or a new copy. This matters because the transport reads accumulator.messages on every streaming token (potentially hundreds per response) -- allocating a new array each time is wasteful. Both the Vercel accumulator and the custom-codec example return the same array (mutated in place), but nothing in the interface documents this. Our initial Anthropic accumulator accidentally created a new array on every access. A one-line JSDoc addition would prevent this.

zknill · 2026-04-13T15:28:48Z

+      // -- Complete assistant message (non-streaming or post-stream).
+      // Publishes the full BetaMessage as data. For typical responses this is
+      // well under Ably's 64 KB message limit, but very large tool inputs or
+      // multi-block responses could approach it. Streaming mode avoids this
+      // because content arrives as small deltas instead.
+      case 'assistant': {
+        const messageId = event.message.id;
+        const h = headerWriter()
+          .str('messageId', messageId)
+          .str('uuid', event.uuid)
+          .str('sessionId', event.session_id)
+          .str('parentToolUseId', event.parent_tool_use_id ?? undefined)
+          .build();
+        await this._core.publishDiscrete({ name: 'assistant-message', data: event.message, headers: h }, perWrite);
+        break;


I don't think we want to re-publish the full assistant response as a separate message after we're done streaming it.

I was a bit confused on what to do here after coming back to it for a few days. I do understand the problem and at first I thought I was going to remove this case, but I don't think that's what we want here. I think the correct thing is to know when to skip this so that's the approach I took now. LMK what you think

zknill · 2026-04-13T15:35:10Z

+// Default implementation
+// ---------------------------------------------------------------------------
+
+class DefaultAgentAccumulator implements MessageAccumulator<AgentCodecEvent, AgentMessage> {


I think this class is missing initMessage and completeMessage ? I get errors compiling the ts

zknill · 2026-04-13T15:38:35Z

+  // Discrete message parts from writeMessages: identified by x-ably-role header.
+  // Only applies to user-message and assistant-message names — other discrete
+  // events (message-start, message-delta, result, etc.) also carry x-ably-role
+  // but must be dispatched by name, not role.
+  if (HEADER_ROLE in h && (input.name === 'user-message' || input.name === 'assistant-message')) {
+    const role = h[HEADER_ROLE];
+    if (role === 'user') {
+      return decodeUserMessage(input);
+    }
+    return decodeAssistantMessage(input);
+  }
+
+  switch (input.name) {
+    case 'message-start': {
+      return decodeMessageStart(input, turnId, lifecycle);
+    }
+    case 'message-delta': {
+      return decodeMessageDelta(input);
+    }
+    case 'message-stop': {
+      return decodeMessageStop(input);
+    }
+    case 'assistant-message': {
+      return decodeAssistantMessage(input);
+    }
+    case 'user-message': {
+      return decodeUserMessage(input);
+    }


this is a bit confusing that user-message and assistant-message are handle at the top and in this switch statement

zknill · 2026-04-13T15:39:20Z

+    switch (eventType) {
+      case 'message_start': {
+        // CAST: message_start carries .message; cast through unknown to Record.
+        const message = (streamEvent as unknown as Record<string, unknown>).message as Record<string, unknown>;


I think we should cast to an object type that has the fields that we are expecting to use on it. e.g. message.id and message.model, etc

zknill · 2026-04-13T15:41:43Z

+        content[index] = {
+          type: 'thinking',
+          thinking: '',
+          signature: '',


I think we need to include the signature here, or the api will reject future inference with thinking blocks in the history

had to do a bit extra work on this one actually, but I think it's fully addressed now. Here's what's been done:

Added signature_delta handler to the accumulator that appends to block.signature on thinking blocks

Added signature_delta handler to the encoder that buffers signature data instead of streaming it (streaming would mix it with thinking text on the wire)

Encoder includes the buffered signature as an x-domain-signature header on closeStream

Decoder reads the signature header and emits a synthetic signature_delta event before content_block_stop

Signatures now survive the full encode/Ably/decode roundtrip, required for multi-turn API continuity with thinking blocks

zknill · 2026-04-13T15:41:59Z

+      // Other delta types (e.g. citations_delta): no-op
+      default: {
+        break;
+      }


handle signature_delta here

addressed - comment above

Implement AgentCodec — a Codec<AgentCodecEvent, AgentMessage> that maps Anthropic Agent SDK streaming events and messages to/from Ably message primitives. Components: - Encoder: converts SDKPartialAssistantMessage stream events into Ably publish/append operations, tracking open content blocks by index - Decoder: reconstructs SDKPartialAssistantMessage events from Ably messages, with lifecycle tracker for mid-stream join handling - Accumulator: builds SDKAssistantMessage state from decoder outputs, tracking concurrent in-progress messages with tool input buffering - AgentCodec: wires the three factories together with isTerminal (SDKResultMessage) and getMessageKey helpers - Transport factories: createClientTransport and createServerTransport pre-bound to AgentCodec Adds @anthropic-ai/sdk as a peer dependency for proper type resolution of transitive types from @anthropic-ai/claude-agent-sdk. Unit tests cover all code paths including content block streaming (text, tool_use, thinking), abort handling, lifecycle tracker phases, lazy message creation for mid-stream joins, and concurrent message tracking. Integration tests validate encode/decode roundtrips over real Ably channels for 9 scenarios. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

- Add src/anthropic/vite.config.ts and build:anthropic script; update package.json exports to point to dist/ artifacts instead of raw .ts - Exclude anthropic/ from core dts plugin, add demo to formatter paths - Wire turn.abortSignal to Agent SDK AbortController in demo route - Consolidate per-line eslint-disable unicorn/no-null into block-level disables in accumulator, decoder lifecycle tracker, and decoder abort - Remove duplicate CAST comments in decoder - Add doc comments on updateMessage identity limitation and encoder message size consideration - Update INVESTIGATION.md issues table and conclusions Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Investigation notes have been distilled into the PR description. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

github-actions · 2026-04-15T09:24:07Z

Coverage Report

Status	Category	Percentage	Covered / Total
🔵	Lines	93.57%	2463 / 2632
🔵	Statements	92.07%	2626 / 2852
🔵	Functions	94.06%	444 / 472
🔵	Branches	79.61%	1148 / 1442

File Coverage

File	Stmts	Branches	Functions	Lines	Uncovered Lines
Changed Files
src/anthropic/codec/accumulator.ts	97.93%	83.65%	100%	98.35%	148, 224, 240, 409
src/anthropic/codec/decoder.ts	100%	87.32%	100%	100%
src/anthropic/codec/encoder.ts	100%	100%	100%	100%
src/anthropic/codec/types.ts	0%	0%	0%	0%

Generated in workflow #134 for commit b11b654 by the Vitest Coverage Report Action

Encoder: skip redundant discrete publish of SDKAssistantMessage when the message was already streamed (tracks message IDs via message_start). Add signature_delta support so thinking block signatures are streamed. Cast message_start payload to a typed interface instead of Record. Accumulator: implement initMessage and completeMessage (required by the MessageAccumulator interface, used by the core transport for cross-turn amendments and history hydration). Add signature_delta handler so thinking blocks have valid signatures for multi-turn API continuity. Decoder: remove redundant HEADER_ROLE early guard in decodeDiscretePayload — the switch statement already dispatches by message name. Demo: update Anthropic React demo to current library API (useView instead of non-existent useHistory/useConversationTree, TransportProvider instead of ChannelProvider, MessageNode instead of ConversationNode). Fix SDK version mismatch via overrides. Configure webpack source aliases to avoid Rolldown CJS shim issues with Next.js. Tests cover streaming guard, signature_delta in encoder and accumulator, and initMessage/completeMessage lifecycle. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

The already-active branch of initMessage replaced the message reference but left contentBlocks, toolInputBuffers, and streamStatus stale. If a cross-turn amendment changed the content blocks, subsequent processOutputs could misroute events against outdated tracking state. Now clears and rebuilds all three maps from the synced message, matching the Vercel accumulator's pattern. Also tightens the list lookup in the not-active branch: assistant messages now match by BetaMessage ID (stable, unique) instead of the weak uuid/session_id heuristic that could collide across messages in the same session. Tests cover tracking rebuild after sync and correct matching when multiple assistant messages share a session_id. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

The encoder was streaming signature_delta data through the same Ably message stream as thinking_delta. Since the decoder maps all appends on a thinking stream to thinking_delta events, signatures were lost on the client side — appended to block.thinking instead of block.signature. Fix: the encoder now buffers signature_delta data internally and includes the accumulated signature as an x-domain-signature header on the closeStream call. The decoder reads this header and emits a synthetic signature_delta event before content_block_stop, so the accumulator correctly populates block.signature on the receiving end. This preserves the signature through the full encode → Ably → decode → accumulate roundtrip, which is required for multi-turn API continuity with thinking blocks. Tests cover buffering (not appending), concatenation of multiple signature deltas, header presence/absence on close, separation from thinking text, and decoder reconstruction from close headers. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

JoaoDiasAbly · 2026-04-15T14:02:59Z

thanks for reviewing @zknill
I think everything as been addressed and it's ready for a re-review 🙏

JoaoDiasAbly force-pushed the feature/anthropic-codecs branch from 36ef9e3 to d8d278b Compare March 31, 2026 11:10

JoaoDiasAbly marked this pull request as ready for review April 6, 2026 17:07

JoaoDiasAbly requested review from mschristensen and zknill April 6, 2026 17:07

zknill reviewed Apr 13, 2026

View reviewed changes

JoaoDiasAbly and others added 11 commits April 15, 2026 10:17

fix dependencies

3a56a26

fix dependencies

fc02e49

fix lint

3581dec

rebase

efdede2

fix flaky test

06a3ad5

fix integration test

2b73a98

working demo

bf0a8f8

remove INVESTIGATION.md

c826b85

Investigation notes have been distilled into the PR description. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

review adjustments

d66d041

JoaoDiasAbly force-pushed the feature/anthropic-codecs branch from f9fd1fe to d66d041 Compare April 15, 2026 09:23

github-actions bot temporarily deployed to staging/pull/17/typedoc April 15, 2026 11:38 Inactive

github-actions bot deployed to staging/pull/17/typedoc April 15, 2026 12:52 View deployment

JoaoDiasAbly requested a review from zknill April 15, 2026 14:02

Conversation

JoaoDiasAbly commented Mar 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Key Differences from Vercel Codec

Known Issues & Limitations

Anthropic Agent SDK broken type declarations

Conclusions

How easy was it to add a new codec?

What went well

What was challenging

Suggestions to improve the base primitives

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Apr 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Coverage Report

Uh oh!

JoaoDiasAbly commented Apr 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Milestone

Development

Uh oh!

2 participants

JoaoDiasAbly commented Mar 30, 2026 •

edited

Loading

github-actions bot commented Apr 15, 2026 •

edited

Loading