transport: fix multi-view selection stability and streaming visibility

Addresses several issues found during design review:

- Add selection pinning: when a new fork appears, views that were
  already showing the original message pin their selection to it,
  preventing unintended branch shifts from other views' edits.
  Views that initiate a fork auto-select the new sibling via send().
- Fix streaming visibility: _visibleChanged now compares message
  references alongside msgIds, so in-place content updates (streaming
  accumulation) trigger view 'update' events for observers.
- Fix ably-message filtering: _onTreeAblyMessage now checks that the
  msgId is in the visible set, not just not-withheld. Messages on
  non-selected branches are no longer forwarded.
- Validate messageId in regenerate/edit: throw Ably.ErrorInfo when
  the target node doesn't exist in the tree.
- Add getSiblingIndex() to TreeInternal for selection pinning.
- Add constructor trace log, warn on _getHistoryBefore fallback,
  comment on ChatTransport's default-view assumption.

Tests cover pinning stability, fork auto-select, streaming emit,
off-branch ably-message suppression, unknown messageId validation,
and loadOlder on a closed view.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: mschristensen

src/core/transport/tree.ts

@@ -53,6 +53,12 @@ export interface TreeInternal<TMessage> extends Tree<TMessage> {
    */
   getGroupRoot(msgId: string): string;
 
+  /**
+   * Get the index of a specific msgId within its sibling group.
+   * Returns -1 if the msgId is not found.
+   */
+  getSiblingIndex(msgId: string): number;
+
   /** Forward a raw Ably message event to tree subscribers. */
   emitAblyMessage(msg: Ably.InboundMessage): void;
   /** Forward a turn lifecycle event to tree subscribers. */
@@ -344,6 +350,10 @@ export class DefaultTree<TMessage> implements TreeInternal<TMessage> {
     return this._getSiblingGroup(msgId).length > 1;
   }
 
+  getSiblingIndex(msgId: string): number {
+    return this._getSiblingGroup(msgId).findIndex((n) => n.msgId === msgId);
+  }
+
   getNode(msgId: string): TreeNode<TMessage> | undefined {
     this._logger.trace('DefaultTree.getNode();', { msgId });
     return this._nodeIndex.get(msgId)?.node;

src/core/transport/view.ts

@@ -13,9 +13,10 @@
  * visible nodes, and 'turn' only for turns with visible messages.
  */
 
-import type * as Ably from 'ably';
+import * as Ably from 'ably';
 
 import { HEADER_MSG_ID, HEADER_TURN_ID } from '../../constants.js';
+import { ErrorCode } from '../../errors.js';
 import { EventEmitter } from '../../event-emitter.js';
 import type { Logger } from '../../logger.js';
 import { getHeaders } from '../../utils.js';
@@ -96,9 +97,12 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
   /** Msg-ids loaded from history but not yet revealed to the UI. */
   private readonly _withheldMsgIds = new Set<string>();
 
-  /** Snapshot of visible msgIds — used to detect whether tree updates affect the view. */
+  /** Snapshot of visible msgIds — used to detect structural changes and for selection pinning. */
   private _lastVisibleIds: string[] = [];
 
+  /** Snapshot of visible message references — used to detect in-place content updates (streaming). */
+  private _lastVisibleMessages: TMessage[] = [];
+
   /** Whether there are more history pages to fetch from the channel. */
   private _hasMoreHistory = false;
 
@@ -119,10 +123,11 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
     this._codec = options.codec;
     this._sendDelegate = options.sendDelegate;
     this._logger = options.logger.withContext({ component: 'View' });
+    this._logger.trace('DefaultView();');
     this._emitter = new EventEmitter<ViewEventsMap>(this._logger);
 
     // Snapshot initial visible state
-    this._lastVisibleIds = this._computeVisibleIds();
+    this._updateVisibleSnapshot();
 
     // Subscribe to tree events and re-emit scoped versions
     this._unsubs.push(
@@ -242,15 +247,37 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
   // Spec: AIT-CT3, AIT-CT4
   async send(input: TMessage | TMessage[], options?: SendOptions): Promise<ActiveTurn<TEvent>> {
     this._logger.trace('DefaultView.send();');
-    return this._sendDelegate(input, options, { flattenNodes: () => this.flattenNodes() });
+    const result = await this._sendDelegate(input, options, { flattenNodes: () => this.flattenNodes() });
+
+    // Auto-select the new fork in this view when creating a fork.
+    // The delegate's optimistic insert created the sibling — pin this
+    // view to the latest (new) sibling so the user sees their edit/regeneration.
+    if (options?.forkOf) {
+      const groupRoot = this._tree.getGroupRoot(options.forkOf);
+      const siblings = this._tree.getSiblings(options.forkOf);
+      if (siblings.length > 1) {
+        this._selections.set(groupRoot, siblings.length - 1);
+        this._updateVisibleSnapshot();
+        this._emitter.emit('update');
+      }
+    }
+
+    return result;
   }
 
   // Spec: AIT-CT5
   async regenerate(messageId: string, options?: SendOptions): Promise<ActiveTurn<TEvent>> {
     this._logger.trace('DefaultView.regenerate();', { messageId });
 
     const node = this._tree.getNode(messageId);
-    const parentId = node?.parentId;
+    if (!node) {
+      throw new Ably.ErrorInfo(
+        `unable to regenerate; message not found in tree: ${messageId}`,
+        ErrorCode.InvalidArgument,
+        400,
+      );
+    }
+    const parentId = node.parentId;
 
     return this.send([], {
       ...options,
@@ -272,7 +299,14 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
     this._logger.trace('DefaultView.edit();', { messageId });
 
     const node = this._tree.getNode(messageId);
-    const parentId = node?.parentId;
+    if (!node) {
+      throw new Ably.ErrorInfo(
+        `unable to edit; message not found in tree: ${messageId}`,
+        ErrorCode.InvalidArgument,
+        400,
+      );
+    }
+    const parentId = node.parentId;
 
     return this.send(newMessages, {
       ...options,
@@ -289,7 +323,13 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
     this._logger.trace('DefaultView._getHistoryBefore();', { messageId });
     const all = this.flattenNodes();
     const idx = all.findIndex((n) => n.msgId === messageId);
-    return idx === -1 ? all : all.slice(0, idx);
+    if (idx === -1) {
+      this._logger.warn('DefaultView._getHistoryBefore(); target not in visible nodes, returning full list', {
+        messageId,
+      });
+      return all;
+    }
+    return all.slice(0, idx);
   }
 
   // -------------------------------------------------------------------------
@@ -456,22 +496,50 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
   // Private: scoped event forwarding
   // -------------------------------------------------------------------------
 
-  private _computeVisibleIds(): string[] {
-    return this.flattenNodes().map((n) => n.msgId);
+  private _computeVisibleNodes(): TreeNode<TMessage>[] {
+    return this.flattenNodes();
   }
 
   private _updateVisibleSnapshot(): void {
-    this._lastVisibleIds = this._computeVisibleIds();
+    const nodes = this._computeVisibleNodes();
+    this._lastVisibleIds = nodes.map((n) => n.msgId);
+    this._lastVisibleMessages = nodes.map((n) => n.message);
   }
 
   private _onTreeUpdate(): void {
-    const newIds = this._computeVisibleIds();
-    if (this._visibleChanged(newIds)) {
+    // Pin selections for previously-visible nodes that now have siblings.
+    // This prevents new forks (from other views' edits/regenerates) from
+    // shifting this view to a branch the user didn't navigate to.
+    this._pinVisibleSelections();
+
+    const nodes = this._computeVisibleNodes();
+    const newIds = nodes.map((n) => n.msgId);
+    const newMessages = nodes.map((n) => n.message);
+    if (this._visibleChanged(newIds, newMessages)) {
       this._lastVisibleIds = newIds;
+      this._lastVisibleMessages = newMessages;
       this._emitter.emit('update');
     }
   }
 
+  /**
+   * For each previously-visible message that now has siblings but no
+   * explicit selection, pin the selection to that message's index.
+   * This preserves the current branch when new forks appear from
+   * other views or external sources.
+   */
+  private _pinVisibleSelections(): void {
+    for (const msgId of this._lastVisibleIds) {
+      if (!this._tree.hasSiblings(msgId)) continue;
+      const groupRoot = this._tree.getGroupRoot(msgId);
+      if (this._selections.has(groupRoot)) continue;
+      const idx = this._tree.getSiblingIndex(msgId);
+      if (idx >= 0) {
+        this._selections.set(groupRoot, idx);
+      }
+    }
+  }
+
   private _onTreeAblyMessage(msg: Ably.InboundMessage): void {
     // Re-emit only if the message corresponds to a visible node
     const headers = getHeaders(msg);
@@ -481,7 +549,8 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
       this._emitter.emit('ably-message', msg);
       return;
     }
-    if (!this._withheldMsgIds.has(msgId)) {
+    // Check that msgId is on the visible branch and not withheld
+    if (this._lastVisibleIds.includes(msgId)) {
       this._emitter.emit('ably-message', msg);
     }
   }
@@ -498,11 +567,15 @@ export class DefaultView<TEvent, TMessage> implements View<TEvent, TMessage> {
     }
   }
 
-  private _visibleChanged(newIds: string[]): boolean {
+  private _visibleChanged(newIds: string[], newMessages: TMessage[]): boolean {
     if (newIds.length !== this._lastVisibleIds.length) return true;
     for (const [i, newId] of newIds.entries()) {
       if (newId !== this._lastVisibleIds[i]) return true;
     }
+    // Also detect in-place content updates (e.g. streaming) via reference comparison
+    for (const [i, msg] of newMessages.entries()) {
+      if (msg !== this._lastVisibleMessages[i]) return true;
+    }
     return false;
   }
 }

src/vercel/transport/chat-transport.ts

@@ -182,6 +182,7 @@ export const createChatTransport = (
       // Look up the parent of the message being regenerated.
       // messageId comes from useChat (UIMessage.id) — scan the flattened
       // nodes to find the one whose domain message matches this ID.
+      // Uses the transport's default view — ChatTransport is single-view (one useChat per channel).
       const node = transport.view.flattenNodes().find((n) => n.message.id === messageId);
       if (node) {
         // Use the tree node's msgId (x-ably-msg-id) as forkOf — this is