fix(core,sdk,cli): handover-hang, buffer-seq desync, MCP legacy filter

Three follow-ups from PR review.

`chat-server.ts stitchHandoverStream` was blocking ~60s after turn-
complete because control records never enqueue into the SSE data
stream, so the `for await` loop had no way to wake up and S2's long-
poll held the connection open until it timed out. A dedicated
`subscriptionAbort` AbortController is now aborted from `onControl`
when turn-complete fires, the for-await exits via AbortError, and the
catch treats that as the expected exit path so the surrounding
session-state emission and close still run.

`SessionStreamManager` was breaking its parallel-array invariant on
the (theoretical) `undefined`-seqNum path: a record with NaN-parsing
`part.id` would push to `buffer` but skip `bufferSeqNums`, drifting
the arrays so the next `once()` / `shiftBuffer` shifted a stale
seqNum into `lastDispatchedSeqNum`. `bufferSeqNums` now stores
`number | undefined` and always pushes; the shift sites already gate
the cursor advance on `seqNum !== undefined`.

`cli-v3/mcp/tools/agentChat.ts` picks up the same belt-and-suspenders
filter the dashboard's `AgentView.tsx` already carries: skip any
data chunk whose `type` starts with `trigger:`. Covers any in-flight
session whose `.out` was populated by an older SDK that emitted
turn-complete / upgrade-required as data records.

Author: ericallam

packages/cli-v3/src/mcp/tools/agentChat.ts

@@ -443,6 +443,15 @@ async function collectAgentResponse(
       if (value.chunk != null && typeof value.chunk === "object") {
         const chunk = value.chunk as Record<string, unknown>;
 
+        // Legacy belt-and-suspenders: prior SDK versions emitted
+        // `trigger:turn-complete` / `trigger:upgrade-required` as
+        // data records (`chunk.type`) instead of header-form control
+        // records. Filter them so an in-flight session whose `.out`
+        // was populated by an older agent doesn't stall this loop.
+        if (typeof chunk.type === "string" && chunk.type.startsWith("trigger:")) {
+          continue;
+        }
+
         if (chunk.type === "text-delta" && typeof chunk.delta === "string") {
           text += chunk.delta;
           // Accumulate into a text part

packages/core/src/v3/sessionStreams/manager.ts

@@ -49,7 +49,13 @@ export class StandardSessionStreamManager implements SessionStreamManager {
   // a buffered record into a waiter, the cursor (`lastDispatchedSeqNums`)
   // can advance to that record's seq. Kept as a separate map so the
   // existing `peek()` shape (returns `unknown`) stays unchanged.
-  private bufferSeqNums = new Map<string, number[]>();
+  //
+  // Entries are `number | undefined` so the array stays length-locked
+  // with `buffer` even if a record arrives without a parseable seq —
+  // shifting `undefined` is just a no-op for the cursor advance, but
+  // the slot still gets consumed. Drifting lengths would map seq_nums
+  // to the wrong records on subsequent shifts.
+  private bufferSeqNums = new Map<string, Array<number | undefined>>();
   private tails = new Map<string, TailState>();
   // Per-stream lower-bound timestamp filter. When set, records whose
   // SSE timestamp is <= the bound are dropped before dispatch — used by
@@ -503,14 +509,16 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       this.buffer.set(key, buffered);
     }
     buffered.push(data);
-    if (seqNum !== undefined) {
-      let bufferedSeqs = this.bufferSeqNums.get(key);
-      if (!bufferedSeqs) {
-        bufferedSeqs = [];
-        this.bufferSeqNums.set(key, bufferedSeqs);
-      }
-      bufferedSeqs.push(seqNum);
+    let bufferedSeqs = this.bufferSeqNums.get(key);
+    if (!bufferedSeqs) {
+      bufferedSeqs = [];
+      this.bufferSeqNums.set(key, bufferedSeqs);
     }
+    // Always push, even when `seqNum` is undefined (e.g. NaN from a
+    // malformed `part.id`). Skipping the push here would drift the two
+    // arrays apart and misattribute seq_nums to records on the next
+    // shift.
+    bufferedSeqs.push(seqNum);
   }
 
   #invokeHandlers(key: string, data: unknown): void {

packages/trigger-sdk/src/v3/chat-server.ts

@@ -567,14 +567,21 @@ async function openHandoverSession(opts: {
           // legacy chunk shape for the customer-server-to-browser hop).
           let latestEventId: string | undefined;
           let turnComplete = false;
+          // Dedicated abort signal for this agent subscription. Aborted
+          // from `onControl` the moment turn-complete fires so the
+          // `for await` loop below exits immediately instead of blocking
+          // until S2's long-poll closes (~60s). Combined with the outer
+          // `abortController.signal` via `AbortSignal.any` so a request-
+          // wide abort still tears the subscription down.
+          const subscriptionAbort = new AbortController();
           const agentStream = await apiClient.subscribeToSessionStream<UIMessageChunk>(
             chatId,
             "out",
             {
               ...(customerLastEventId != null
                 ? { lastEventId: customerLastEventId }
                 : {}),
-              signal: abortController.signal,
+              signal: AbortSignal.any([abortController.signal, subscriptionAbort.signal]),
               onPart: (part) => {
                 if (part.id) latestEventId = part.id;
               },
@@ -589,18 +596,28 @@ async function openHandoverSession(opts: {
                   controller.enqueue({
                     type: "trigger:turn-complete",
                   } as unknown as UIMessageChunk);
+                  // Stop the SSE read now. Without this the `for await`
+                  // can't see the control event (control records are
+                  // never enqueued into the data stream) and would idle
+                  // until S2's long-poll timeout closes the connection.
+                  subscriptionAbort.abort();
                 }
               },
             }
           );
 
-          for await (const chunk of agentStream) {
-            // Data records only — control records are routed via
-            // `onControl` above. Stop reading as soon as we see the
-            // turn-complete control event (the loop may have one more
-            // data record buffered, but that's fine — we break out).
-            controller.enqueue(chunk);
-            if (turnComplete) break;
+          try {
+            for await (const chunk of agentStream) {
+              // Data records only — control records are routed via
+              // `onControl` above and trigger the subscription abort.
+              controller.enqueue(chunk);
+              if (turnComplete) break;
+            }
+          } catch (err) {
+            // AbortError from `subscriptionAbort` is the expected exit
+            // path once turn-complete fires; surface anything else.
+            const isAbort = err instanceof Error && err.name === "AbortError";
+            if (!isAbort || !turnComplete) throw err;
           }
 
           // Final control chunk: hand the browser transport the