fix(sdk): stopping a turn is an owning-transport action, not an abort side effect (TRI-13070)

Author: kathiekiwi

.changeset/watch-mode-keepalive.md

@@ -3,3 +3,5 @@
 ---
 
 Watch-mode chat subscriptions now stay connected across quiet periods.
+
+Read-only chat subscriptions no longer stop a turn when they disconnect.

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

@@ -1334,6 +1334,94 @@ describe("TriggerChatTransport", () => {
     });
   });
 
+  describe("reconnectToStream stop-on-abort ownership (TRI-13070)", () => {
+    // A quiet stream: EOF, no records, never settled — the subscription
+    // stays alive (watch mode) so an abort mid-flight exercises the stop path.
+    function quietWatchTransport(): {
+      transport: TriggerChatTransport;
+      appends: () => number;
+    } {
+      let appendCount = 0;
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        if (isSessionStreamAppendUrl(urlStr)) {
+          appendCount++;
+          return defaultAppendResponse();
+        }
+        if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse([]);
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        watch: true,
+        sessions: { "chat-own": { publicAccessToken: "p", isStreaming: true } },
+      });
+      return { transport, appends: () => appendCount };
+    }
+
+    it("passive subscriber aborting writes no stop chunk to .in", async () => {
+      vi.useFakeTimers();
+      try {
+        const { transport, appends } = quietWatchTransport();
+        const abort = new AbortController();
+        const stream = await transport.reconnectToStream({
+          chatId: "chat-own",
+          abortSignal: abort.signal,
+        });
+        const drained = drainChunks(stream!);
+        await vi.advanceTimersByTimeAsync(1_000);
+        abort.abort();
+        await drained;
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appends()).toBe(0);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("owning subscriber with stopOnAbort:true sends a stop chunk on abort", async () => {
+      vi.useFakeTimers();
+      try {
+        const { transport, appends } = quietWatchTransport();
+        const abort = new AbortController();
+        const stream = await transport.reconnectToStream({
+          chatId: "chat-own",
+          abortSignal: abort.signal,
+          stopOnAbort: true,
+        });
+        const drained = drainChunks(stream!);
+        await vi.advanceTimersByTimeAsync(1_000);
+        abort.abort();
+        await drained;
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appends()).toBe(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("abortSignal presence alone (stopOnAbort unset) sends no stop", async () => {
+      vi.useFakeTimers();
+      try {
+        const { transport, appends } = quietWatchTransport();
+        const abort = new AbortController();
+        const stream = await transport.reconnectToStream({
+          chatId: "chat-own",
+          abortSignal: abort.signal,
+        });
+        const drained = drainChunks(stream!);
+        await vi.advanceTimersByTimeAsync(1_000);
+        abort.abort();
+        await drained;
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appends()).toBe(0);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+  });
+
   describe("multi-tab coordination", () => {
     it("isReadOnly defaults to false when multiTab is disabled", () => {
       const transport = new TriggerChatTransport({

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

@@ -873,7 +873,11 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     state.isStreaming = true;
     this.notifySessionChange(chatId, state);
 
-    return this.subscribeToSessionStream(state, abortSignal, chatId, { sinceInSeq: inSeq });
+    // Owning turn: aborting this live send stops the turn the user drives.
+    return this.subscribeToSessionStream(state, abortSignal, chatId, {
+      sinceInSeq: inSeq,
+      sendStopOnAbort: true,
+    });
   };
 
   /**
@@ -1146,6 +1150,13 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     options: {
       chatId: string;
       abortSignal?: AbortSignal | undefined;
+      /**
+       * Whether aborting this subscription sends `{kind:"stop"}` on `.in`.
+       * A subscription ending is not session ownership — a passive/watch
+       * reader unmounting must never stop a turn it doesn't drive. Only
+       * pass `true` from a caller that owns the live turn. @default false
+       */
+      stopOnAbort?: boolean;
     } & ChatRequestOptions
   ): Promise<ReadableStream<UIMessageChunk> | null> => {
     const state = this.sessions.get(options.chatId);
@@ -1163,7 +1174,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
 
     return this.subscribeToSessionStream(state, abortSignal, options.chatId, {
       resumed: true,
-      sendStopOnAbort: !!options.abortSignal,
+      sendStopOnAbort: options.stopOnAbort ?? false,
       // Reconnect-on-reload opts into the server's settled-peek shortcut
       // so the SSE doesn't hang for 60s when no turn is in flight. Active
       // send-a-message paths must keep wait=60 to avoid racing the
@@ -1266,7 +1277,11 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     state.isStreaming = true;
     this.notifySessionChange(chatId, state);
 
-    return this.subscribeToSessionStream(state, undefined, chatId, { sinceInSeq: inSeq });
+    // Owning action: aborting this send stops the turn the user drives.
+    return this.subscribeToSessionStream(state, undefined, chatId, {
+      sinceInSeq: inSeq,
+      sendStopOnAbort: true,
+    });
   };
 
   // -------------------------------------------------------------------------