fix(sdk,core): harden chat.agent message delivery and idempotency

Stop delivering a user message twice when it arrives mid-stream: the
session stream manager now lets a handler consume a record so it is not
also buffered for the next turn, which previously re-ran the message as
a duplicate turn. Input appends carry an X-Part-Id idempotency key so a
retried send cannot duplicate a message. Stopping a generation clears
the streaming state and persists it, so a page reload no longer replays
the stopped turn. Promoting a queued message to steering no longer sends
inside a React state updater. Runs keep up to the full tag limit instead
of being silently truncated. The in-memory test stream manager now
mirrors the production consume semantics so this class of bug is covered.

Author: ericallam

.changeset/chat-agent-hardening.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Reliability fixes for `chat.agent`. A user message sent while the agent is streaming is no longer delivered twice (which could run a duplicate turn), input appends now carry an idempotency key so a retried send can't duplicate a message, stopping a generation clears the streaming state so a page reload doesn't replay the stopped turn, and runs can now carry the full set of dashboard tags instead of being silently truncated.

packages/core/src/v3/apiClient/index.ts

@@ -1,3 +1,4 @@
+import { nanoid } from "nanoid";
 import { z } from "zod";
 import { VERSION } from "../../version.js";
 import { generateJWT } from "../jwt.js";
@@ -1276,12 +1277,16 @@ export class ApiClient {
     part: TBody,
     requestOptions?: ZodFetchOptions
   ) {
+    // Generated once per logical append, outside zodfetch, so its internal
+    // retries reuse the same part id and the server-side dedupe collapses a
+    // retried POST whose first attempt actually committed.
+    const partId = nanoid(7);
     return zodfetch(
       AppendToStreamResponseBody,
       `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}/append`,
       {
         method: "POST",
-        headers: this.#getHeaders(false),
+        headers: { ...this.#getHeaders(false), "X-Part-Id": partId },
         body: part,
       },
       mergeRequestOptions(this.defaultRequestOptions, requestOptions)

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

@@ -34,7 +34,7 @@ export class SessionStreamsAPI implements SessionStreamManager {
   public on(
     sessionId: string,
     io: SessionChannelIO,
-    handler: (data: unknown) => void | Promise<void>
+    handler: (data: unknown) => void | boolean | Promise<void>
   ): { off: () => void } {
     return this.#getManager().on(sessionId, io, handler);
   }

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

@@ -9,7 +9,11 @@ import { computeReconnectDelayMs } from "../utils/reconnectBackoff.js";
 import { SessionChannelIO, SessionStreamManager } from "./types.js";
 import { controlSubtype } from "./wireProtocol.js";
 
-type SessionStreamHandler = (data: unknown) => void | Promise<void>;
+// A handler that synchronously returns `true` CONSUMES the record: it is
+// not buffered for a later `once()` and the committed-consume cursor
+// advances past it. Anything else (void, a Promise) leaves the record
+// available to other consumers. See `SessionStreamManager.on` in types.ts.
+type SessionStreamHandler = (data: unknown) => void | boolean | Promise<void>;
 
 type OnceWaiter = {
   resolve: (result: InputStreamOnceResult<unknown>) => void;
@@ -113,30 +117,41 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     this.explicitlyDisconnected.delete(key);
     this.#ensureTailConnected(sessionId, io);
 
+    // Selective drain: offer each buffered record to the new handler and
+    // remove ONLY the ones it consumed (returned `true` — e.g. the
+    // messages facade for message-kind records). Consumed records advance
+    // the committed-consume cursor, so a worker using `messagesInput.on()`
+    // for user-message delivery persists a `.in` cursor that matches what
+    // the handler processed. Records the handler did not consume (other
+    // kinds) STAY buffered for a future `once()` or a different handler —
+    // a blind drain here either swallowed them (delivered to a handler
+    // that filtered them out, then deleted) or re-delivered already-
+    // processed messages into every newly attached per-turn handler,
+    // duplicating turns.
     const buffered = this.buffer.get(key);
     if (buffered && buffered.length > 0) {
-      for (const data of buffered) {
-        this.#invokeHandler(handler, data);
-      }
-      // Advance the committed-consume cursor to the highest seq drained
-      // into the new handler. `on()`-drain removes the records from the
-      // buffer, so they're no longer available to a future `once()` —
-      // from the manager's perspective they've been consumed. Without
-      // this, a worker that uses `messagesInput.on()` for user-message
-      // delivery (pendingMessages mode) would persist a `.in` cursor
-      // that lags behind the records the handler already processed, and
-      // the next boot would re-deliver them.
-      const seqList = this.bufferSeqNums.get(key);
-      if (seqList) {
-        for (const s of seqList) {
+      const seqList = this.bufferSeqNums.get(key) ?? [];
+      const keptRecords: unknown[] = [];
+      // Kept in lock-step with `keptRecords` — drifting lengths would map
+      // seq_nums to the wrong records on subsequent shifts.
+      const keptSeqNums: Array<number | undefined> = [];
+      for (let i = 0; i < buffered.length; i++) {
+        const consumed = this.#invokeHandler(handler, buffered[i]);
+        if (consumed) {
+          const s = seqList[i];
           if (s !== undefined) this.#advanceLastDispatched(key, s);
+        } else {
+          keptRecords.push(buffered[i]);
+          keptSeqNums.push(seqList[i]);
         }
       }
-      this.buffer.delete(key);
-      // Keep `bufferSeqNums` in lock-step with `buffer` — without this,
-      // the parallel array desyncs and the next `#dispatch` that buffers
-      // a record would shift a stale seqNum into `lastDispatchedSeqNum`.
-      this.bufferSeqNums.delete(key);
+      if (keptRecords.length > 0) {
+        this.buffer.set(key, keptRecords);
+        this.bufferSeqNums.set(key, keptSeqNums);
+      } else {
+        this.buffer.delete(key);
+        this.bufferSeqNums.delete(key);
+      }
     }
 
     return {
@@ -509,13 +524,21 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       return;
     }
 
-    // Persistent handlers (e.g. `stopInput.on(...)`) get a copy of the chunk,
-    // but they don't "consume" it — handlers usually filter by `kind` and
-    // ignore chunks they don't care about. Buffer the chunk regardless so a
-    // subsequent `once()` (e.g. `messagesInput.waitWithIdleTimeout` in
-    // chat.agent's preload) can still pick up the same chunk that arrived
-    // before its waiter was registered.
-    this.#invokeHandlers(key, data);
+    // Persistent handlers get a copy of the chunk. A handler that
+    // synchronously returns `true` CONSUMES it (e.g. the messages facade
+    // for message-kind records): the record must not also be buffered, or
+    // the next `on()` attach / `once()` would deliver it a second time —
+    // in chat.agent's turn loop that duplicated user messages into a
+    // second turn. Records no handler consumed (e.g. a message arriving
+    // while only the stop facade is attached during preload) are buffered
+    // so a subsequent `once()` can still pick them up.
+    const consumed = this.#invokeHandlers(key, data);
+    if (consumed) {
+      if (seqNum !== undefined) {
+        this.#advanceLastDispatched(key, seqNum);
+      }
+      return;
+    }
 
     let buffered = this.buffer.get(key);
     if (!buffered) {
@@ -535,17 +558,24 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     bufferedSeqs.push(seqNum);
   }
 
-  #invokeHandlers(key: string, data: unknown): void {
+  /** Returns true when any handler consumed the record. All handlers are invoked regardless. */
+  #invokeHandlers(key: string, data: unknown): boolean {
     const handlers = this.handlers.get(key);
-    if (!handlers) return;
+    if (!handlers) return false;
+    let consumed = false;
     for (const handler of handlers) {
-      this.#invokeHandler(handler, data);
+      if (this.#invokeHandler(handler, data)) {
+        consumed = true;
+      }
     }
+    return consumed;
   }
 
-  #invokeHandler(handler: SessionStreamHandler, data: unknown): void {
+  /** Returns true when the handler synchronously consumed the record (returned `true`). */
+  #invokeHandler(handler: SessionStreamHandler, data: unknown): boolean {
     try {
       const result = handler(data);
+      if (result === true) return true;
       if (result && typeof result === "object" && "catch" in result) {
         (result as Promise<void>).catch((error) => {
           if (this.debug) {
@@ -558,6 +588,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
         console.error("[SessionStreamManager] Handler error:", error);
       }
     }
+    return false;
   }
 
   #removeOnceWaiter(key: string, waiter: OnceWaiter): void {

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

@@ -6,7 +6,7 @@ export class NoopSessionStreamManager implements SessionStreamManager {
   on(
     _sessionId: string,
     _io: SessionChannelIO,
-    _handler: (data: unknown) => void | Promise<void>
+    _handler: (data: unknown) => void | boolean | Promise<void>
   ): { off: () => void } {
     return { off: () => {} };
   }

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

@@ -22,11 +22,20 @@ export type SessionChannelIO = "out" | "in";
  * `.on` / `.once` / `.peek` / `.wait` / `.waitWithIdleTimeout`.
  */
 export interface SessionStreamManager {
-  /** Register a handler that fires every time data arrives on the given channel. */
+  /**
+   * Register a handler that fires every time data arrives on the given channel.
+   *
+   * A handler that synchronously returns `true` CONSUMES the record: it is
+   * not buffered for a later `once()` and the committed-consume cursor
+   * advances past it. Any other return value (including a Promise) leaves
+   * the record available to other consumers. Kind-filtering facades return
+   * `true` for the kinds they own so the same record is never delivered
+   * twice — once to the handler and again via a buffer drain.
+   */
   on(
     sessionId: string,
     io: SessionChannelIO,
-    handler: (data: unknown) => void | Promise<void>
+    handler: (data: unknown) => void | boolean | Promise<void>
   ): { off: () => void };
 
   /** Wait for the next record on the given channel (buffered or live). */

packages/core/src/v3/test/test-session-stream-manager.ts

@@ -16,7 +16,10 @@ type OnceWaiter = {
   abortHandler?: () => void;
 };
 
-type Handler = (data: unknown) => void | Promise<void>;
+// Same contract as the production manager: a handler that synchronously
+// returns `true` CONSUMES the record (not buffered, not re-delivered on a
+// future `on()` attach). See `SessionStreamManager.on` in types.ts.
+type Handler = (data: unknown) => void | boolean | Promise<void>;
 
 function keyFor(sessionId: string, io: SessionChannelIO): string {
   return `${sessionId}:${io}`;
@@ -51,20 +54,32 @@ export class TestSessionStreamManager implements SessionStreamManager {
     }
     set.add(handler);
 
-    // Note: we intentionally do NOT replay buffered records into the
-    // newly-registered handler, and we do NOT drain the buffer. The
-    // buffer is owned by `once()` — registering a passive observer
-    // (`on`) must not consume records destined for a future `once`
-    // waiter. This matches production SSE semantics where handlers
-    // observe records as they arrive, not retroactively.
-    //
-    // Earlier versions drained the buffer here, which caused user
-    // messages buffered during the runtime's `runFn` boot phase to be
-    // silently swallowed by the `stopInput.on()` handler registered at
-    // ai.ts:4806 (the stop handler ignores `kind: "message"` chunks).
-    // The next `messagesInput.waitWithIdleTimeout` then waited 30s for
-    // a record that had already been "delivered" to a handler that
-    // didn't want it.
+    // Selective drain, matching the production manager: offer each
+    // buffered record to the new handler and remove ONLY the ones it
+    // consumed (returned `true`). Records the handler filtered out (other
+    // kinds) stay buffered for a future `once()`. This is the corrected
+    // form of two historical bugs: a blind drain swallowed boot-phase user
+    // messages into the stop facade (which ignores `kind: "message"`),
+    // and no-drain-at-all let production re-deliver already-processed
+    // messages into every newly attached per-turn handler.
+    const buffered = this.buffer.get(key);
+    if (buffered && buffered.length > 0) {
+      const kept: unknown[] = [];
+      for (const data of buffered) {
+        let consumed = false;
+        try {
+          consumed = handler(data) === true;
+        } catch {
+          // Never let a handler error break test state
+        }
+        if (!consumed) kept.push(data);
+      }
+      if (kept.length > 0) {
+        this.buffer.set(key, kept);
+      } else {
+        this.buffer.delete(key);
+      }
+    }
 
     return {
       off: () => {
@@ -212,20 +227,20 @@ export class TestSessionStreamManager implements SessionStreamManager {
   /**
    * Push a record onto the given channel.
    *
-   * Dispatch rules — similar to the production manager, but with a tweak
-   * that makes unit tests deterministic:
+   * Dispatch rules — same as the production manager:
+   *
+   * 1. **A pending `.once` waiter consumes first.** Handlers still observe
+   *    a copy.
+   * 2. **Otherwise handlers observe.** A handler that synchronously
+   *    returns `true` consumes the record (kind-filtering facades do this
+   *    for the kinds they own) — it is NOT buffered.
+   * 3. **Records no one consumed are buffered** for the next `.once` call
+   *    or the next consuming `on()` attach.
    *
-   * 1. **Handlers always observe** (like production). A session-level `.on`
-   *    is a filter-observer — it fires every time a record arrives,
-   *    regardless of whether a `.once` waiter is also active.
-   * 2. **First waiter consumes** the record if present (like production).
-   * 3. **If no waiter, the record is buffered for the next `.once` call.**
-   *    Production discards records that only match handlers — but in
-   *    production the SSE tail introduces enough latency that the next
-   *    `.once` is usually registered before the next record arrives. Tests
-   *    send synchronously right after `turn-complete`, so without this
-   *    buffer the next `waitWithIdleTimeout` would race and lose the
-   *    message. The buffer is the only deviation from production semantics.
+   * Handler promises are awaited before resolving so test code can rely
+   * on async handler work having settled by the time `__sendFromTest`
+   * resolves. Consumption is decided on the synchronous return value,
+   * exactly like production.
    */
   async __sendFromTest(
     sessionId: string,
@@ -234,23 +249,6 @@ export class TestSessionStreamManager implements SessionStreamManager {
   ): Promise<void> {
     const key = keyFor(sessionId, io);
 
-    const handlers = this.handlers.get(key);
-    if (handlers && handlers.size > 0) {
-      // Awaited so test code can rely on handlers having completed by the
-      // time `__sendFromTest` resolves. Wrapped per-handler so a
-      // throwing/rejecting handler doesn't poison Promise.all and break
-      // unrelated test state.
-      await Promise.all(
-        Array.from(handlers).map(async (h) => {
-          try {
-            await h(data);
-          } catch {
-            // Never let a handler error break test state
-          }
-        })
-      );
-    }
-
     const waiters = this.onceWaiters.get(key);
     if (waiters && waiters.length > 0) {
       const w = waiters.shift()!;
@@ -260,6 +258,27 @@ export class TestSessionStreamManager implements SessionStreamManager {
         w.signal.removeEventListener("abort", w.abortHandler);
       }
       w.resolve({ ok: true, output: data });
+      await this.#invokeHandlers(key, data);
+      return;
+    }
+
+    const consumed = await this.#invokeHandlers(key, data);
+    if (consumed) return;
+
+    // Re-check waiters: handler invocation above is awaited (unlike the
+    // synchronous production dispatch), and the runtime commonly registers
+    // its next `once()` during that window — e.g. the turn loop reaching
+    // `waitWithIdleTimeout` while a handler settles. Without this second
+    // look the record would be buffered while the fresh waiter hangs.
+    const lateWaiters = this.onceWaiters.get(key);
+    if (lateWaiters && lateWaiters.length > 0) {
+      const w = lateWaiters.shift()!;
+      if (lateWaiters.length === 0) this.onceWaiters.delete(key);
+      if (w.timer) clearTimeout(w.timer);
+      if (w.signal && w.abortHandler) {
+        w.signal.removeEventListener("abort", w.abortHandler);
+      }
+      w.resolve({ ok: true, output: data });
       return;
     }
 
@@ -271,6 +290,34 @@ export class TestSessionStreamManager implements SessionStreamManager {
     buffered.push(data);
   }
 
+  /**
+   * Invoke all handlers; resolves once any returned promises settle.
+   * Returns true when any handler synchronously consumed the record.
+   * Wrapped per-handler so a throwing/rejecting handler doesn't poison
+   * Promise.all and break unrelated test state.
+   */
+  async #invokeHandlers(key: string, data: unknown): Promise<boolean> {
+    const handlers = this.handlers.get(key);
+    if (!handlers || handlers.size === 0) return false;
+
+    let consumed = false;
+    await Promise.all(
+      Array.from(handlers).map(async (h) => {
+        try {
+          const result = h(data);
+          if (result === true) {
+            consumed = true;
+            return;
+          }
+          await result;
+        } catch {
+          // Never let a handler error break test state
+        }
+      })
+    );
+    return consumed;
+  }
+
   /**
    * Immediately resolve every pending `once()` waiter for the given channel
    * with a timeout error. Simulates a closed stream (e.g. session closed).