Refactor accumulator for turn continuation model

Replace seedMessages/completeSeeded with initMessage/completeMessage
on MessageAccumulator to clarify the turn continuation lifecycle.
initMessage is idempotent — creates tracking state from an existing
message when not active, or syncs with an externally updated message
when already active. completeMessage marks a message as completed.

Extract shared tool output transition logic into tool-transitions.ts
so the accumulator's _processToolOutput delegates to a single
transitionToolPart function instead of duplicating the switch.

Author: zknill

examples/custom-codec/codec.ts

@@ -514,17 +514,18 @@ class AgentAccumulator implements MessageAccumulator<AgentEvent, AgentMessage> {
   }
 
   /**
-   * Seed the accumulator with an existing message for cross-turn events.
+   * Initialize the accumulator with an existing message for continuation.
    * Not supported by this simple codec.
-   * @param _messages - The messages to seed (unused).
+   * @param _messageId - The message identifier (unused).
+   * @param _message - The existing message (unused).
    */
   // eslint-disable-next-line @typescript-eslint/no-unused-vars -- interface contract requires parameters
-  seedMessages(_messages: { messageId: string; message: AgentMessage }[]): void {
+  initMessage(_messageId: string, _message: AgentMessage): void {
     // No-op — cross-turn events are not supported by this codec.
   }
 
   // eslint-disable-next-line @typescript-eslint/no-unused-vars -- interface contract requires parameters
-  completeSeeded(_messageId: string): void {
+  completeMessage(_messageId: string): void {
     // No-op — cross-turn events are not supported by this codec.
   }
 }

src/core/codec/types.ts

@@ -204,18 +204,18 @@ export interface MessageAccumulator<TEvent, TMessage> {
   /** Apply an external update to a message (e.g. from an update callback). */
   updateMessage(message: TMessage): void;
   /**
-   * Seed the accumulator with existing messages so subsequent events
-   * targeting these messageIds update the existing state rather than creating
-   * blank messages. Used for cross-turn events (e.g. tool result delivery)
-   * and history replay where multiple messages may need seeding.
+   * Ensure the accumulator is ready to process events for the given message.
+   * If not already active, creates internal tracking state from the message.
+   * If already active, syncs internal state with the provided message
+   * (picking up external changes like cross-turn amendments).
+   * Idempotent — safe to call before every processOutputs.
    */
-  seedMessages(messages: { messageId: string; message: TMessage }[]): void;
+  initMessage(messageId: string, message: TMessage): void;
   /**
-   * Mark a previously seeded message as completed. Called after processing
-   * cross-turn event outputs — the seed temporarily re-activates the message,
-   * and this call finalizes it so it appears in {@link completedMessages}.
+   * Mark a message as completed. Removes it from active tracking so it
+   * appears in {@link completedMessages}. No-op if not active.
    */
-  completeSeeded(messageId: string): void;
+  completeMessage(messageId: string): void;
   /** All messages accumulated so far (in-progress and completed). */
   readonly messages: TMessage[];
   /** Only messages whose streams have finished. */

src/core/transport/client-transport.ts

@@ -375,7 +375,7 @@ class DefaultClientTransport<TEvent, TMessage> implements ClientTransport<TEvent
     }
 
     const accumulator = this._codec.createAccumulator();
-    accumulator.seedMessages([{ messageId: targetMsgId, message: existingNode.message }]);
+    accumulator.initMessage(targetMsgId, existingNode.message);
     accumulator.processOutputs([output]);
 
     const updatedMsg = accumulator.messages.at(-1);
@@ -445,14 +445,14 @@ class DefaultClientTransport<TEvent, TMessage> implements ClientTransport<TEvent
     if (!observer) return;
 
     // Sync the accumulator with the tree before processing. If the message
-    // was updated externally (via cross-turn events), seedMessages updates the
+    // was updated externally (via cross-turn events), initMessage syncs the
     // accumulator's state so the update isn't lost when processing
     // late turn events like finish-step/finish.
     const msgId = observer.headers[HEADER_MSG_ID];
     if (msgId) {
       const treeNode = this._tree.getNode(msgId);
       if (treeNode) {
-        observer.accumulator.seedMessages([{ messageId: msgId, message: treeNode.message }]);
+        observer.accumulator.initMessage(msgId, treeNode.message);
       }
     }
 
@@ -608,7 +608,7 @@ class DefaultClientTransport<TEvent, TMessage> implements ClientTransport<TEvent
             messageId: node.msgId,
           }));
           const accumulator = this._codec.createAccumulator();
-          accumulator.seedMessages([{ messageId: node.msgId, message: existingNode.message }]);
+          accumulator.initMessage(node.msgId, existingNode.message);
           accumulator.processOutputs(outputs);
           const updatedMsg = accumulator.messages.at(-1);
           if (updatedMsg) {

src/core/transport/decode-history.ts

@@ -102,28 +102,17 @@ const decodeAll = <TEvent, TMessage>(state: HistoryState<TEvent, TMessage>): Dec
     const amendTarget = headers[HEADER_AMEND];
 
     // Cross-turn events target an existing message from a different turn.
-    // Route them to the turn accumulator that owns the target msgId
-    // so the tool part state transitions correctly in history.
+    // Route to the owning turn's accumulator via initMessage lifecycle.
     if (amendTarget) {
       for (const turn of turns.values()) {
         if (turn.msgHeaders.has(amendTarget)) {
-          // Seed the accumulator with the message's current state so the
-          // cross-turn events update the correct tool parts. Use the
-          // accumulator's full messages list (includes in-progress) and
-          // match by position against the msgHeaders insertion order.
           const headerKeys = [...turn.msgHeaders.keys()];
           const msgIndex = headerKeys.indexOf(amendTarget);
           const currentMsg = msgIndex === -1 ? undefined : turn.accumulator.messages[msgIndex];
           if (currentMsg) {
-            turn.accumulator.seedMessages([{ messageId: amendTarget, message: currentMsg }]);
+            turn.accumulator.initMessage(amendTarget, currentMsg);
           }
           turn.accumulator.processOutputs(outputs);
-          // Defer completion until after all events are processed. A finish
-          // event that follows this update in serial order must still be
-          // able to find the message in _activeMessages (e.g. to apply
-          // messageMetadata). If no finish event arrives, the deferred
-          // completeSeeded ensures the message still appears in
-          // completedMessages.
           deferredCompletions.push({ accumulator: turn.accumulator, messageId: amendTarget });
           break;
         }
@@ -175,11 +164,11 @@ const decodeAll = <TEvent, TMessage>(state: HistoryState<TEvent, TMessage>): Dec
     }
   }
 
-  // Complete any seeded-for-update messages that were not already completed
-  // by a finish/abort event. Idempotent — if finish already removed the message
-  // from _activeMessages, completeSeeded is a no-op.
+  // Complete any messages that were re-activated for cross-turn updates.
+  // Idempotent — if finish already removed the message from active tracking,
+  // completeMessage is a no-op.
   for (const { accumulator, messageId } of deferredCompletions) {
-    accumulator.completeSeeded(messageId);
+    accumulator.completeMessage(messageId);
   }
 
   // Collect completed messages in chronological order (oldest first) by turn.

src/core/transport/view.ts

@@ -597,7 +597,7 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
   // Private: scoped event forwarding
   // -------------------------------------------------------------------------
 
-  private _updateVisibleSnapshot(nodes?: TreeNode<TMessage>[]): void {
+  private _updateVisibleSnapshot(nodes?: MessageNode<TMessage>[]): void {
     const resolved = nodes ?? this.flattenNodes();
     this._lastVisibleIds = resolved.map((n) => n.msgId);
     this._lastVisibleMessages = resolved.map((n) => n.message);

src/vercel/codec/accumulator.ts

@@ -17,6 +17,7 @@ import type * as AI from 'ai';
 
 import type { DecoderOutput, MessageAccumulator } from '../../core/codec/types.js';
 import { stripUndefined } from '../../utils.js';
+import { toolBase, transitionToolPart } from './tool-transitions.js';
 
 // ---------------------------------------------------------------------------
 // Internal types
@@ -38,15 +39,6 @@ interface ToolPartTracker {
   inputText: string;
 }
 
-/** Fields shared by all DynamicToolUIPart state variants. */
-interface ToolBaseFields {
-  type: 'dynamic-tool';
-  toolName: string;
-  toolCallId: string;
-  title?: string;
-  providerExecuted?: boolean;
-}
-
 /** Bundled per-message state for an in-progress message. */
 interface ActiveMessageState {
   message: AI.UIMessage;
@@ -56,34 +48,6 @@ interface ActiveMessageState {
   streamStatus: Map<string, StreamStatus>;
 }
 
-// ---------------------------------------------------------------------------
-// Tool base helper
-// ---------------------------------------------------------------------------
-
-/**
- * Extract the state-independent base fields for a DynamicToolUIPart.
- * Works with both chunks (tool-input-start, etc.) and existing parts.
- * @param source - Any object containing the required tool identity fields.
- * @param source.toolCallId - The tool call identifier.
- * @param source.toolName - The tool name.
- * @param source.title - Optional display title.
- * @param source.providerExecuted - Whether the provider executed the tool.
- * @returns Base fields shared across all DynamicToolUIPart state variants.
- */
-const toolBase = (source: {
-  toolCallId: string;
-  toolName: string;
-  title?: string;
-  providerExecuted?: boolean;
-}): ToolBaseFields =>
-  stripUndefined({
-    type: 'dynamic-tool' as const,
-    toolCallId: source.toolCallId,
-    toolName: source.toolName,
-    title: source.title,
-    providerExecuted: source.providerExecuted,
-  });
-
 // ---------------------------------------------------------------------------
 // DeltaStreamTracker — manages text or reasoning stream accumulation
 // ---------------------------------------------------------------------------
@@ -172,23 +136,13 @@ class DefaultUIMessageAccumulator implements MessageAccumulator<AI.UIMessageChun
     }
   }
 
-  seedMessages(messages: { messageId: string; message: AI.UIMessage }[]): void {
-    for (const { messageId, message } of messages) {
-      this._seedOne(messageId, message);
-    }
-  }
-
-  completeSeeded(messageId: string): void {
-    this._activeMessages.delete(messageId);
-  }
-
-  private _seedOne(messageId: string, message: AI.UIMessage): void {
+  initMessage(messageId: string, message: AI.UIMessage): void {
     const existing = this._activeMessages.get(messageId);
 
     if (existing) {
-      // Update: sync the active state with an externally updated message.
+      // Already active — sync with the externally updated message.
       // Replace the message and rebuild tool trackers so the accumulator
-      // reflects updates (e.g. tool results published via cross-turn events)
+      // reflects updates (e.g. cross-turn amendments applied to the tree)
       // that happened outside the streaming flow.
       const cloned = structuredClone(message);
       const listIdx = this._messageList.indexOf(existing.message);
@@ -207,16 +161,15 @@ class DefaultUIMessageAccumulator implements MessageAccumulator<AI.UIMessageChun
       return;
     }
 
+    // Not active — create tracking state from the existing message.
     const cloned = structuredClone(message);
     const toolTrackers: Record<string, ToolPartTracker> = {};
     const streamStatus = new Map<string, StreamStatus>();
 
-    // Reconstruct tool trackers from existing dynamic-tool parts
     for (let i = 0; i < cloned.parts.length; i++) {
       const part = cloned.parts[i];
       if (part?.type === 'dynamic-tool') {
         toolTrackers[part.toolCallId] = { partIndex: i, inputText: '' };
-        // All existing tool parts have completed their input phase
         streamStatus.set(part.toolCallId, 'finished');
       }
     }
@@ -232,8 +185,7 @@ class DefaultUIMessageAccumulator implements MessageAccumulator<AI.UIMessageChun
     this._activeMessages.set(messageId, state);
 
     // If this message is already in the list (completed previously),
-    // replace it in-place so updates apply to the same reference.
-    // Otherwise push as a new entry.
+    // replace in-place. Otherwise push as a new entry.
     const existingIdx = this._messageList.findIndex((m) => m.id === message.id);
     if (existingIdx === -1) {
       this._messageList.push(state.message);
@@ -242,6 +194,10 @@ class DefaultUIMessageAccumulator implements MessageAccumulator<AI.UIMessageChun
     }
   }
 
+  completeMessage(messageId: string): void {
+    this._activeMessages.delete(messageId);
+  }
+
   // -------------------------------------------------------------------------
   // Shared helpers
   // -------------------------------------------------------------------------
@@ -573,48 +529,7 @@ class DefaultUIMessageAccumulator implements MessageAccumulator<AI.UIMessageChun
     const found = this._getToolPart(chunk.toolCallId, state);
     if (!found) return;
 
-    switch (chunk.type) {
-      case 'tool-output-available': {
-        state.message.parts[found.tracker.partIndex] = stripUndefined({
-          ...toolBase(found.part),
-          state: 'output-available' as const,
-          input: found.part.input,
-          output: chunk.output,
-          preliminary: chunk.preliminary,
-        });
-        break;
-      }
-
-      case 'tool-output-error': {
-        state.message.parts[found.tracker.partIndex] = {
-          ...toolBase(found.part),
-          state: 'output-error',
-          input: found.part.input,
-          errorText: chunk.errorText,
-        };
-        break;
-      }
-
-      case 'tool-output-denied': {
-        state.message.parts[found.tracker.partIndex] = {
-          ...toolBase(found.part),
-          state: 'output-denied',
-          input: found.part.input,
-          approval: { id: '', approved: false },
-        };
-        break;
-      }
-
-      case 'tool-approval-request': {
-        state.message.parts[found.tracker.partIndex] = {
-          ...toolBase(found.part),
-          state: 'approval-requested',
-          input: found.part.input,
-          approval: { id: chunk.approvalId },
-        };
-        break;
-      }
-    }
+    state.message.parts[found.tracker.partIndex] = transitionToolPart(found.part, chunk);
   }
 
   // -------------------------------------------------------------------------