fix(core,webapp): stop a failed session.out write from killing the process

fix(core,webapp): stop a failed session.out write from killing the
process

`StreamsWriterV2` and `SessionStreamInstance` stored the promise from
`initializeServerStream()` without a rejection handler. A writer created
detached and awaited only later turned an early append failure into a
floating rejection; without an `unhandledRejection` listener Node
escalated it and the webapp exited, and the same path could fail a run
whose append had exhausted its retries. The stored promise now carries a
handler while `wait()` still rejects, the webapp logs unhandled
rejections next to its `uncaughtException` handler, the chat server logs
a dropped session output write instead of swallowing it, and the v1
writer gets the same guard. Regression tests run against a real HTTP/2
fake stream server.

Mono-RevId: 4a4e2d30dfec23de72c7e2525dcd0589c6e5f29d

Author: kathiekiwi

.changeset/log-dropped-session-out-write.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+A failed write to a realtime or chat session stream no longer crashes the process running it, and a dropped chat session output write is now logged instead of swallowed.

apps/webapp/app/entry.server.tsx

@@ -290,6 +290,17 @@ function logError(error: unknown, request?: Request) {
   console.error(error);
 }
 
+// Without a listener Node escalates a rejection into the uncaughtException
+// handler below, which exits the process. Log it (Sentry via Logger.onError)
+// and keep serving. Wrapped in singleton() so Remix's dev-mode CJS reloads
+// don't stack duplicate listeners.
+singleton("UnhandledRejectionHandler", () => {
+  process.on("unhandledRejection", (reason) => {
+    logger.error("unhandledRejection", { error: reason });
+  });
+  return true;
+});
+
 process.on("uncaughtException", (error, origin) => {
   if (
     error instanceof Prisma.PrismaClientKnownRequestError ||

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

@@ -40,6 +40,9 @@ export class SessionStreamInstance<T> implements StreamsWriter {
 
   constructor(private options: SessionStreamInstanceOptions<T>) {
     this.streamPromise = this.initializeWriter();
+    // Same detached-writer guard as `StreamsWriterV2`: the error is still
+    // surfaced by `wait()` / `stream`.
+    this.streamPromise.catch(() => {});
   }
 
   private async initializeWriter(): Promise<StreamsWriterV2<T>> {

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

@@ -54,6 +54,9 @@ export class StreamsWriterV1<T> implements StreamsWriter {
     this.startBuffering();
 
     this.streamPromise = this.initializeServerStream();
+    // Same detached-writer guard as `StreamsWriterV2`: `wait()` still surfaces
+    // the error, but an unawaited failure never reaches Node as unhandled.
+    this.streamPromise.catch(() => {});
   }
 
   private generateClientId(): string {

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

@@ -1,7 +1,10 @@
+import http2 from "node:http2";
+import net from "node:net";
+import type { AddressInfo } from "node:net";
 import { describe, expect, it } from "vitest";
 
 import { ChatChunkTooLargeError, isChatChunkTooLargeError } from "../errors.js";
-import { encodeChunkOrError } from "./streamsWriterV2.js";
+import { StreamsWriterV2, encodeChunkOrError } from "./streamsWriterV2.js";
 
 // The size cap and discriminant extraction are the only S2-independent bits
 // of `StreamsWriterV2` that benefit from unit coverage. Both live in the
@@ -75,3 +78,113 @@ describe("isChatChunkTooLargeError", () => {
     expect(isChatChunkTooLargeError(undefined)).toBe(false);
   });
 });
+
+// A session `.out` stream that S2 reports as not found (the head-start drain
+// can append before the stream is visible). The writer is detached — nothing
+// awaits it until the handover flush — so a rejecting append used to escape
+// as an unhandled rejection and take the whole process down.
+describe("StreamsWriterV2 against a stream S2 does not have", () => {
+  it("keeps the append failure on `wait()` instead of emitting an unhandled rejection", async () => {
+    const server = http2.createServer((_req, res) => {
+      res.writeHead(404, { "content-type": "application/json" });
+      res.end(
+        JSON.stringify({ message: "stream sessions/chat_x/out not found", code: "not_found" })
+      );
+    });
+    const sessions = new Set<http2.ServerHttp2Session>();
+    server.on("session", (session) => {
+      sessions.add(session);
+      session.on("close", () => sessions.delete(session));
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", () => resolve()));
+    const { port } = server.address() as AddressInfo;
+
+    const rejections: unknown[] = [];
+    const onUnhandledRejection = (reason: unknown) => rejections.push(reason);
+    process.on("unhandledRejection", onUnhandledRejection);
+
+    try {
+      const writer = new StreamsWriterV2<string>({
+        basin: "trigger-local",
+        stream: "sessions/chat_x/out",
+        accessToken: "token",
+        endpoint: `http://127.0.0.1:${port}/v1`,
+        source: new ReadableStream<string>({
+          start(controller) {
+            controller.enqueue("hello");
+            controller.close();
+          },
+        }),
+        flushIntervalMs: 5,
+      });
+
+      // Long enough for the append to fail and for Node to flag an unhandled
+      // rejection, without anything having awaited the writer yet.
+      await new Promise((resolve) => setTimeout(resolve, 1_000));
+
+      expect(rejections).toEqual([]);
+      await expect(writer.wait()).rejects.toThrow(/not found/);
+    } finally {
+      process.off("unhandledRejection", onUnhandledRejection);
+      for (const session of sessions) session.destroy();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  }, 20_000);
+
+  // Same escape route when the SDK's own append retries run out. Reached via
+  // the retry loop's session-creation branch, not the ack-timeout branch.
+  it("keeps a retry-exhausted append on `wait()`", async () => {
+    // Holds the port and refuses every connection, so the retry budget is
+    // spent on connection failures with no window for another listener.
+    let connectionAttempts = 0;
+    const server = net.createServer((socket) => {
+      connectionAttempts++;
+      socket.destroy();
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", () => resolve()));
+    const { port } = server.address() as AddressInfo;
+
+    const rejections: unknown[] = [];
+    const onUnhandledRejection = (reason: unknown) => rejections.push(reason);
+    process.on("unhandledRejection", onUnhandledRejection);
+
+    try {
+      const writer = new StreamsWriterV2<string>({
+        basin: "trigger-local",
+        stream: "sessions/chat_x/out",
+        accessToken: "token",
+        endpoint: `http://127.0.0.1:${port}/v1`,
+        source: new ReadableStream<string>({
+          start(controller) {
+            controller.enqueue("hello");
+            controller.close();
+          },
+        }),
+        flushIntervalMs: 5,
+      });
+
+      // Side channel: wait for the budget (3 attempts) to be spent without
+      // touching the writer, then let the abort propagate.
+      const deadline = Date.now() + 5_000;
+      while (connectionAttempts < 3 && Date.now() < deadline) {
+        await new Promise((resolve) => setTimeout(resolve, 10));
+      }
+      expect(connectionAttempts).toBeGreaterThanOrEqual(3);
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      expect(rejections).toEqual([]);
+      // Rejects in a microtask if it already failed; the 0ms timer wins if the
+      // writer is somehow still pending, so a slow runner fails instead of
+      // passing vacuously.
+      await expect(
+        Promise.race([
+          writer.wait(),
+          new Promise((_, reject) => setTimeout(() => reject(new Error("still pending")), 0)),
+        ])
+      ).rejects.toThrow(/Max attempts \(3\) exhausted/);
+    } finally {
+      process.off("unhandledRejection", onUnhandledRejection);
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  }, 20_000);
+});

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

@@ -108,6 +108,10 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
     this.consumerStream = consumerStream;
 
     this.streamPromise = this.initializeServerStream();
+    // Detached writers (e.g. the head-start drain) call `wait()` long after a
+    // failed append rejects; without this the rejection is unhandled and Node
+    // takes the process down. `wait()` still surfaces the error.
+    this.streamPromise.catch(() => {});
   }
 
   private handleAbort(): void {

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

@@ -749,10 +749,17 @@ async function openHandoverSession(opts: {
     if (!sessionWriter) return;
     try {
       await sessionWriter.wait();
-    } catch {
-      // Drop write errors — the customer's response stream is the
-      // source of truth for what the user sees. Durability/resume
-      // best-effort.
+    } catch (error) {
+      // Dropped, not thrown: the customer's response stream is the source of
+      // truth for what the user sees. Logged so the loss is diagnosable.
+      const { message, code, status, origin } = (error ?? {}) as Record<string, unknown>;
+      console.warn("[chat.handover] session.out write failed", {
+        chatId,
+        message,
+        code,
+        status,
+        origin,
+      });
     }
   };