fix(sdk): more PR review findings (round 3) — reconnect backoff + design-note divergence

Three more #3542 review fixes addressing the design-question bucket.

sessionStreams/manager.ts + inputStreams/manager.ts — both #runTail
loops swallowed errors and the .finally reconnected immediately
whenever hasHandlers || hasWaiters. A persistent backend failure
(auth rejection, 5xx, DNS) would reconnect in a tight loop with no
rate limiting. Both managers now exponentially back off: 1s base,
doubling per attempt, capped at 30s, plus 0–1s jitter. A
reconnectAttempts counter resets to 0 on every successful #dispatch
(any record flowing through = healthy connection), so transient
blips don't accumulate. Per-waiter timeouts still bound how long any
once() waits regardless.

realtimeStreams/streamsWriterV2.ts + .test.ts — extracted the
size-check + discriminant-extraction logic into encodeChunkOrError,
a pure helper. Tests now exercise it directly, no `vi.mock("@s2-dev/
streamstore")` shim. The original vi.mock conflicted with the
codebase rule of using testcontainers / not mocking; the new tests
are framework-pure and faster.

trigger-sdk/v3/shared.ts — added an in-code comment in
triggerAndSubscribe explaining the error-shape divergence from
triggerAndWait. The SerializedError surfaced by subscribeToRun
strips the TaskRunError type discriminator at the server boundary
(createJsonErrorObject in errors.ts:274), so the SDK can't
reconstruct the discriminator on the receive side. Callers needing
exact error-type matching should use triggerAndWait.

Author: ericallam

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

@@ -6,6 +6,7 @@ import {
   InputStreamTimeoutError,
 } from "./types.js";
 import { InputStreamOnceOptions } from "../realtimeStreams/types.js";
+import { computeReconnectDelayMs } from "../utils/reconnectBackoff.js";
 
 type InputStreamHandler = (data: unknown) => void | Promise<void>;
 
@@ -29,6 +30,12 @@ export class StandardInputStreamManager implements InputStreamManager {
   private seqNums = new Map<string, number>();
   private currentRunId: string | null = null;
   private streamsVersion: string | undefined;
+  // Reconnect attempt counter per streamId. Drives the exponential
+  // backoff applied by `#ensureStreamTailConnected`'s `.finally` so a
+  // persistent backend failure (auth rejection, 5xx, DNS, etc.) doesn't
+  // reconnect in a tight loop. Reset to 0 by `#dispatch` whenever a
+  // record flows through.
+  private reconnectAttempts = new Map<string, number>();
 
   constructor(
     private apiClient: ApiClient,
@@ -204,6 +211,7 @@ export class StandardInputStreamManager implements InputStreamManager {
     this.streamsVersion = undefined;
     this.seqNums.clear();
     this.handlers.clear();
+    this.reconnectAttempts.clear();
 
     // Reject all pending once waiters
     for (const [, waiters] of this.onceWaiters) {
@@ -238,13 +246,29 @@ export class StandardInputStreamManager implements InputStreamManager {
         .finally(() => {
           this.tails.delete(streamId);
 
-          // Auto-reconnect if there are still active handlers or waiters
+          // Auto-reconnect with exponential backoff if there are still
+          // active handlers or waiters. Without backoff a persistent
+          // failure (auth rejected, 5xx, DNS) would reconnect in a tight
+          // loop because `#runTail`'s error path only logs. `#dispatch`
+          // resets the counter on every successful record.
           const hasHandlers =
             this.handlers.has(streamId) && this.handlers.get(streamId)!.size > 0;
           const hasWaiters =
             this.onceWaiters.has(streamId) && this.onceWaiters.get(streamId)!.length > 0;
           if (hasHandlers || hasWaiters) {
-            this.#ensureStreamTailConnected(streamId);
+            const attempt = this.reconnectAttempts.get(streamId) ?? 0;
+            this.reconnectAttempts.set(streamId, attempt + 1);
+            const delayMs = computeReconnectDelayMs(attempt);
+            setTimeout(() => {
+              if (this.tails.has(streamId)) return;
+              const stillHasHandlers =
+                this.handlers.has(streamId) && this.handlers.get(streamId)!.size > 0;
+              const stillHasWaiters =
+                this.onceWaiters.has(streamId) &&
+                this.onceWaiters.get(streamId)!.length > 0;
+              if (!stillHasHandlers && !stillHasWaiters) return;
+              this.#ensureStreamTailConnected(streamId);
+            }, delayMs);
           }
         });
       this.tails.set(streamId, { abortController, promise });
@@ -310,6 +334,10 @@ export class StandardInputStreamManager implements InputStreamManager {
   }
 
   #dispatch(streamId: string, data: unknown): void {
+    // Any record flowing through = healthy connection; reset the backoff
+    // counter so the next disconnect starts fresh.
+    this.reconnectAttempts.delete(streamId);
+
     // First try to resolve a once waiter
     const waiters = this.onceWaiters.get(streamId);
     if (waiters && waiters.length > 0) {

packages/core/src/v3/realtimeStreams/streamsWriterV2.test.ts

@@ -1,150 +1,77 @@
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { describe, expect, it } from "vitest";
 
 import { ChatChunkTooLargeError, isChatChunkTooLargeError } from "../errors.js";
+import { encodeChunkOrError } from "./streamsWriterV2.js";
 
-const lastAckedPosition = vi.fn(() => undefined);
-
-const appendSession = vi.fn(async () => {
-  // A WritableStream that just consumes records — we never reach S2 because
-  // the size check fires upstream of this for the oversize case, but we still
-  // need a valid writable for the small-chunk path.
-  const writable = new WritableStream<unknown>({});
-  return {
-    writable,
-    lastAckedPosition,
-  };
-});
-
-vi.mock("@s2-dev/streamstore", async (importOriginal) => {
-  const actual = await importOriginal<typeof import("@s2-dev/streamstore")>();
-  return {
-    ...actual,
-    S2: class FakeS2 {
-      basin() {
-        return {
-          stream: () => ({
-            appendSession,
-          }),
-        };
-      }
-    },
-  };
-});
-
-import { StreamsWriterV2 } from "./streamsWriterV2.js";
-
-afterEach(() => {
-  vi.clearAllMocks();
-});
+// The size cap and discriminant extraction are the only S2-independent bits
+// of `StreamsWriterV2` that benefit from unit coverage. Both live in the
+// `encodeChunkOrError` pure helper, so the tests exercise it directly — no
+// `vi.mock("@s2-dev/streamstore", ...)` shim needed.
 
-describe("StreamsWriterV2", () => {
-  it("rejects with ChatChunkTooLargeError when a single chunk exceeds the per-record cap", async () => {
+describe("encodeChunkOrError", () => {
+  it("flags oversize chunks and carries the chunk's `type` discriminant", () => {
     const oversized = {
       type: "tool-output-available",
       output: { text: "x".repeat(2_000_000) },
     };
-    const source = new ReadableStream<unknown>({
-      start(controller) {
-        controller.enqueue(oversized);
-        controller.close();
-      },
-    });
-
-    const writer = new StreamsWriterV2({
-      basin: "test",
-      stream: "test",
-      accessToken: "test",
-      source,
-    });
-
-    await expect(writer.wait()).rejects.toBeInstanceOf(ChatChunkTooLargeError);
-
-    let captured: unknown;
-    try {
-      await writer.wait();
-    } catch (err) {
-      captured = err;
-    }
-    expect(isChatChunkTooLargeError(captured)).toBe(true);
-    const e = captured as ChatChunkTooLargeError;
-    expect(e.chunkType).toBe("tool-output-available");
-    expect(e.chunkSize).toBeGreaterThan(1_000_000);
-    expect(e.maxSize).toBe(1024 * 1024 - 1024);
-    expect(e.message).toMatch(/tool-output-available/);
-    expect(e.message).toMatch(/chat\.agent chunk/);
+
+    const result = encodeChunkOrError(oversized);
+
+    expect(result.ok).toBe(false);
+    if (result.ok) return; // type guard
+    expect(isChatChunkTooLargeError(result.error)).toBe(true);
+    expect(result.error.chunkType).toBe("tool-output-available");
+    expect(result.error.chunkSize).toBeGreaterThan(1_000_000);
+    expect(result.error.maxSize).toBe(1024 * 1024 - 1024);
+    expect(result.error.message).toMatch(/tool-output-available/);
+    expect(result.error.message).toMatch(/chat\.agent chunk/);
   });
 
-  it("uses chunk.kind when chunk.type is missing (ChatInputChunk-style)", async () => {
-    const oversized = {
-      kind: "action",
-      payload: "x".repeat(2_000_000),
-    };
-    const source = new ReadableStream<unknown>({
-      start(controller) {
-        controller.enqueue(oversized);
-        controller.close();
-      },
-    });
-
-    const writer = new StreamsWriterV2({
-      basin: "test",
-      stream: "test",
-      accessToken: "test",
-      source,
-    });
-
-    let captured: unknown;
-    try {
-      await writer.wait();
-    } catch (err) {
-      captured = err;
-    }
-    expect(isChatChunkTooLargeError(captured)).toBe(true);
-    expect((captured as ChatChunkTooLargeError).chunkType).toBe("action");
+  it("falls back to chunk.kind when chunk.type is missing (ChatInputChunk-style)", () => {
+    const oversized = { kind: "action", payload: "x".repeat(2_000_000) };
+
+    const result = encodeChunkOrError(oversized);
+
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.error.chunkType).toBe("action");
   });
 
-  it("omits chunkType when chunk has no discriminant", async () => {
+  it("omits chunkType when the chunk has no discriminant", () => {
     const oversized = "x".repeat(2_000_000);
-    const source = new ReadableStream<unknown>({
-      start(controller) {
-        controller.enqueue(oversized);
-        controller.close();
-      },
-    });
-
-    const writer = new StreamsWriterV2({
-      basin: "test",
-      stream: "test",
-      accessToken: "test",
-      source,
-    });
-
-    let captured: unknown;
-    try {
-      await writer.wait();
-    } catch (err) {
-      captured = err;
-    }
-    expect(isChatChunkTooLargeError(captured)).toBe(true);
-    expect((captured as ChatChunkTooLargeError).chunkType).toBeUndefined();
+
+    const result = encodeChunkOrError(oversized);
+
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.error.chunkType).toBeUndefined();
   });
 
-  it("does not reject for chunks under the cap", async () => {
+  it("returns the encoded body for chunks under the cap", () => {
     const small = { type: "text-delta", delta: "hello" };
-    const source = new ReadableStream<unknown>({
-      start(controller) {
-        controller.enqueue(small);
-        controller.close();
-      },
-    });
-
-    const writer = new StreamsWriterV2({
-      basin: "test",
-      stream: "test",
-      accessToken: "test",
-      source,
-    });
-
-    await expect(writer.wait()).resolves.toBeDefined();
+
+    const result = encodeChunkOrError(small);
+
+    expect(result.ok).toBe(true);
+    if (!result.ok) return;
+    const parsed = JSON.parse(result.body) as { data: unknown; id: string };
+    expect(parsed.data).toEqual(small);
+    expect(parsed.id).toMatch(/^[A-Za-z0-9_-]{7}$/); // nanoid(7)
+  });
+});
+
+// Cross-check the ChatChunkTooLargeError type-guard helper itself. Trivial,
+// but keeps the test surface here exercising the public error helpers a
+// consumer would import from the same module.
+describe("isChatChunkTooLargeError", () => {
+  it("recognizes its own error class", () => {
+    const err = new ChatChunkTooLargeError(2_000_000, 1024 * 1024 - 1024, "x");
+    expect(isChatChunkTooLargeError(err)).toBe(true);
+  });
+
+  it("rejects unrelated errors", () => {
+    expect(isChatChunkTooLargeError(new Error("nope"))).toBe(false);
+    expect(isChatChunkTooLargeError("string")).toBe(false);
+    expect(isChatChunkTooLargeError(undefined)).toBe(false);
   });
 });

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -161,16 +161,12 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 controller.error(new Error("Stream aborted"));
                 return;
               }
-              const body = JSON.stringify({ data: chunk, id: nanoid(7) });
-              const size = utf8Encoder.encode(body).length;
-              if (size > RECORD_BODY_MAX_BYTES) {
-                const chunkType = extractChunkType(chunk);
-                controller.error(
-                  new ChatChunkTooLargeError(size, RECORD_BODY_MAX_BYTES, chunkType)
-                );
+              const encoded = encodeChunkOrError(chunk);
+              if (!encoded.ok) {
+                controller.error(encoded.error);
                 return;
               }
-              controller.enqueue(AppendRecord.string({ body }));
+              controller.enqueue(AppendRecord.string({ body: encoded.body }));
             },
           })
         )
@@ -258,3 +254,29 @@ function extractChunkType(chunk: unknown): string | undefined {
   if (typeof c.kind === "string") return c.kind;
   return undefined;
 }
+
+/**
+ * Encode a chunk as a JSON record body for S2, enforcing the per-record
+ * size cap. Exported so the size/discriminant logic can be unit-tested
+ * directly without spinning up an S2 client or mocking `@s2-dev/streamstore`.
+ *
+ * Returns `{ ok: true, body }` when the encoded chunk fits within
+ * `RECORD_BODY_MAX_BYTES`, or `{ ok: false, error }` carrying a
+ * `ChatChunkTooLargeError` annotated with the chunk's discriminant
+ * (`type` or `kind`, whichever is present) so the surfaced error is
+ * useful — "tool-output-available chunk too large" beats a bare
+ * "chunk too large" by a lot.
+ */
+export function encodeChunkOrError(
+  chunk: unknown
+): { ok: true; body: string } | { ok: false; error: ChatChunkTooLargeError } {
+  const body = JSON.stringify({ data: chunk, id: nanoid(7) });
+  const size = utf8Encoder.encode(body).length;
+  if (size > RECORD_BODY_MAX_BYTES) {
+    return {
+      ok: false,
+      error: new ChatChunkTooLargeError(size, RECORD_BODY_MAX_BYTES, extractChunkType(chunk)),
+    };
+  }
+  return { ok: true, body };
+}