fix(chat,sdk): attach the input router per run, not per chat

A worker process is reused across runs, and the executor tears the channel
subscription down at the end of each one. The router was cached per chat, so a
second run of the same chat in the same process skipped the attach and then had
no subscription feeding it: no messages arrived, and the conversation hung with
no error raised.

Caching it per run instead means every run attaches its own subscription, while a
nested `chat.createSession` inside one run still shares the router. Reported by
review, reproduced first as a failing test that drives two runs through one
process with the executor's teardown in between.

Author: ericallam

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

@@ -1871,17 +1871,18 @@ const CHAT_ROUTE_STOP = "stop";
 const CHAT_ROUTE_HANDOVER = "handover";
 
 /**
- * The `.in` router for the chat this worker is serving.
- *
- * One slot rather than a map: a worker process serves one chat, and the facades
- * have to reach the same router the boot attached without depending on a locals
- * scope being active. Tagged with its chat so a nested `chat.createSession` for
- * the same chat reuses the attached router while a different chat gets a fresh
- * one.
+ * The `.in` router for the run this worker is currently serving.
+ *
+ * One slot rather than a map, because the facades have to reach the same router
+ * the boot attached without depending on a locals scope being active. Tagged
+ * with the run as well as the chat: a warm process is reused across runs and the
+ * executor tears the channel subscription down at the end of each one, so
+ * reusing a router across runs would leave the new run with no input at all. A
+ * nested `chat.createSession` within the same run still shares it.
  * @internal
  */
 let currentChatInputRouter:
-  | { chatId: string; router: SessionChannelRouter; attached: boolean }
+  | { chatId: string; runId: string | undefined; router: SessionChannelRouter; attached: boolean }
   | undefined;
 
 /**
@@ -2021,13 +2022,18 @@ async function installChatInputRouter(
 
 function chatInputRouterEntry(chatId: string): {
   chatId: string;
+  runId: string | undefined;
   router: SessionChannelRouter;
   attached: boolean;
 } {
-  if (currentChatInputRouter?.chatId === chatId) return currentChatInputRouter;
+  const runId = taskContext.ctx?.run.id;
+  if (currentChatInputRouter?.chatId === chatId && currentChatInputRouter.runId === runId) {
+    return currentChatInputRouter;
+  }
 
   currentChatInputRouter = {
     chatId,
+    runId,
     router: new SessionChannelRouter(CHAT_INPUT_ROUTES, {
       onDrop: (record, reason) => {
         if (reason === "unroutable" || reason === "malformed") {

packages/trigger-sdk/test/chat-warm-process-reuse.test.ts

@@ -0,0 +1,60 @@
+import "../src/v3/test/index.js";
+
+import { resourceCatalog, sessionStreams } from "@trigger.dev/core/v3";
+import { runInMockTaskContext } from "@trigger.dev/core/v3/test";
+import { describe, expect, it } from "vitest";
+import { chat, type ChatTaskWirePayload } from "../src/v3/ai.js";
+
+function userPayload(chatId: string, id: string): ChatTaskWirePayload {
+  return {
+    chatId,
+    trigger: "submit-message",
+    message: { id, role: "user", parts: [{ type: "text", text: id }] },
+  };
+}
+
+/**
+ * A worker process is reused across runs. The end of every run tears down the
+ * channel subscription via `sessionStreams.clearHandlers()`, so a second run of
+ * the same chat in the same process has to attach a fresh one. Anything cached
+ * across runs that skips the attach leaves the new run with no input at all, and
+ * the conversation hangs with no error raised.
+ */
+describe("chat input across runs in one warm process", () => {
+  it("delivers messages to a second run of the same chat", async () => {
+    const chatId = "warm-reuse";
+    const seen: string[] = [];
+
+    const agent = chat.customAgent({
+      id: "chat-warm-process-reuse",
+      run: async () => {
+        const record = await chat.messages.next({ timeoutInSeconds: 2 });
+        const part = record?.payload.message?.parts?.[0];
+        if (part && part.type === "text") seen.push(part.text);
+      },
+    });
+    const run = resourceCatalog.getTask(agent.id)?.fns.run;
+    if (!run) throw new Error("custom agent was not registered");
+
+    for (const attempt of ["first", "second"]) {
+      await runInMockTaskContext(
+        async (drivers) => {
+          const runPromise = run(
+            { chatId, trigger: "submit-message" },
+            { ctx: drivers.ctx, signal: new AbortController().signal }
+          );
+          await drivers.sessions.in.send(
+            chatId,
+            { kind: "message", payload: userPayload(chatId, attempt) },
+            "in"
+          );
+          await runPromise;
+        },
+        { ctx: { run: { id: `run_${attempt}` } } }
+      );
+      sessionStreams.clearHandlers();
+    }
+
+    expect(seen).toEqual(["first", "second"]);
+  });
+});