fix(sdk): re-suspend chat agents after unmatched wakes

Chat agents now return to a durable wait when session activity does not
deliver a matching message. Repeated wakes preserve the original turn
timeout instead of leaving the run active until its maximum duration.

Mono-RevId: f9bb2238bbd2ea663eaac75ce3d92106a83c1104

Author: carderne

.changeset/chat-route-spurious-wakes.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Chat agents now return to a durable wait when a session wake does not deliver a matching message. This prevents resumed runs from staying active until their maximum duration and preserves the configured turn timeout across repeated wakes.

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

@@ -66,6 +66,7 @@ import type {
 // ESM-only `ai@7` (see ../imports/ai-runtime.ts).
 import { type Attributes, trace } from "@opentelemetry/api";
 import { traceSessionIdle } from "./sessionTracing.js";
+import { waitForChatRouteAfterIdle } from "./chatRouteWait.js";
 import {
   tool as aiTool,
   convertToModelMessages,
@@ -1780,31 +1781,21 @@ async function waitOnChatRoute<T>(
       if (options.onSuspend) await options.onSuspend();
 
       span.setAttribute("wait.resolved", "suspended");
-      while (true) {
-        /**
-         * The floor doubles as the wake cursor: the server completes the
-         * waitpoint immediately if anything sits after this sequence, so a
-         * floor that has advanced past an unread record parks a waitpoint
-         * nothing will complete. Recorded on the span so a run that never woke
-         * can be diagnosed from its trace alone.
-         */
-        const wakeFrom = router.resumeFloor();
-        span.setAttribute("wait.lastSeqNum", wakeFrom ?? -1);
-        const wake = await session.in.awaitWake({
-          timeout: options.timeout,
-          lastSeqNum: wakeFrom,
-        });
-        if (!wake.ok) {
-          span.recordException(wake.error);
-          return { ok: false as const, error: wake.error };
-        }
-
-        const record = await router.next(route);
-        if (!record) continue;
+      const result = await waitForChatRouteAfterIdle(router, route, {
+        timeout: options.timeout,
+        wake: async (timeout, lastSeqNum) => {
+          span.setAttribute("wait.lastSeqNum", lastSeqNum ?? -1);
+          return session.in.awaitWake({ timeout, lastSeqNum });
+        },
+      });
 
-        if (options.onResume) await options.onResume();
-        return { ok: true as const, output: record.data as T, record };
+      if (!result.ok) {
+        span.recordException(result.error);
+        return result;
       }
+
+      if (options.onResume) await options.onResume();
+      return { ok: true as const, output: result.record.data as T, record: result.record };
     },
     {
       attributes: {

packages/trigger-sdk/src/v3/chatRouteWait.test.ts

@@ -0,0 +1,101 @@
+import { SessionChannelRouter, WaitpointTimeoutError } from "@trigger.dev/core/v3";
+import { describe, expect, it } from "vitest";
+import { waitForChatRouteAfterIdle } from "./chatRouteWait.js";
+
+function inputRouter() {
+  return new SessionChannelRouter({
+    kindOf: (data) => (data as { kind?: string }).kind,
+    routes: [
+      { name: "messages", delivery: "queue", replayable: true, kinds: ["message"] },
+      { name: "control", delivery: "queue", replayable: false, kinds: ["control"] },
+    ],
+  });
+}
+
+describe("chat route waits after suspension", () => {
+  it("re-suspends after a wake that has no record for the requested route", async () => {
+    const router = inputRouter();
+    const timeouts: Array<string | undefined> = [];
+    let now = 0;
+
+    const result = await waitForChatRouteAfterIdle(router, "messages", {
+      timeout: "10s",
+      postWakeTimeoutMs: 0,
+      now: () => now,
+      wake: async (timeout) => {
+        timeouts.push(timeout);
+        if (timeouts.length === 1) {
+          router.ingest({ id: "control-1", seqNum: 1, data: { kind: "control" } });
+          now = 4_000;
+          return { ok: true, waitpointId: "waitpoint-1" };
+        }
+        return { ok: false, error: new WaitpointTimeoutError("Timed out") };
+      },
+    });
+
+    expect(result.ok).toBe(false);
+    expect(timeouts).toEqual(["10s", "6s"]);
+  });
+
+  it("does not re-wake on an unrelated queued message while waiting for another route", async () => {
+    const router = inputRouter();
+    const cursors: Array<number | undefined> = [];
+
+    const result = await waitForChatRouteAfterIdle(router, "control", {
+      postWakeTimeoutMs: 0,
+      wake: async (_timeout, lastSeqNum) => {
+        cursors.push(lastSeqNum);
+        if (cursors.length === 1) {
+          router.ingest({ id: "message-1", seqNum: 1, data: { kind: "message" } });
+          return { ok: true, waitpointId: "waitpoint-1" };
+        }
+        if (lastSeqNum === undefined || lastSeqNum < 1) {
+          return { ok: true, waitpointId: "waitpoint-repeated" };
+        }
+        return { ok: false, error: new WaitpointTimeoutError("Timed out") };
+      },
+    });
+
+    expect(result.ok).toBe(false);
+    expect(cursors).toEqual([undefined, 1]);
+    expect(router.resumeFloor()).toBe(0);
+    expect(router.hasPending("messages")).toBe(true);
+  });
+
+  it("expires an absolute timeout after an unmatched wake", async () => {
+    const router = inputRouter();
+    const deadline = Date.parse("2026-09-28T10:15:00Z");
+    let now = deadline - 1_000;
+    const timeouts: Array<string | undefined> = [];
+
+    const result = await waitForChatRouteAfterIdle(router, "messages", {
+      timeout: "2026-09-28T10:15:00Z",
+      postWakeTimeoutMs: 0,
+      now: () => now,
+      wake: async (timeout) => {
+        timeouts.push(timeout);
+        now = deadline + 1;
+        return { ok: true, waitpointId: "waitpoint-1" };
+      },
+    });
+
+    expect(result.ok).toBe(false);
+    expect(timeouts).toEqual(["1s"]);
+  });
+
+  it("returns a matching record delivered after the channel wakes", async () => {
+    const router = inputRouter();
+    const message = { id: "message-1", seqNum: 1, data: { kind: "message", text: "hello" } };
+
+    const result = await waitForChatRouteAfterIdle(router, "messages", {
+      timeout: "10s",
+      postWakeTimeoutMs: 0,
+      wake: async () => {
+        router.ingest(message);
+        return { ok: true, waitpointId: "waitpoint-1" };
+      },
+    });
+
+    expect(result).toEqual({ ok: true, record: message });
+  });
+});

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

@@ -0,0 +1,64 @@
+import {
+  type SessionChannelRouter,
+  type SessionStreamRecord,
+  WaitpointTimeoutError,
+} from "@trigger.dev/core/v3";
+import { parseNaturalLanguageDurationInMs } from "@trigger.dev/core/v3/isomorphic";
+
+const POST_WAKE_TIMEOUT_MS = 5_000;
+
+type ChatRouteWakeResult = { ok: true; waitpointId: string } | { ok: false; error: Error };
+
+export async function waitForChatRouteAfterIdle(
+  router: SessionChannelRouter,
+  route: string,
+  options: {
+    timeout?: string;
+    postWakeTimeoutMs?: number;
+    now?: () => number;
+    wake: (
+      timeout: string | undefined,
+      lastSeqNum: number | undefined
+    ) => Promise<ChatRouteWakeResult>;
+  }
+): Promise<{ ok: true; record: SessionStreamRecord } | { ok: false; error: Error }> {
+  const now = options.now ?? Date.now;
+  const timeoutMs = options.timeout ? parseNaturalLanguageDurationInMs(options.timeout) : undefined;
+  const parsedDeadline =
+    timeoutMs !== undefined
+      ? now() + timeoutMs
+      : options.timeout === undefined
+        ? undefined
+        : Date.parse(options.timeout);
+  const deadline =
+    parsedDeadline !== undefined && Number.isFinite(parsedDeadline) ? parsedDeadline : undefined;
+  const postWakeTimeoutMs = options.postWakeTimeoutMs ?? POST_WAKE_TIMEOUT_MS;
+
+  while (true) {
+    const remainingMs = deadline === undefined ? undefined : deadline - now();
+    if (remainingMs !== undefined && remainingMs <= 0) {
+      return { ok: false, error: new WaitpointTimeoutError("Timed out") };
+    }
+
+    const wakeTimeout =
+      remainingMs === undefined
+        ? options.timeout
+        : `${Math.max(1, Math.ceil(remainingMs / 1000))}s`;
+    // A queued record holds the replay floor back, but should not wake this
+    // waitpoint again after we have already seen it. Snapshot before checking
+    // the route so a record arriving during registration still wakes us.
+    const wakeFrom = router.appliedThrough();
+    const buffered = await router.next(route, { timeoutMs: 0 });
+    if (buffered) return { ok: true, record: buffered };
+
+    const wake = await options.wake(wakeTimeout, wakeFrom);
+    if (!wake.ok) return wake;
+
+    const deliveryBudget =
+      deadline === undefined
+        ? postWakeTimeoutMs
+        : Math.max(0, Math.min(postWakeTimeoutMs, deadline - now()));
+    const record = await router.next(route, { timeoutMs: deliveryBudget });
+    if (record) return { ok: true, record };
+  }
+}