fix(chat): avoid settled peek for active reconnects

Author: gtremper

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

@@ -1207,12 +1207,12 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       sendStopOnAbort: options.stopOnAbort ?? false,
       sinceInSeq: state.activeInputSeq,
       // 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
-      // freshly-triggered turn's first chunk. Watch mode must NOT peek: a
-      // settled peek between turns sets sessionSettled and closes the
+      // so the SSE doesn't hang for 60s when no turn is in flight. A known
+      // active input must not peek because the previous turn's completion
+      // can remain at the tail until the current turn writes its first chunk.
+      // Watch mode must NOT peek: a settled peek between turns closes the
       // standing subscription, so the viewer never sees the next turn.
-      peekSettled: !this.watchMode,
+      peekSettled: !this.watchMode && state.activeInputSeq === undefined,
     });
   };
 

packages/trigger-sdk/test/chat-turn-correlation.test.ts

@@ -165,6 +165,45 @@ describe("transport turn correlation", () => {
     expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
   });
 
+  it("does not request a settled peek while reconnecting a known active input", async () => {
+    vi.useFakeTimers();
+    try {
+      const subscribeHeaders: Headers[] = [];
+      const transport = new TriggerChatTransport({
+        task: "test-task",
+        accessToken: async () => "tok_test",
+        sessions: {
+          c1: { publicAccessToken: "tok_test", isStreaming: true, activeInputSeq: 5 },
+        },
+        fetch: async (_url, init) => {
+          const headers = new Headers(init?.headers);
+          subscribeHeaders.push(headers);
+
+          if (subscribeHeaders.length === 1) {
+            // Match the server shortcut: a peek sees the previous turn's
+            // boundary at the tail and marks this otherwise-normal EOF settled.
+            return batchResponse([turnComplete(10, 4)], headers.has("X-Peek-Settled"));
+          }
+
+          return batchResponse([textDelta(11, "current"), turnComplete(12, 5)]);
+        },
+      });
+
+      const stream = await transport.reconnectToStream({ chatId: "c1" });
+
+      expect(stream).not.toBeNull();
+      const deltas = readDeltas(stream!);
+      await vi.advanceTimersByTimeAsync(1_000);
+      await expect(deltas).resolves.toEqual(["current"]);
+      expect(subscribeHeaders).toHaveLength(2);
+      expect(subscribeHeaders[0]?.get("X-Peek-Settled")).toBeNull();
+      expect(transport.getSession("c1")?.isStreaming).toBe(false);
+      expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("keeps turn correlation through a lost append response, retry, and reload", async () => {
     const committed = new Map<string, number>();
     const appendPartIds: string[] = [];