fix(sdk,chat): route pipeChat through session.out + chat-agent smoke test

pipeChat (the internal that auto-pipes a chat.agent's returned
streamText result to the chat output) was still calling
streams.pipe(CHAT_STREAM_KEY, stream) — a run-scoped run-streams
path. After the session migration, the module-level facades
(chatStream, messagesInput, stopInput) routed correctly, but
pipeChat bypassed the facade and went straight to the old
run-scoped pipe. Result: the turn-complete control chunk reached
the session.out subscriber (written via chatStream.writer in
writeTurnCompleteChunk) but every streamed UIMessageChunk from the
LLM's turn was written to the dead run-scoped stream and never
surfaced on session.out.

Swap the pipe target to chatStream.pipe (the session-routed
facade). The target / streamKey options still type-check for
API parity but are no longer meaningful — sessions are the
address, and sub-agents that need to write to a parent chat open
the parent's Session explicitly. Smoke now catches all 14 chunks
(start / start-step / text-start / 7x text-delta / text-end /
finish-step / finish / trigger:turn-complete) with ids 0 through
13 from session.out, match: true.

Also adds references/hello-world/src/trigger/chatAgentSmoke.ts —
end-to-end validation:
- sessions.create with externalId = chatId
- trigger test-agent with {chatId, sessionId, messages, …}
- handle.out.read(...) SSE subscribe, capture chunks by id+type
- sessions.close on completion

Triggered from the dashboard or MCP as chat-agent-smoke. Requires
OPENAI_API_KEY in the dev env (the test-agent uses
openai:gpt-4o-mini).

Author: ericallam

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

@@ -2687,7 +2687,6 @@ async function pipeChat(
   options?: PipeChatOptions
 ): Promise<void> {
   locals.set(chatPipeCountKey, (locals.get(chatPipeCountKey) ?? 0) + 1);
-  const streamKey = options?.streamKey ?? CHAT_STREAM_KEY;
 
   let stream: AsyncIterable<unknown> | ReadableStream<unknown>;
 
@@ -2702,18 +2701,27 @@ async function pipeChat(
     );
   }
 
-  const pipeOptions: PipeStreamOptions = {};
+  const pipeOptions: SessionPipeStreamOptions = {};
   if (options?.signal) {
     pipeOptions.signal = options.signal;
   }
-  if (options?.target) {
-    pipeOptions.target = options.target;
-  }
   if (options?.spanName) {
     pipeOptions.spanName = options.spanName;
   }
-
-  const { waitUntilComplete } = streams.pipe(streamKey, stream, pipeOptions);
+  // `options.target` / `options.streamKey` are accepted for API parity
+  // with the pre-migration run-scoped pipe but no longer have meaning —
+  // sessions are the address (single stream per session, no sub-run
+  // targeting). Sub-agents that need to write into a parent's chat now
+  // open that session explicitly via `sessions.open(parentSessionId).out.pipe`.
+
+  // The generic is typed for `UIMessageChunk`, but `pipeChat` also
+  // accepts opaque UIMessageStreamable / raw iterables whose element
+  // type we don't know at compile time. Cast — runtime behaviour is
+  // identical (bytes go to session.out either way).
+  const { waitUntilComplete } = chatStream.pipe(
+    stream as ReadableStream<UIMessageChunk> | AsyncIterable<UIMessageChunk>,
+    pipeOptions
+  );
   await waitUntilComplete();
 }
 

references/hello-world/src/trigger/chatAgentSmoke.ts

@@ -0,0 +1,111 @@
+import { logger, sessions, task, tasks } from "@trigger.dev/sdk";
+
+/**
+ * End-to-end smoke test for the chat.agent -> Sessions migration.
+ *
+ * Flow:
+ *   1. Create a Session with a deterministic externalId so the
+ *      `test-agent` run can `sessions.open(...)` it on startup.
+ *   2. Trigger `test-agent` with `{chatId, sessionId, messages, trigger,
+ *      metadata}` — mirrors what TriggerChatTransport would send for a
+ *      first message, minus the browser-triggered access token layer.
+ *   3. `session.out.read({...})` — consume the agent's UIMessageChunks
+ *      as they stream out. Bail after the first text-delta (good
+ *      enough to prove output flow + SSE subscription).
+ *   4. `sessions.close(...)` — tidy up.
+ *
+ * Trigger from the dashboard or MCP:
+ *
+ *   mcp__trigger__trigger_task(taskId: "chat-agent-smoke", payload: {})
+ *
+ * Expects OPENAI_API_KEY set in the env (the test-agent uses
+ * `openai:gpt-4o-mini`). If the key is missing the smoke reports an
+ * error payload without crashing.
+ */
+export const chatAgentSmoke = task({
+  id: "chat-agent-smoke",
+  run: async () => {
+    const stamp = Date.now();
+    const chatId = `chat-agent-smoke-${stamp}`;
+
+    logger.info("creating chat.agent backing session", { externalId: chatId });
+    const session = await sessions.create({
+      type: "chat.agent",
+      externalId: chatId,
+      tags: ["chat-agent-smoke"],
+    });
+
+    logger.info("triggering test-agent run", {
+      chatId,
+      sessionId: session.id,
+    });
+    await tasks.trigger("test-agent", {
+      chatId,
+      sessionId: session.id,
+      trigger: "submit-message",
+      messages: [
+        {
+          id: `m-${stamp}`,
+          role: "user",
+          parts: [{ type: "text", text: "Say hello in five words." }],
+        },
+      ],
+      metadata: { userId: "smoke", model: "openai:gpt-4o-mini" },
+    });
+
+    logger.info("subscribing to session.out, waiting for first chunks");
+    const handle = sessions.open(session.id);
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(), 60_000);
+
+    const received: Array<{ type?: string; id?: string }> = [];
+    let firstTextDelta: string | undefined;
+    let turnCompleteSeen = false;
+
+    try {
+      const stream = await handle.out.read<Record<string, unknown>>({
+        signal: controller.signal,
+        timeoutInSeconds: 30,
+        // Start from seq 0 so we don't race the agent's early writes.
+        lastEventId: "-1",
+        onPart: (part) => {
+          // Record the event id alongside the chunk so we can see the
+          // full sequence that came down the wire.
+          received.push({
+            id: part.id,
+            type: (part.chunk as { type?: string } | null)?.type,
+          });
+        },
+      });
+
+      for await (const chunk of stream) {
+        if (chunk.type === "text-delta" && typeof chunk.delta === "string") {
+          firstTextDelta ??= chunk.delta;
+        }
+        if (chunk.type === "trigger:turn-complete") {
+          turnCompleteSeen = true;
+          break;
+        }
+        if (received.length > 500) break;
+      }
+    } catch (err) {
+      if ((err as Error).name !== "AbortError") throw err;
+    } finally {
+      clearTimeout(timeout);
+    }
+
+    await sessions.close(session.id, { reason: "chat-agent-smoke-done" });
+
+    return {
+      ok: received.length > 0,
+      chatId,
+      sessionId: session.id,
+      chunkCount: received.length,
+      firstTextDelta,
+      turnCompleteSeen,
+      types: [...new Set(received.map((c) => c.type ?? "<unknown>"))],
+      firstFiveIds: received.slice(0, 5).map((c) => `${c.id}:${c.type ?? "<u>"}`),
+      lastFiveIds: received.slice(-5).map((c) => `${c.id}:${c.type ?? "<u>"}`),
+    };
+  },
+});