test(webapp): full-stack chat.agent e2e with the real turn loop

Runs the genuine chat.agent run loop in-process against the testcontainer
stack (webapp + Postgres + Redis + s2-lite + MinIO): the agent's `.in`/`.out`
go through the real webapp Session streams and its snapshots through the real
object store, with the model injected as a deterministic MockLanguageModelV3.
Turns are driven by appending to `.in` over HTTP and read back through the SSE
proxy, so no real LLM or dev worker is involved. Covers a basic turn,
multi-turn continuation on one run, and hydrateMessages.

Adds two small test-only utilities to make this possible: a
`StandardSessionStreamManager` export and a `sessionStreamManager` option on
`runInMockTaskContext`, both from `@trigger.dev/core/v3/test`.

Author: ericallam

apps/webapp/test/helpers/agentHarness.ts

@@ -0,0 +1,92 @@
+import { apiClientManager, resourceCatalog } from "@trigger.dev/core/v3";
+import type { LocalsKey } from "@trigger.dev/core/v3";
+import type { LanguageModel } from "ai";
+import { runInMockTaskContext, StandardSessionStreamManager } from "@trigger.dev/core/v3/test";
+
+export type RunRealChatAgentOptions = {
+  agentId: string;
+  baseUrl: string;
+  addressingKey: string;
+  /**
+   * The environment secret key. The agent writes `.out` and reads `.in` as the
+   * backend (PRIVATE auth) — the `.out` channel rejects client session tokens.
+   */
+  secretKey: string;
+  model: LanguageModel;
+  modelLocal: LocalsKey<LanguageModel>;
+  runId?: string;
+};
+
+export type RunningAgent = {
+  done: Promise<void>;
+  close: () => Promise<void>;
+};
+
+/**
+ * Run the real `chat.agent` turn loop in-process, wired to a running webapp:
+ * `apiClientManager` + a real `StandardSessionStreamManager` point the agent's
+ * `.in`/`.out` at the webapp's Session streams (real S2 + SSE), the model is
+ * injected via locals (so it survives without serialization), and turns are
+ * driven by appending to `.in` over HTTP. Callers keep each message inside the
+ * idle window and `close()` promptly so the run-engine suspend path is never
+ * reached.
+ */
+export function runRealChatAgent(opts: RunRealChatAgentOptions): RunningAgent {
+  apiClientManager.setGlobalAPIClientConfiguration({
+    baseURL: opts.baseUrl,
+    accessToken: opts.secretKey,
+  });
+  const apiClient = apiClientManager.clientOrThrow();
+  const manager = new StandardSessionStreamManager(apiClient, opts.baseUrl);
+
+  const taskEntry = resourceCatalog.getTask(opts.agentId);
+  if (!taskEntry) {
+    throw new Error(`runRealChatAgent: agent "${opts.agentId}" is not registered`);
+  }
+  const runFn = taskEntry.fns.run as (
+    payload: unknown,
+    params: { ctx: unknown; signal: AbortSignal }
+  ) => Promise<unknown>;
+
+  const runSignal = new AbortController();
+  const runId = opts.runId ?? `run_${opts.addressingKey}`;
+
+  const done = runInMockTaskContext(
+    async (drivers) => {
+      drivers.locals.set(opts.modelLocal, opts.model);
+      await runFn(
+        { chatId: opts.addressingKey, trigger: "preload", metadata: {} },
+        { ctx: drivers.ctx, signal: runSignal.signal }
+      );
+    },
+    { ctx: { run: { id: runId } }, sessionStreamManager: manager }
+  ) as Promise<void>;
+
+  return {
+    done,
+    close: async () => {
+      try {
+        await fetch(
+          `${opts.baseUrl}/realtime/v1/sessions/${encodeURIComponent(opts.addressingKey)}/in/append`,
+          {
+            method: "POST",
+            headers: {
+              Authorization: `Bearer ${opts.secretKey}`,
+              "Content-Type": "application/json",
+              "X-Part-Id": "close",
+            },
+            body: JSON.stringify({
+              kind: "message",
+              payload: { chatId: opts.addressingKey, trigger: "close" },
+            }),
+          }
+        );
+      } catch {}
+      runSignal.abort();
+      await Promise.race([
+        done.catch(() => {}),
+        new Promise((resolve) => setTimeout(resolve, 10_000)),
+      ]);
+    },
+  };
+}

apps/webapp/test/helpers/testChatAgent.ts

@@ -0,0 +1,63 @@
+import { chat } from "@trigger.dev/sdk/ai";
+import { locals } from "@trigger.dev/core/v3";
+import { streamText, type LanguageModel, type UIMessage } from "ai";
+import { z } from "zod";
+
+/**
+ * The model is injected through locals (an in-process, by-reference value)
+ * rather than clientData, because a `MockLanguageModelV3` can't survive the
+ * JSON round-trip through the real `.in/append` route. The harness sets this
+ * before the run starts.
+ */
+export const testChatModelLocal = locals.create<LanguageModel>("e2e-test-chat.model");
+
+export type TestChatClientData = { hydrated?: UIMessage[] };
+
+function firstText(m: UIMessage): string {
+  const p = m.parts?.[0];
+  return p?.type === "text" ? p.text : "";
+}
+
+export const testChatAgent = chat
+  .withClientData({
+    schema: z.custom<TestChatClientData>((v) => v == null || typeof v === "object"),
+  })
+  .agent({
+    id: "e2e-test-chat",
+
+    onValidateMessages: async ({ messages }) => {
+      for (const m of messages) {
+        if (m.role === "user" && firstText(m).toLowerCase().includes("blocked-word")) {
+          throw new Error("Message blocked by content filter");
+        }
+      }
+      return messages;
+    },
+
+    hydrateMessages: async ({ clientData, incomingMessages }) => {
+      if (!clientData?.hydrated) return incomingMessages;
+      const merged = [...clientData.hydrated];
+      for (const m of incomingMessages) {
+        const idx = merged.findIndex((x) => x.id === m.id);
+        if (idx === -1) merged.push(m);
+        else merged[idx] = m;
+      }
+      return merged;
+    },
+
+    actionSchema: z.discriminatedUnion("type", [z.object({ type: z.literal("undo") })]),
+
+    onAction: async ({ action }) => {
+      if (action.type === "undo") {
+        chat.history.slice(0, -2);
+      }
+    },
+
+    run: async ({ messages, signal }) => {
+      const model = locals.get(testChatModelLocal);
+      if (!model) {
+        throw new Error("test model not injected via locals");
+      }
+      return streamText({ model, messages, abortSignal: signal });
+    },
+  });