fix(core,cli): stop chat.agent uncaught exception on mid-stream abort

Bumps @s2-dev/streamstore to 0.22.10. The realtime stream writer batches
chunks with a linger timer; when a turn was aborted mid-stream while a
record was still buffered, the timer fired enqueue() on an already-closed
stream controller and threw from a setTimeout callback — surfacing as
TASK_RUN_UNCAUGHT_EXCEPTION ("Invalid state: Unable to enqueue"). The
upstream patch wraps the linger flush in a try/catch.

Author: ericallam

.changeset/s2-batch-transform-linger-fix.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"trigger.dev": patch
+---
+
+Bump `@s2-dev/streamstore` to `0.22.10` to fix a `TASK_RUN_UNCAUGHT_EXCEPTION` ("Invalid state: Unable to enqueue") when a `chat.agent` turn is aborted mid-stream.

apps/webapp/package.json

@@ -112,7 +112,7 @@
     "@remix-run/serve": "2.17.4",
     "@remix-run/server-runtime": "2.17.4",
     "@remix-run/v1-meta": "^0.1.3",
-    "@s2-dev/streamstore": "^0.22.5",
+    "@s2-dev/streamstore": "^0.22.10",
     "@sentry/remix": "9.46.0",
     "@slack/web-api": "7.9.1",
     "@socket.io/redis-adapter": "^8.3.0",

packages/cli-v3/package.json

@@ -94,7 +94,7 @@
     "@opentelemetry/resources": "2.0.1",
     "@opentelemetry/sdk-trace-node": "2.0.1",
     "@opentelemetry/semantic-conventions": "1.36.0",
-    "@s2-dev/streamstore": "^0.22.5",
+    "@s2-dev/streamstore": "^0.22.10",
     "@trigger.dev/build": "workspace:4.5.0-rc.2",
     "@trigger.dev/core": "workspace:4.5.0-rc.2",
     "@trigger.dev/schema-to-json": "workspace:4.5.0-rc.2",

packages/core/package.json

@@ -206,7 +206,7 @@
     "@opentelemetry/sdk-trace-base": "2.0.1",
     "@opentelemetry/sdk-trace-node": "2.0.1",
     "@opentelemetry/semantic-conventions": "1.36.0",
-    "@s2-dev/streamstore": "0.22.5",
+    "@s2-dev/streamstore": "0.22.10",
     "dequal": "^2.0.3",
     "eventsource": "^3.0.5",
     "eventsource-parser": "^3.0.0",

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

@@ -1,4 +1,5 @@
 import { describe, expect, it } from "vitest";
+import { AppendRecord, BatchTransform } from "@s2-dev/streamstore";
 
 import { ChatChunkTooLargeError, isChatChunkTooLargeError } from "../errors.js";
 import { encodeChunkOrError } from "./streamsWriterV2.js";
@@ -75,3 +76,65 @@ describe("isChatChunkTooLargeError", () => {
     expect(isChatChunkTooLargeError(undefined)).toBe(false);
   });
 });
+
+// Regression guard for the `@s2-dev/streamstore` linger-timer race that
+// surfaced as `TASK_RUN_UNCAUGHT_EXCEPTION` ("Invalid state: Unable to
+// enqueue") in `chat.agent`. `StreamsWriterV2` pipes records through a
+// `BatchTransform` into S2's `session.writable`. When a run aborts mid-turn
+// while a record is still buffered in the linger window, the writable is
+// aborted and the transform's readable controller errors — but the pending
+// linger `setTimeout` still fires and calls `controller.enqueue()` on the
+// now-dead controller, throwing from a timer callback (so it's uncaught).
+//
+// Fixed upstream in `@s2-dev/streamstore@0.22.10` by wrapping the linger
+// flush in a try/catch that discards the closed-controller `TypeError`. This
+// test exercises the *real* `BatchTransform` (no mock) and fails if the
+// dependency is ever downgraded below the fix.
+describe("BatchTransform linger-timer abort safety (s2 dependency contract)", () => {
+  it("does not throw an uncaught error when the controller dies before the linger fires", async () => {
+    const lingerDurationMillis = 50;
+    const captured: unknown[] = [];
+    const onUncaught = (err: unknown) => captured.push(err);
+
+    // Intercept uncaught errors from the linger timer for the duration of the
+    // test — the throw happens in a `setTimeout`, so it can't be caught with a
+    // surrounding try/catch.
+    const prevUncaught = process.listeners("uncaughtException");
+    const prevUnhandled = process.listeners("unhandledRejection");
+    process.removeAllListeners("uncaughtException");
+    process.removeAllListeners("unhandledRejection");
+    process.on("uncaughtException", onUncaught);
+    process.on("unhandledRejection", onUncaught);
+
+    try {
+      // Mirror the StreamsWriterV2 pipeline shape: source -> BatchTransform ->
+      // session.writable. The downstream never acks, so the buffered record
+      // stays in the linger window until we abort.
+      const batcher = new BatchTransform({ lingerDurationMillis });
+      const downstream = new WritableStream({
+        write() {
+          return new Promise(() => {});
+        },
+      });
+      batcher.readable.pipeTo(downstream).catch(() => {});
+
+      const writer = batcher.writable.getWriter();
+      // Buffer a record — this arms the linger setTimeout.
+      await writer.write(AppendRecord.string({ body: "hello" }));
+      // Abort the downstream before the linger fires (== run suspend/abort ->
+      // session.writable.abort()), which errors the transform's readable side.
+      await downstream.abort?.("aborted").catch(() => {});
+      writer.abort("aborted").catch(() => {});
+      // Wait past the linger window so the pending timer fires on the dead
+      // controller.
+      await new Promise((r) => setTimeout(r, lingerDurationMillis + 150));
+    } finally {
+      process.removeListener("uncaughtException", onUncaught);
+      process.removeListener("unhandledRejection", onUncaught);
+      prevUncaught.forEach((l) => process.on("uncaughtException", l));
+      prevUnhandled.forEach((l) => process.on("unhandledRejection", l));
+    }
+
+    expect(captured).toEqual([]);
+  });
+});