test(sdk,core): session-aware mockChatAgent harness + chat.test.ts mocks

Unblocks the unit tests after the chat.agent -> Sessions migration
(phases B/C/D). Before: 43 passed / 43 failed (35 in chat.test.ts + 10
mockChatAgent + 2 skillsRuntime). After: 86 passed / 0 failed.

Core (@trigger.dev/core/v3/test)

- TestSessionStreamManager: in-memory SessionStreamManager keyed on
  (sessionId, io) mirroring TestInputStreamManager. Dispatch rules
  match production with one test-only tweak — when a record arrives
  and only handlers are registered (no .once waiter), it's buffered
  for the next once() instead of discarded. Production doesn't need
  this because the SSE tail naturally serializes records after the
  agent's turn-loop has re-registered a waiter; tests send
  synchronously right after turn-complete, so without the buffer
  the next waitWithIdleTimeout loses the message.
- runInMockTaskContext installs the manager via
  sessionStreams.setGlobalManager, exposes drivers.sessions.in.send
  / .close, and tears down on exit.

SDK (@trigger.dev/sdk/v3/test)

- __setSessionOpenImplForTests hook in sessions.ts lets the harness
  override sessions.open(id) with an in-memory SessionHandle.
  SessionHandle constructor now accepts { in?, out? } overrides.
- TestSessionOutputChannel extends SessionOutputChannel and
  intercepts pipe / writer / append into a shared TestSessionOutState
  (chunks + listener registry). Never constructs SessionStreamInstance
  so it avoids initializeSessionStream / StreamsWriterV2 entirely.
- mockChatAgent rewritten: drops CHAT_MESSAGES_STREAM_ID /
  CHAT_STOP_STREAM_ID / the "chat" output stream key. sendMessage /
  sendRegenerate / sendAction push ChatInputChunk { kind: "message",
  payload } via drivers.sessions.in.send. sendStop pushes
  { kind: "stop" }. Turn-complete detection moves from
  drivers.outputs.onWrite to a TestSessionOutputChannel listener.

chat.test.ts

- New URL-predicate helpers at the top (isSessionCreateUrl,
  isTriggerTaskUrl, isSessionOutSubscribeUrl,
  isSessionStreamAppendUrl) + defaultSessionCreateResponse /
  defaultAppendResponse so every global.fetch mock speaks the
  same vocabulary.
- Bulk-updated all 25 mock blocks: added session-create handler
  (transport's accessToken path now lazily upserts via POST
  /api/v1/sessions before trigger), swapped /realtime/v1/streams/
  for /realtime/v1/sessions/ URL matchers, and replaced
  (streams/ && /input/) append-URL matchers with
  isSessionStreamAppendUrl.
- Three tests updated for new semantics: onSessionChange fires
  twice on first message (ensureSession -> sessionId only, then
  triggerNewRun -> adds runId + isStreaming). Async-token call
  count goes 1 -> 2 on first message because ensureSession and
  trigger both resolve the token with purpose: "trigger".
- "minimal wire payloads" test's body parsing updated — session.in
  append body is a raw JSON.stringify({ kind, payload }) string,
  not a { data } wrapper.
- Replaced the vestigial "custom streamKey URL" test with a
  "subscribes to the backing Session's .out" assertion. streamKey
  option is a no-op under sessions; removal can land in a follow-up.
- One test (stream closes without control chunk) legitimately
  needs 9s for SSE-close fallback — bumped its timeout to 15s.

Author: ericallam

packages/core/src/v3/test/index.ts

@@ -6,3 +6,4 @@ export {
 export { TestInputStreamManager } from "./test-input-stream-manager.js";
 export { TestRealtimeStreamsManager } from "./test-realtime-streams-manager.js";
 export { TestRunMetadataManager } from "./test-run-metadata-manager.js";
+export { TestSessionStreamManager } from "./test-session-stream-manager.js";

packages/core/src/v3/test/mock-task-context.ts

@@ -1,5 +1,6 @@
 import { inputStreams } from "../input-streams-api.js";
 import { realtimeStreams } from "../realtime-streams-api.js";
+import { sessionStreams } from "../session-streams-api.js";
 import { localsAPI } from "../locals-api.js";
 import { runMetadata } from "../run-metadata-api.js";
 import { taskContext } from "../task-context-api.js";
@@ -11,9 +12,11 @@ import { NoopRuntimeManager } from "../runtime/noopRuntimeManager.js";
 import { unregisterGlobal } from "../utils/globals.js";
 import type { ServerBackgroundWorker, TaskRunContext } from "../schemas/index.js";
 import type { LocalsKey } from "../locals/types.js";
+import type { SessionChannelIO } from "../sessionStreams/types.js";
 import { TestInputStreamManager } from "./test-input-stream-manager.js";
 import { TestRealtimeStreamsManager } from "./test-realtime-streams-manager.js";
 import { TestRunMetadataManager } from "./test-run-metadata-manager.js";
+import { TestSessionStreamManager } from "./test-session-stream-manager.js";
 
 /**
  * Shallow-partial overrides applied on top of the default mock
@@ -83,6 +86,23 @@ export type MockTaskContextDrivers = {
      */
     set<T>(key: LocalsKey<T>, value: T): void;
   };
+  /**
+   * Session-scoped channel drivers. The `.in` side is backed by a
+   * {@link TestSessionStreamManager} installed as the `sessionStreams`
+   * global — so the task's `session.in.on/once/peek/waitWithIdleTimeout`
+   * calls receive records sent through this driver.
+   */
+  sessions: {
+    in: {
+      /**
+       * Send a record onto `session.in` for the given session. Resolves
+       * pending `once()` waiters and fires all `on()` handlers.
+       */
+      send(sessionId: string, data: unknown, io?: SessionChannelIO): Promise<void>;
+      /** Close pending `once()` waiters with a timeout error. */
+      close(sessionId: string, io?: SessionChannelIO): void;
+    };
+  };
   /** The mock `TaskRunContext` assembled from defaults + user overrides. */
   ctx: TaskRunContext;
 };
@@ -198,6 +218,7 @@ export async function runInMockTaskContext<T>(
   const metadataManager = new TestRunMetadataManager();
   const inputManager = new TestInputStreamManager();
   const outputManager = new TestRealtimeStreamsManager();
+  const sessionStreamManager = new TestSessionStreamManager();
 
   // Unregister any previously-installed managers so `setGlobal*` wins —
   // `registerGlobal` returns false silently if an entry already exists.
@@ -207,6 +228,7 @@ export async function runInMockTaskContext<T>(
   unregisterGlobal("run-metadata");
   unregisterGlobal("input-streams");
   unregisterGlobal("realtime-streams");
+  unregisterGlobal("session-streams");
   unregisterGlobal("task-context");
 
   localsAPI.setGlobalLocalsManager(localsManager);
@@ -215,6 +237,7 @@ export async function runInMockTaskContext<T>(
   runMetadata.setGlobalManager(metadataManager);
   inputStreams.setGlobalManager(inputManager);
   realtimeStreams.setGlobalManager(outputManager);
+  sessionStreams.setGlobalManager(sessionStreamManager);
   taskContext.setGlobalTaskContext({
     ctx,
     worker,
@@ -237,6 +260,14 @@ export async function runInMockTaskContext<T>(
       set: <TValue>(key: LocalsKey<TValue>, value: TValue) =>
         localsManager.setLocal(key, value),
     },
+    sessions: {
+      in: {
+        send: (sessionId, data, io = "in") =>
+          sessionStreamManager.__sendFromTest(sessionId, io, data),
+        close: (sessionId, io = "in") =>
+          sessionStreamManager.__closeFromTest(sessionId, io),
+      },
+    },
     ctx,
   };
 
@@ -251,10 +282,12 @@ export async function runInMockTaskContext<T>(
     unregisterGlobal("task-context");
     unregisterGlobal("input-streams");
     unregisterGlobal("realtime-streams");
+    unregisterGlobal("session-streams");
     unregisterGlobal("run-metadata");
     localsManager.reset();
     inputManager.reset();
     outputManager.reset();
+    sessionStreamManager.reset();
     metadataManager.reset();
   }
 }

packages/core/src/v3/test/test-session-stream-manager.ts

@@ -0,0 +1,271 @@
+import {
+  InputStreamOncePromise,
+  InputStreamOnceResult,
+  InputStreamTimeoutError,
+} from "../inputStreams/types.js";
+import type { InputStreamOnceOptions } from "../realtimeStreams/types.js";
+import type {
+  SessionChannelIO,
+  SessionStreamManager,
+} from "../sessionStreams/types.js";
+
+type OnceWaiter = {
+  resolve: (value: InputStreamOnceResult<unknown>) => void;
+  timer?: ReturnType<typeof setTimeout>;
+  signal?: AbortSignal;
+  abortHandler?: () => void;
+};
+
+type Handler = (data: unknown) => void | Promise<void>;
+
+function keyFor(sessionId: string, io: SessionChannelIO): string {
+  return `${sessionId}:${io}`;
+}
+
+/**
+ * In-memory implementation of `SessionStreamManager` for unit tests. Same
+ * shape as {@link TestInputStreamManager} but keyed on `(sessionId, io)`.
+ *
+ * Tests push data via `__sendFromTest(sessionId, io, data)` — any pending
+ * `once()` waiters resolve immediately, and all `on()` handlers fire (awaited
+ * if they return a promise). Records that arrive before a listener is
+ * registered are buffered so the first `once()` picks them up.
+ */
+export class TestSessionStreamManager implements SessionStreamManager {
+  private handlers = new Map<string, Set<Handler>>();
+  private onceWaiters = new Map<string, OnceWaiter[]>();
+  private buffer = new Map<string, unknown[]>();
+  private seqNums = new Map<string, number>();
+
+  on(
+    sessionId: string,
+    io: SessionChannelIO,
+    handler: Handler
+  ): { off: () => void } {
+    const key = keyFor(sessionId, io);
+
+    let set = this.handlers.get(key);
+    if (!set) {
+      set = new Set();
+      this.handlers.set(key, set);
+    }
+    set.add(handler);
+
+    const buffered = this.buffer.get(key);
+    if (buffered && buffered.length > 0) {
+      for (const data of buffered) {
+        this.invoke(handler, data);
+      }
+      this.buffer.delete(key);
+    }
+
+    return {
+      off: () => {
+        this.handlers.get(key)?.delete(handler);
+      },
+    };
+  }
+
+  once(
+    sessionId: string,
+    io: SessionChannelIO,
+    options?: InputStreamOnceOptions
+  ): InputStreamOncePromise<unknown> {
+    const key = keyFor(sessionId, io);
+
+    return new InputStreamOncePromise<unknown>((resolve) => {
+      if (options?.signal?.aborted) {
+        resolve({
+          ok: false,
+          error: new InputStreamTimeoutError(key, options.timeoutMs ?? 0),
+        });
+        return;
+      }
+
+      const buffered = this.buffer.get(key);
+      if (buffered && buffered.length > 0) {
+        const next = buffered.shift();
+        if (buffered.length === 0) this.buffer.delete(key);
+        resolve({ ok: true, output: next });
+        return;
+      }
+
+      const waiter: OnceWaiter = { resolve, signal: options?.signal };
+
+      if (options?.timeoutMs !== undefined) {
+        waiter.timer = setTimeout(() => {
+          this.removeWaiter(key, waiter);
+          resolve({
+            ok: false,
+            error: new InputStreamTimeoutError(key, options.timeoutMs!),
+          });
+        }, options.timeoutMs);
+      }
+
+      if (options?.signal) {
+        const abortHandler = () => {
+          this.removeWaiter(key, waiter);
+          if (waiter.timer) clearTimeout(waiter.timer);
+          resolve({
+            ok: false,
+            error: new InputStreamTimeoutError(key, options.timeoutMs ?? 0),
+          });
+        };
+        waiter.abortHandler = abortHandler;
+        options.signal.addEventListener("abort", abortHandler, { once: true });
+      }
+
+      let waiters = this.onceWaiters.get(key);
+      if (!waiters) {
+        waiters = [];
+        this.onceWaiters.set(key, waiters);
+      }
+      waiters.push(waiter);
+    });
+  }
+
+  peek(sessionId: string, io: SessionChannelIO): unknown | undefined {
+    const buffered = this.buffer.get(keyFor(sessionId, io));
+    if (buffered && buffered.length > 0) return buffered[0];
+    return undefined;
+  }
+
+  lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
+    return this.seqNums.get(keyFor(sessionId, io));
+  }
+
+  setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
+    this.seqNums.set(keyFor(sessionId, io), seqNum);
+  }
+
+  shiftBuffer(sessionId: string, io: SessionChannelIO): boolean {
+    const key = keyFor(sessionId, io);
+    const buffered = this.buffer.get(key);
+    if (buffered && buffered.length > 0) {
+      buffered.shift();
+      if (buffered.length === 0) this.buffer.delete(key);
+      return true;
+    }
+    return false;
+  }
+
+  disconnectStream(_sessionId: string, _io: SessionChannelIO): void {
+    // no-op — no real SSE tail in tests
+  }
+
+  clearHandlers(): void {
+    this.handlers.clear();
+  }
+
+  reset(): void {
+    for (const waiters of this.onceWaiters.values()) {
+      for (const w of waiters) {
+        if (w.timer) clearTimeout(w.timer);
+        if (w.signal && w.abortHandler) {
+          w.signal.removeEventListener("abort", w.abortHandler);
+        }
+      }
+    }
+    this.onceWaiters.clear();
+    this.handlers.clear();
+    this.buffer.clear();
+    this.seqNums.clear();
+  }
+
+  disconnect(): void {
+    this.reset();
+  }
+
+  // ── Test driver API (not part of SessionStreamManager interface) ──────
+
+  /**
+   * Push a record onto the given channel.
+   *
+   * Dispatch rules — similar to the production manager, but with a tweak
+   * that makes unit tests deterministic:
+   *
+   * 1. **Handlers always observe** (like production). A session-level `.on`
+   *    is a filter-observer — it fires every time a record arrives,
+   *    regardless of whether a `.once` waiter is also active.
+   * 2. **First waiter consumes** the record if present (like production).
+   * 3. **If no waiter, the record is buffered for the next `.once` call.**
+   *    Production discards records that only match handlers — but in
+   *    production the SSE tail introduces enough latency that the next
+   *    `.once` is usually registered before the next record arrives. Tests
+   *    send synchronously right after `turn-complete`, so without this
+   *    buffer the next `waitWithIdleTimeout` would race and lose the
+   *    message. The buffer is the only deviation from production semantics.
+   */
+  async __sendFromTest(
+    sessionId: string,
+    io: SessionChannelIO,
+    data: unknown
+  ): Promise<void> {
+    const key = keyFor(sessionId, io);
+
+    const handlers = this.handlers.get(key);
+    if (handlers && handlers.size > 0) {
+      await Promise.all(
+        Array.from(handlers).map((h) => Promise.resolve().then(() => h(data)))
+      );
+    }
+
+    const waiters = this.onceWaiters.get(key);
+    if (waiters && waiters.length > 0) {
+      const w = waiters.shift()!;
+      if (waiters.length === 0) this.onceWaiters.delete(key);
+      if (w.timer) clearTimeout(w.timer);
+      if (w.signal && w.abortHandler) {
+        w.signal.removeEventListener("abort", w.abortHandler);
+      }
+      w.resolve({ ok: true, output: data });
+      return;
+    }
+
+    let buffered = this.buffer.get(key);
+    if (!buffered) {
+      buffered = [];
+      this.buffer.set(key, buffered);
+    }
+    buffered.push(data);
+  }
+
+  /**
+   * Immediately resolve every pending `once()` waiter for the given channel
+   * with a timeout error. Simulates a closed stream (e.g. session closed).
+   */
+  __closeFromTest(sessionId: string, io: SessionChannelIO): void {
+    const key = keyFor(sessionId, io);
+    const waiters = this.onceWaiters.get(key);
+    if (!waiters) return;
+    const pending = waiters.splice(0);
+    for (const w of pending) {
+      if (w.timer) clearTimeout(w.timer);
+      if (w.signal && w.abortHandler) {
+        w.signal.removeEventListener("abort", w.abortHandler);
+      }
+      w.resolve({
+        ok: false,
+        error: new InputStreamTimeoutError(key, 0),
+      });
+    }
+  }
+
+  private invoke(handler: Handler, data: unknown): void {
+    try {
+      const result = handler(data);
+      if (result && typeof result === "object" && "catch" in result) {
+        (result as Promise<void>).catch(() => {});
+      }
+    } catch {
+      // Never let a handler error break test state
+    }
+  }
+
+  private removeWaiter(key: string, waiter: OnceWaiter): void {
+    const waiters = this.onceWaiters.get(key);
+    if (!waiters) return;
+    const idx = waiters.indexOf(waiter);
+    if (idx >= 0) waiters.splice(idx, 1);
+  }
+}