feat(sdk,core): Session client SDK + hello-world smoke test

Client-side pair to the Session primitive server PR (TRI-8627). Run-scoped streams.pipe / streams.input are untouched. @trigger.dev/core ApiClient - createSession / retrieveSession / updateSession / closeSession — zodfetch against /api/v1/sessions control plane - listSessions — CursorPagePromise<SessionItem>, follows the runs/waitpoints convention (page[size], page[after], page[before] + filter[*]) - initializeSessionStream — PUT /realtime/v1/sessions/:session/:io, returns S2 creds in headers (feeds StreamsWriterV2 directly) - appendToSessionStream — POST …/append - subscribeToSessionStream — reuses SSEStreamSubscription for SSE subscribes (auto-retry, Last-Event-ID resume, abort propagation), so session subscribers get the exact same semantics as runs.fetchStream. Returns AsyncIterableStream<T>. @trigger.dev/sdk sessions namespace - sessions.create / retrieve / update / close / list — wraps the ApiClient with the standard tracer + accessoryAttributes + mergeRequestOptions. Returns ApiPromise / CursorPagePromise. - sessions.open(id) returns a SessionHandle with .out and .in SessionChannels. Each channel exposes append / send / subscribe / initialize. The handle is polymorphic on friendlyId or externalId. - auth.ts adds the `sessions` permission on PublicTokenPermissionProperties so auth.createPublicToken({ read: { sessions: ["session_abc"] } }) works. Reference - references/hello-world/src/trigger/sessionsSmoke.ts — idempotent Trigger.dev task that exercises every code path (control-plane CRUD, polymorphic lookup, list with tag/type/status/externalId filters, cursor pagination, out.initialize + append + subscribe SSE round-trip, in.send, close + idempotent re-close). Trigger via mcp__trigger__trigger_task(taskId: "sessions-smoke"). Verified live against the local webapp (project hello-world): 10/10 steps pass end-to-end, S2 round-trip returns appended chunks through the shared SSEStreamSubscription pipeline.
2026-04-20 14:50:38 +01:00
parent a8466343ba
commit 85db7bd612
5 changed files with 714 additions and 0 deletions
@@ -13,8 +13,16 @@ import {
  BatchTriggerTaskV3RequestBody,
  BatchTriggerTaskV3Response,
  CanceledRunResponse,
+  CloseSessionRequestBody,
  CompleteWaitpointTokenRequestBody,
  CompleteWaitpointTokenResponseBody,
+  CreatedSessionResponseBody,
+  CreateSessionRequestBody,
+  ListSessionsOptions,
+  ListSessionsResponseBody,
+  ListedSessionItem,
+  RetrieveSessionResponseBody,
+  UpdateSessionRequestBody,
  CreateBatchRequestBody,
  CreateBatchResponse,
  CreateEnvironmentVariableRequestBody,
@@ -1095,6 +1103,182 @@ export class ApiClient {
    );
  }

+  // ========================================================================
+  // Sessions
+  // ========================================================================
+
+  createSession(body: CreateSessionRequestBody, requestOptions?: ZodFetchOptions) {
+    return zodfetch(
+      CreatedSessionResponseBody,
+      `${this.baseUrl}/api/v1/sessions`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(body),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  retrieveSession(sessionIdOrExternalId: string, requestOptions?: ZodFetchOptions) {
+    return zodfetch(
+      RetrieveSessionResponseBody,
+      `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`,
+      {
+        method: "GET",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  updateSession(
+    sessionIdOrExternalId: string,
+    body: UpdateSessionRequestBody,
+    requestOptions?: ZodFetchOptions
+  ) {
+    return zodfetch(
+      RetrieveSessionResponseBody,
+      `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`,
+      {
+        method: "PATCH",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(body),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  closeSession(
+    sessionIdOrExternalId: string,
+    body?: CloseSessionRequestBody,
+    requestOptions?: ZodFetchOptions
+  ) {
+    return zodfetch(
+      RetrieveSessionResponseBody,
+      `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/close`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(body ?? {}),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  listSessions(
+    options?: ListSessionsOptions,
+    requestOptions?: ZodFetchOptions
+  ): CursorPagePromise<typeof ListedSessionItem> {
+    const searchParams = createSearchQueryForListSessions(options);
+
+    return zodfetchCursorPage(
+      ListedSessionItem,
+      `${this.baseUrl}/api/v1/sessions`,
+      {
+        query: searchParams,
+        limit: options?.limit,
+        after: options?.after,
+        before: options?.before,
+      },
+      {
+        method: "GET",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  // ========================================================================
+  // Session realtime channels
+  // ========================================================================
+
+  async initializeSessionStream(
+    sessionIdOrExternalId: string,
+    io: "out" | "in",
+    requestOptions?: ZodFetchOptions
+  ) {
+    // The server returns S2 credentials in response headers alongside a tiny
+    // JSON body with the realtime version. Follow the same shape as
+    // `createStream` so downstream clients can feed them into
+    // `StreamsWriterV2`.
+    return zodfetch(
+      CreateStreamResponseBody,
+      `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`,
+      {
+        method: "PUT",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    )
+      .withResponse()
+      .then(({ data, response }) => ({
+        ...data,
+        headers: Object.fromEntries(response.headers.entries()),
+      }));
+  }
+
+  async appendToSessionStream<TBody extends BodyInit>(
+    sessionIdOrExternalId: string,
+    io: "out" | "in",
+    part: TBody,
+    requestOptions?: ZodFetchOptions
+  ) {
+    return zodfetch(
+      AppendToStreamResponseBody,
+      `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}/append`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: part,
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  /**
+   * Subscribe to SSE records on a Session channel. Reuses the same
+   * {@link SSEStreamSubscription} plumbing as `readStream` for run-scoped
+   * realtime streams — auto-retry, Last-Event-ID resume, abort-on-cancel.
+   */
+  async subscribeToSessionStream<T = unknown>(
+    sessionIdOrExternalId: string,
+    io: "out" | "in",
+    options?: {
+      signal?: AbortSignal;
+      baseUrl?: string;
+      timeoutInSeconds?: number;
+      onComplete?: () => void;
+      onError?: (error: Error) => void;
+      lastEventId?: string;
+      onPart?: (part: SSEStreamPart<T>) => void;
+    }
+  ): Promise<AsyncIterableStream<T>> {
+    const url = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`;
+
+    const subscription = new SSEStreamSubscription(url, {
+      headers: this.getHeaders(),
+      signal: options?.signal,
+      onComplete: options?.onComplete,
+      onError: options?.onError,
+      timeoutInSeconds: options?.timeoutInSeconds,
+      lastEventId: options?.lastEventId,
+    });
+
+    const stream = await subscription.subscribe();
+    const onPart = options?.onPart;
+
+    return stream.pipeThrough(
+      new TransformStream<SSEStreamPart, T>({
+        transform(chunk, controller) {
+          const data = chunk.chunk as T;
+          onPart?.(chunk as SSEStreamPart<T>);
+          controller.enqueue(data);
+        },
+      })
+    );
+  }
+
  async waitForDuration(
    runId: string,
    body: WaitForDurationRequestBody,
@@ -1836,6 +2020,47 @@ function queueNameFromQueueTypeName(queue: QueueTypeName): string {
  return queue.name;
 }

+function createSearchQueryForListSessions(options?: ListSessionsOptions): URLSearchParams {
+  const searchParams = new URLSearchParams();
+
+  if (!options) return searchParams;
+
+  const appendMany = (name: string, value: string | string[] | undefined) => {
+    if (value === undefined) return;
+    searchParams.append(name, Array.isArray(value) ? value.join(",") : value);
+  };
+
+  appendMany("filter[type]", options.type);
+  appendMany("filter[tags]", options.tag);
+  appendMany("filter[taskIdentifier]", options.taskIdentifier);
+
+  if (options.externalId) {
+    searchParams.append("filter[externalId]", options.externalId);
+  }
+
+  appendMany("filter[status]", options.status as string | string[] | undefined);
+
+  if (options.period) {
+    searchParams.append("filter[createdAt][period]", options.period);
+  }
+
+  if (options.from !== undefined) {
+    searchParams.append(
+      "filter[createdAt][from]",
+      options.from instanceof Date ? options.from.getTime().toString() : options.from.toString()
+    );
+  }
+
+  if (options.to !== undefined) {
+    searchParams.append(
+      "filter[createdAt][to]",
+      options.to instanceof Date ? options.to.getTime().toString() : options.to.toString()
+    );
+  }
+
+  return searchParams;
+}
+
 function createSearchQueryForListWaitpointTokens(
  query?: ListWaitpointTokensQueryParams
 ): URLSearchParams {
@@ -67,6 +67,17 @@ type PublicTokenPermissionProperties = {
   * Grant access to send data to input streams on specific runs
   */
  inputStreams?: string | string[];
+
+  /**
+   * Grant access to specific Sessions (the durable, typed I/O primitive that
+   * outlives a single run). Use the session's friendlyId (e.g. `session_abc`).
+   *
+   * `read:sessions:{id}` lets the bearer read both the `.out` and `.in`
+   * channels and list runs on the session. `write:sessions:{id}` lets the
+   * bearer append to the session's channels. `trigger:sessions:{id}` permits
+   * triggering new runs on the session.
+   */
+  sessions?: string | string[];
 };

 export type PublicTokenPermissions = {
@@ -17,6 +17,7 @@ export * from "./otel.js";
 export * from "./schemas.js";
 export * from "./heartbeats.js";
 export * from "./streams.js";
+export * from "./sessions.js";
 export * from "./query.js";
 export type { Context };

@@ -0,0 +1,318 @@
+import type {
+  ApiPromise,
+  ApiRequestOptions,
+  AsyncIterableStream,
+  CloseSessionRequestBody,
+  CreatedSessionResponseBody,
+  CreateSessionRequestBody,
+  ListSessionsOptions,
+  ListedSessionItem,
+  RetrieveSessionResponseBody,
+  UpdateSessionRequestBody,
+} from "@trigger.dev/core/v3";
+import {
+  CursorPagePromise,
+  accessoryAttributes,
+  apiClientManager,
+  mergeRequestOptions,
+} from "@trigger.dev/core/v3";
+import { tracer } from "./tracer.js";
+
+export type {
+  CreatedSessionResponseBody,
+  CreateSessionRequestBody,
+  CloseSessionRequestBody,
+  ListSessionsOptions,
+  ListedSessionItem,
+  RetrieveSessionResponseBody,
+  UpdateSessionRequestBody,
+};
+
+export const sessions = {
+  create: createSession,
+  retrieve: retrieveSession,
+  update: updateSession,
+  close: closeSession,
+  list: listSessions,
+  open,
+};
+
+/**
+ * Create a {@link Session} — a durable, typed, bidirectional I/O primitive
+ * that outlives a single run. Idempotent via `externalId`.
+ */
+function createSession(
+  body: CreateSessionRequestBody,
+  requestOptions?: ApiRequestOptions
+): ApiPromise<CreatedSessionResponseBody> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "sessions.create()",
+      icon: "sessions",
+      attributes: sessionAttributes(body.externalId ?? body.type, {
+        type: body.type,
+        ...(body.externalId ? { externalId: body.externalId } : {}),
+      }),
+    },
+    requestOptions
+  );
+
+  return apiClient.createSession(body, $requestOptions);
+}
+
+/**
+ * Retrieve a Session by `friendlyId` (`session_*`) or user-supplied
+ * `externalId`. The server disambiguates via the `session_` prefix.
+ */
+function retrieveSession(
+  sessionIdOrExternalId: string,
+  requestOptions?: ApiRequestOptions
+): ApiPromise<RetrieveSessionResponseBody> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "sessions.retrieve()",
+      icon: "sessions",
+      attributes: sessionAttributes(sessionIdOrExternalId),
+    },
+    requestOptions
+  );
+
+  return apiClient.retrieveSession(sessionIdOrExternalId, $requestOptions);
+}
+
+/** Update mutable fields on a Session (tags, metadata, externalId). */
+function updateSession(
+  sessionIdOrExternalId: string,
+  body: UpdateSessionRequestBody,
+  requestOptions?: ApiRequestOptions
+): ApiPromise<RetrieveSessionResponseBody> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "sessions.update()",
+      icon: "sessions",
+      attributes: sessionAttributes(sessionIdOrExternalId),
+    },
+    requestOptions
+  );
+
+  return apiClient.updateSession(sessionIdOrExternalId, body, $requestOptions);
+}
+
+/** Mark a Session as closed (terminal, idempotent). */
+function closeSession(
+  sessionIdOrExternalId: string,
+  body?: CloseSessionRequestBody,
+  requestOptions?: ApiRequestOptions
+): ApiPromise<RetrieveSessionResponseBody> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "sessions.close()",
+      icon: "sessions",
+      attributes: sessionAttributes(sessionIdOrExternalId, {
+        ...(body?.reason ? { reason: body.reason } : {}),
+      }),
+    },
+    requestOptions
+  );
+
+  return apiClient.closeSession(sessionIdOrExternalId, body, $requestOptions);
+}
+
+/**
+ * List Sessions in the current environment with filters + cursor pagination.
+ * Returns a {@link CursorPagePromise} so callers can iterate pages with
+ * `for await`.
+ */
+function listSessions(
+  options?: ListSessionsOptions,
+  requestOptions?: ApiRequestOptions
+): CursorPagePromise<typeof ListedSessionItem> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "sessions.list()",
+      icon: "sessions",
+      attributes: {
+        ...(options?.type ? { type: toAttr(options.type) } : {}),
+        ...(options?.tag ? { tag: toAttr(options.tag) } : {}),
+        ...(options?.status ? { status: toAttr(options.status) } : {}),
+        ...(options?.externalId ? { externalId: options.externalId } : {}),
+      },
+    },
+    requestOptions
+  );
+
+  return apiClient.listSessions(options, $requestOptions);
+}
+
+/**
+ * Open a lightweight handle to a Session's realtime channels. Does not
+ * perform a network call on its own — each channel method hits the
+ * corresponding realtime endpoint.
+ */
+function open(sessionIdOrExternalId: string): SessionHandle {
+  return new SessionHandle(sessionIdOrExternalId);
+}
+
+export class SessionHandle {
+  public readonly out: SessionChannel;
+  public readonly in: SessionChannel;
+
+  constructor(public readonly id: string) {
+    this.out = new SessionChannel(id, "out");
+    this.in = new SessionChannel(id, "in");
+  }
+}
+
+export type SessionChannelCredentials = {
+  accessToken: string;
+  basin: string;
+  streamName: string;
+  endpoint?: string;
+  flushIntervalMs?: number;
+  maxRetries?: number;
+};
+
+/** One direction of a Session's bidirectional channel pair. */
+export class SessionChannel {
+  constructor(
+    public readonly sessionId: string,
+    public readonly io: "out" | "in"
+  ) {}
+
+  /**
+   * Append a single record to this channel via the server-side append
+   * endpoint. For high-throughput writes use {@link initialize} to get S2
+   * credentials and write directly to S2.
+   */
+  async append(part: string | unknown, requestOptions?: ApiRequestOptions): Promise<void> {
+    const apiClient = apiClientManager.clientOrThrow();
+    const body = typeof part === "string" ? part : JSON.stringify(part);
+
+    const $requestOptions = mergeRequestOptions(
+      {
+        tracer,
+        name: `sessions.open(${this.sessionId}).${this.io}.append()`,
+        icon: "sessions",
+        attributes: sessionAttributes(this.sessionId, { io: this.io }),
+      },
+      requestOptions
+    );
+
+    await apiClient.appendToSessionStream(this.sessionId, this.io, body, $requestOptions);
+  }
+
+  /**
+   * Friendly alias for `channel.append(value)` used on the `.in` channel by
+   * clients producing messages for the task runtime.
+   */
+  send(value: unknown, requestOptions?: ApiRequestOptions): Promise<void> {
+    return this.append(value, requestOptions);
+  }
+
+  /**
+   * Subscribe to SSE records on this channel. Delegates to the shared
+   * {@link SSEStreamSubscription} plumbing (auto-retry, Last-Event-ID
+   * resume, abort propagation) used by run-scoped realtime streams —
+   * session subscribers get the same guarantees.
+   */
+  async subscribe<T = unknown>(
+    options?: SessionSubscribeOptions<T>
+  ): Promise<AsyncIterableStream<T>> {
+    const apiClient = apiClientManager.clientOrThrow();
+
+    return apiClient.subscribeToSessionStream<T>(this.sessionId, this.io, {
+      signal: options?.signal,
+      timeoutInSeconds: options?.timeoutInSeconds,
+      lastEventId:
+        options?.lastEventId != null ? String(options.lastEventId) : undefined,
+      onPart: options?.onPart,
+      onComplete: options?.onComplete,
+      onError: options?.onError,
+    });
+  }
+
+  /**
+   * Fetch S2 credentials for direct-to-S2 writes. Returns the same header
+   * bag the server hands to {@link StreamsWriterV2}.
+   */
+  async initialize(requestOptions?: ApiRequestOptions): Promise<SessionChannelCredentials> {
+    const apiClient = apiClientManager.clientOrThrow();
+
+    const $requestOptions = mergeRequestOptions(
+      {
+        tracer,
+        name: `sessions.open(${this.sessionId}).${this.io}.initialize()`,
+        icon: "sessions",
+        attributes: sessionAttributes(this.sessionId, { io: this.io }),
+      },
+      requestOptions
+    );
+
+    const response = await apiClient.initializeSessionStream(
+      this.sessionId,
+      this.io,
+      $requestOptions
+    );
+
+    return {
+      accessToken: response.headers["x-s2-access-token"] ?? "",
+      basin: response.headers["x-s2-basin"] ?? "",
+      streamName: response.headers["x-s2-stream-name"] ?? "",
+      endpoint: response.headers["x-s2-endpoint"],
+      flushIntervalMs: numHeader(response.headers, "x-s2-flush-interval-ms"),
+      maxRetries: numHeader(response.headers, "x-s2-max-retries"),
+    };
+  }
+}
+
+export type SessionSubscribeOptions<T = unknown> = {
+  signal?: AbortSignal;
+  lastEventId?: string | number;
+  /** Timeout in seconds for the underlying long-poll (max 600). */
+  timeoutInSeconds?: number;
+  /** Called for each SSE event with the full event metadata (id, timestamp). */
+  onPart?: (part: { id: string; chunk: T; timestamp: number }) => void;
+  /** Called when the server signals end-of-stream. */
+  onComplete?: () => void;
+  /** Called on unrecoverable errors after the retry budget is exhausted. */
+  onError?: (error: Error) => void;
+};
+
+// ─── helpers ────────────────────────────────────────────────────────
+
+function sessionAttributes(id: string, extra?: Record<string, string | number | boolean>) {
+  return {
+    session: id,
+    ...(extra ?? {}),
+    ...accessoryAttributes({
+      items: [{ text: id, variant: "normal" }],
+      style: "codepath",
+    }),
+  };
+}
+
+function toAttr(value: string | string[]): string {
+  return Array.isArray(value) ? value.join(",") : value;
+}
+
+function numHeader(headers: Record<string, string | undefined>, name: string): number | undefined {
+  const raw = headers[name];
+  if (raw == null) return undefined;
+  const parsed = Number(raw);
+  return Number.isFinite(parsed) ? parsed : undefined;
+}
@@ -0,0 +1,159 @@
+import { logger, sessions, task } from "@trigger.dev/sdk";
+
+/**
+ * End-to-end smoke test for the Session SDK and server routes.
+ *
+ * Exercises every code path:
+ *   - control-plane CRUD (create / retrieve / update / close)
+ *   - polymorphic lookup via friendlyId and externalId
+ *   - list with tag / type / externalId filters
+ *   - cursor pagination (page 1 -> page 2)
+ *   - realtime `.out` initialize + append + subscribe (SSE round-trip)
+ *   - realtime `.in` send
+ *   - idempotent close
+ *
+ * Trigger from the dashboard or via the MCP `trigger_task` tool:
+ *
+ *   mcp__trigger__trigger_task(taskId: "sessions-smoke", payload: {})
+ *
+ * Inside a run, the SDK picks up the ambient environment credentials, so
+ * no `configure()` call is needed.
+ */
+export const sessionsSmoke = task({
+  id: "sessions-smoke",
+  run: async () => {
+    const runId = Date.now();
+    const results: Record<string, unknown> = {};
+
+    logger.info("sessions.create");
+    const created = await sessions.create({
+      type: "chat.agent",
+      externalId: `smoke-${runId}`,
+      tags: ["smoketest", "sdk"],
+      metadata: { purpose: "session-smoketest", runId },
+    });
+    results.created = { id: created.id, isCached: created.isCached };
+
+    logger.info("sessions.retrieve by friendlyId");
+    const byId = await sessions.retrieve(created.id);
+    results.retrievedByFriendlyId = byId.externalId;
+
+    logger.info("sessions.retrieve by externalId (polymorphic)");
+    const byExt = await sessions.retrieve(created.externalId!);
+    results.retrievedByExternalId = byExt.id;
+
+    logger.info("sessions.update tags + metadata");
+    const updated = await sessions.update(created.id, {
+      tags: ["smoketest", "sdk", "updated"],
+      metadata: { purpose: "session-smoketest", runId, touched: true },
+    });
+    results.updated = {
+      tags: updated.tags,
+      touched: (updated.metadata as Record<string, unknown> | null)?.touched,
+    };
+
+    const handle = sessions.open(created.externalId!);
+
+    logger.info("sessions.open(...).out.initialize (S2 creds)");
+    const outCreds = await handle.out.initialize();
+    results.outInitialize = { basin: outCreds.basin, streamName: outCreds.streamName };
+
+    logger.info("sessions.open(...).out.append x2 + .in.send x1");
+    await handle.out.append({ chunk: "first", ts: Date.now() });
+    await handle.out.append({ chunk: "second", ts: Date.now() });
+    await handle.in.send({ role: "user", content: "hello from smoketest" });
+
+    logger.info("sessions.open(...).out.subscribe (SSE round-trip)");
+    const controller = new AbortController();
+    const timer = setTimeout(() => controller.abort(), 4000);
+    const received: Array<{ id: string; chunk: unknown }> = [];
+    try {
+      const stream = await handle.out.subscribe({
+        signal: controller.signal,
+        timeoutInSeconds: 3,
+        onPart: (part) => {
+          received.push({ id: part.id, chunk: part.chunk });
+        },
+      });
+      let count = 0;
+      for await (const _chunk of stream) {
+        count += 1;
+        if (count >= 2) break;
+      }
+    } catch (err) {
+      if ((err as Error).name !== "AbortError") throw err;
+    } finally {
+      clearTimeout(timer);
+    }
+    results.subscribed = received;
+
+    // Seed a couple extra sessions so list queries have multiple hits.
+    await sessions.create({
+      type: "chat.agent",
+      externalId: `smoke-${runId}-b`,
+      tags: ["smoketest"],
+    });
+    await sessions.create({
+      type: "run.output",
+      externalId: `smoke-${runId}-c`,
+      tags: ["smoketest"],
+    });
+
+    // Let ClickHouse replication catch up.
+    await new Promise((resolve) => setTimeout(resolve, 1500));
+
+    logger.info("sessions.list by tag");
+    const listedAll = await sessions.list({ tag: "smoketest", limit: 50 });
+    results.listByTag = listedAll.data.length;
+
+    logger.info("sessions.list type + tag");
+    const listedChat = await sessions.list({
+      type: "chat.agent",
+      tag: "smoketest",
+      limit: 50,
+    });
+    results.listByTypeAndTag = {
+      count: listedChat.data.length,
+      types: [...new Set(listedChat.data.map((session) => session.type))],
+    };
+
+    logger.info("sessions.list by externalId");
+    const listedOne = await sessions.list({ externalId: `smoke-${runId}` });
+    results.listByExternalId = {
+      count: listedOne.data.length,
+      match: listedOne.data[0]?.id === created.id,
+    };
+
+    logger.info("sessions.list pagination");
+    const page1 = await sessions.list({ tag: "smoketest", limit: 2 });
+    let page2Ids: string[] = [];
+    if (page1.pagination.next) {
+      const page2 = await sessions.list({
+        tag: "smoketest",
+        limit: 2,
+        after: page1.pagination.next,
+      });
+      page2Ids = page2.data.map((s) => s.id);
+    }
+    results.pagination = {
+      page1Ids: page1.data.map((s) => s.id),
+      next: page1.pagination.next,
+      page2Ids,
+    };
+
+    logger.info("sessions.close");
+    const closed = await sessions.close(created.externalId!, { reason: "smoketest-done" });
+    results.closed = { closedAt: closed.closedAt, reason: closed.closedReason };
+
+    logger.info("sessions.close (idempotent)");
+    const reclosed = await sessions.close(created.externalId!, {
+      reason: "should-not-clobber",
+    });
+    results.idempotentClose = {
+      closedAtUnchanged: reclosed.closedAt?.toString() === closed.closedAt?.toString(),
+      reasonUnchanged: reclosed.closedReason === closed.closedReason,
+    };
+
+    return { ok: true, runId, results };
+  },
+});