feat: AsyncIterable streaming support for library SDK

Streaming commands (log list --follow, dashboard view --refresh) now
return AsyncIterable instead of throwing. Consumer break and AbortSignal
both wire through to command abort for clean shutdown.

Closes #585

Author: BYK

AGENTS.md

@@ -893,6 +893,9 @@ mock.module("./some-module", () => ({
 
 ### Architecture
 
+<!-- lore:019d2d10-671c-77d8-9dbc-c32d1604dcf7 -->
+* **AsyncIterable streaming for SDK implemented via AsyncChannel push/pull pattern**: \`src/lib/async-channel.ts\` provides a dual-queue channel: producer calls \`push()\`/\`close()\`/\`error()\`, consumer iterates via \`for await...of\`. \`break\` triggers \`onReturn\` callback for cleanup. \`executeWithStream()\` in \`sdk-invoke.ts\` runs the command in background, pipes \`captureObject\` calls to the channel, and returns the channel immediately. Streaming detection: \`hasStreamingFlag()\` checks for \`--refresh\`/\`--follow\`/\`-f\`. \`buildInvoker\` accepts \`meta.streaming\` flag; \`buildRunner\` auto-detects from args. Abort wiring: \`AbortController\` created per stream, signal placed on fake \`process.abortSignal\`, \`channel.onReturn\` calls \`controller.abort()\`. Both \`log/list.ts\` and \`dashboard/view.ts\` check \`this.process?.abortSignal\` alongside SIGINT. Codegen generates callable interface overloads for streaming commands.
+
 <!-- lore:019cbeba-e4d3-748c-ad50-fe3c3d5c0a0d -->
 * **Auth token env var override pattern: SENTRY\_AUTH\_TOKEN > SENTRY\_TOKEN > SQLite**: Auth in \`src/lib/db/auth.ts\` follows layered precedence: \`SENTRY\_AUTH\_TOKEN\` > \`SENTRY\_TOKEN\` > SQLite OAuth token. \`getEnvToken()\` trims env vars (empty/whitespace = unset). \`AuthSource\` tracks provenance. \`ENV\_SOURCE\_PREFIX = "env:"\` — use \`.length\` not hardcoded 4. Env tokens bypass refresh/expiry. \`isEnvTokenActive()\` guards auth commands. Logout must NOT clear stored auth when env token active. These functions stay in \`db/auth.ts\` despite not touching DB because they're tightly coupled with token retrieval.
 

README.md

@@ -112,6 +112,9 @@ Options (all optional):
 - `project` — Default project slug.
 - `text` — Return human-readable string instead of parsed JSON (affects `run()` only).
 - `cwd` — Working directory for DSN auto-detection. Defaults to `process.cwd()`.
+- `signal` — `AbortSignal` to cancel streaming commands (`--follow`, `--refresh`).
+
+Streaming commands return `AsyncIterable` — use `for await...of` and `break` to stop.
 
 Errors are thrown as `SentryError` with `.exitCode` and `.stderr`.
 

docs/src/content/docs/library-usage.md

@@ -151,6 +151,7 @@ const sdk = createSentrySDK({ token: "...", text: true, cwd: "/my/project" });
 | `project` | `string` | Auto-detected | Default project slug |
 | `text` | `boolean` | `false` | Return human-readable text instead of parsed JSON (`run()` only) |
 | `cwd` | `string` | `process.cwd()` | Working directory for DSN auto-detection |
+| `signal` | `AbortSignal` | — | Abort signal for cancelling streaming commands |
 
 ## Return Values
 
@@ -217,7 +218,49 @@ Calls should be sequential (awaited one at a time).
 - **Node.js >= 22** (required for `node:sqlite`)
 - Or **Bun** (any recent version)
 
-:::caution
-Streaming flags (`--refresh`, `--follow`) are not supported in library mode
-and will throw a `SentryError`. Use the CLI binary directly for live-streaming commands.
+## Streaming Commands
+
+Two commands support real-time streaming: `log list --follow` and `dashboard view --refresh`.
+When using streaming flags, methods return an `AsyncIterable` instead of a `Promise`:
+
+```typescript
+const sdk = createSentrySDK({ token: "sntrys_..." });
+
+// Stream logs as they arrive (polls every 5 seconds)
+for await (const log of sdk.log.list({ follow: "5", orgProject: "acme/backend" })) {
+  console.log(log);
+}
+
+// Auto-refresh dashboard (polls every 30 seconds)
+for await (const snapshot of sdk.run("dashboard", "view", "123", "--refresh", "30")) {
+  console.log(snapshot);
+}
+
+// Stop streaming by breaking out of the loop
+for await (const log of sdk.log.list({ follow: "2" })) {
+  if (someCondition) break; // Streaming stops immediately
+}
+```
+
+### Cancellation
+
+`break` in a `for await...of` loop immediately signals the streaming command to stop.
+You can also pass an `AbortSignal` via `SentryOptions` for programmatic cancellation:
+
+```typescript
+const controller = new AbortController();
+const sdk = createSentrySDK({ token: "...", signal: controller.signal });
+
+// Cancel after 30 seconds
+setTimeout(() => controller.abort(), 30_000);
+
+for await (const log of sdk.log.list({ follow: "5" })) {
+  console.log(log);
+}
+// Loop exits when signal fires
+```
+
+:::note
+Concurrent streaming calls are not supported. Each streaming invocation
+uses an isolated environment — only one can be active at a time.
 :::

script/bundle.ts

@@ -226,6 +226,14 @@ const CORE_DECLARATIONS = `export type SentryOptions = {
   text?: boolean;
   /** Working directory (affects DSN detection, project root). Defaults to process.cwd(). */
   cwd?: string;
+  /** AbortSignal to cancel streaming commands (e.g. log list --follow). */
+  signal?: AbortSignal;
+};
+
+export type AsyncChannel<T> = AsyncIterable<T> & {
+  push(value: T): void;
+  close(): void;
+  error(err: Error): void;
 };
 
 export declare class SentryError extends Error {

script/generate-sdk.ts

@@ -34,11 +34,11 @@ const INTERNAL_FLAGS = new Set([
   "log-level",
   "verbose",
   "fields",
-  // Streaming flags produce infinite output — not supported in library mode
-  "refresh",
-  "follow",
 ]);
 
+/** Flags that trigger streaming mode — included in params but change return type */
+const STREAMING_FLAGS = new Set(["refresh", "follow"]);
+
 /** Regex for stripping angle-bracket/ellipsis decorators from placeholder names */
 const PLACEHOLDER_CLEAN_RE = /[<>.]/g;
 
@@ -341,13 +341,12 @@ function generateParamsInterface(
   return { name: interfaceName, code };
 }
 
-/** Generate the method body (invoke call) for a command. */
-function generateMethodBody(
+/** Build the flag object expression and positional expression for an invoke call. */
+function buildInvokeArgs(
   path: string[],
   positional: PositionalInfo,
-  flags: SdkFlagInfo[],
-  returnType: string
-): string {
+  flags: SdkFlagInfo[]
+): { flagObj: string; positionalExpr: string; pathStr: string } {
   const flagEntries = flags.map((f) => {
     const camel = camelCase(f.name);
     if (f.name !== camel) {
@@ -368,8 +367,65 @@ function generateMethodBody(
 
   const flagObj =
     flagEntries.length > 0 ? `{ ${flagEntries.join(", ")} }` : "{}";
+  const pathStr = JSON.stringify(path);
+
+  return { flagObj, positionalExpr, pathStr };
+}
+
+/**
+ * Generate the method body (invoke call) for a non-streaming command.
+ * Uses `as Promise<T>` to narrow the invoke union return type, since
+ * non-streaming calls never pass `meta.streaming` and always return a Promise.
+ */
+function generateMethodBody(
+  path: string[],
+  positional: PositionalInfo,
+  flags: SdkFlagInfo[],
+  returnType: string
+): string {
+  const { flagObj, positionalExpr, pathStr } = buildInvokeArgs(
+    path,
+    positional,
+    flags
+  );
+  return `invoke<${returnType}>(${pathStr}, ${flagObj}, ${positionalExpr}) as Promise<${returnType}>`;
+}
+
+/** Options for generating a streaming method body. */
+type StreamingMethodOpts = {
+  path: string[];
+  positional: PositionalInfo;
+  flags: SdkFlagInfo[];
+  returnType: string;
+  streamingFlagNames: string[];
+  indent: string;
+};
 
-  return `invoke<${returnType}>(${JSON.stringify(path)}, ${flagObj}, ${positionalExpr})`;
+/**
+ * Generate the method body for a streaming-capable command.
+ *
+ * Detects at runtime whether any streaming flag is present and passes
+ * `{ streaming: true }` to the invoker when it is.
+ */
+function generateStreamingMethodBody(opts: StreamingMethodOpts): string {
+  const { flagObj, positionalExpr, pathStr } = buildInvokeArgs(
+    opts.path,
+    opts.positional,
+    opts.flags
+  );
+
+  // Build the streaming condition: params?.follow !== undefined || params?.refresh !== undefined
+  const conditions = opts.streamingFlagNames
+    .map((name) => `params?.${camelCase(name)} !== undefined`)
+    .join(" || ");
+
+  const lines = [
+    "{",
+    `${opts.indent}    const streaming = ${conditions};`,
+    `${opts.indent}    return invoke<${opts.returnType}>(${pathStr}, ${flagObj}, ${positionalExpr}, { streaming });`,
+    `${opts.indent}  }`,
+  ];
+  return lines.join("\n");
 }
 
 // ---------------------------------------------------------------------------
@@ -472,6 +528,12 @@ for (const { path, command } of allCommands) {
   const flags = extractSdkFlags(command);
   const positional = derivePositional(command);
 
+  // Detect streaming-capable commands
+  const streamingFlagNames = flags
+    .filter((f) => STREAMING_FLAGS.has(f.name))
+    .map((f) => f.name);
+  const isStreaming = streamingFlagNames.length > 0;
+
   // Generate return type from schema
   const schemaTypeName = `${buildTypeName(path)}Result`;
   const returnTypeInfo = generateReturnType(command, schemaTypeName);
@@ -493,40 +555,87 @@ for (const { path, command } of allCommands) {
   let paramsArg: string;
   let body: string;
 
+  const brief = command.brief || path.join(" ");
+  const methodName = path.at(-1) ?? path[0];
+  const indent = "    ".repeat(path.length - 1);
+
   if (hasVariadicPositional) {
     // Variadic: (params: XParams, ...positional: string[]) or (params?: XParams, ...positional: string[])
     // Required flags make params required even with variadic positionals
     const paramsOpt = hasRequiredFlags ? "" : "?";
     paramsArg = params
       ? `params${paramsOpt}: ${params.name}, ...positional: string[]`
       : "...positional: string[]";
-    body = generateMethodBody(path, positional, flags, returnType);
+    body = isStreaming
+      ? generateStreamingMethodBody({
+          path,
+          positional,
+          flags,
+          returnType,
+          streamingFlagNames,
+          indent,
+        })
+      : generateMethodBody(path, positional, flags, returnType);
   } else if (params) {
     const paramsRequired = hasRequiredFlags;
     paramsArg = paramsRequired
       ? `params: ${params.name}`
       : `params?: ${params.name}`;
-    body = generateMethodBody(path, positional, flags, returnType);
+    body = isStreaming
+      ? generateStreamingMethodBody({
+          path,
+          positional,
+          flags,
+          returnType,
+          streamingFlagNames,
+          indent,
+        })
+      : generateMethodBody(path, positional, flags, returnType);
   } else {
     body = generateMethodBody(path, positional, flags, returnType);
     paramsArg = "";
   }
 
-  const brief = command.brief || path.join(" ");
-  const methodName = path.at(-1) ?? path[0];
-  const indent = "    ".repeat(path.length - 1);
   const sig = paramsArg ? `(${paramsArg})` : "()";
-  const methodCode = [
-    `${indent}    /** ${brief} */`,
-    `${indent}    ${methodName}: ${sig}: Promise<${returnType}> =>`,
-    `${indent}      ${body},`,
-  ].join("\n");
-
-  // Type declaration: method signature without implementation
-  const typeDecl = [
-    `${indent}    /** ${brief} */`,
-    `${indent}    ${methodName}${sig}: Promise<${returnType}>;`,
-  ].join("\n");
+
+  let methodCode: string;
+  let typeDecl: string;
+
+  if (isStreaming) {
+    // Streaming commands use a function body (not arrow expression)
+    // because they need runtime streaming detection
+    methodCode = [
+      `${indent}    /** ${brief} */`,
+      `${indent}    ${methodName}: ${sig} => ${body},`,
+    ].join("\n");
+
+    // Type declaration: callable interface with overloaded signatures
+    const streamingFlagTypes = streamingFlagNames.map((name) => {
+      const flag = flags.find((f) => f.name === name);
+      return `${camelCase(name)}: ${flag?.tsType ?? "string"}`;
+    });
+    const streamingConstraint = streamingFlagTypes.join("; ");
+
+    typeDecl = [
+      `${indent}    /** ${brief} */`,
+      `${indent}    ${methodName}: {`,
+      `${indent}      (params: ${params?.name ?? "Record<string, never>"} & { ${streamingConstraint} }): AsyncIterable<unknown>;`,
+      `${indent}      ${sig}: Promise<${returnType}>;`,
+      `${indent}    };`,
+    ].join("\n");
+  } else {
+    methodCode = [
+      `${indent}    /** ${brief} */`,
+      `${indent}    ${methodName}: ${sig}: Promise<${returnType}> =>`,
+      `${indent}      ${body},`,
+    ].join("\n");
+
+    // Type declaration: method signature without implementation
+    typeDecl = [
+      `${indent}    /** ${brief} */`,
+      `${indent}    ${methodName}${sig}: Promise<${returnType}>;`,
+    ].join("\n");
+  }
 
   insertMethod(root, path, methodCode, typeDecl);
 }

src/commands/dashboard/view.ts

@@ -219,6 +219,13 @@ export const viewCommand = buildCommand({
       const stop = () => controller.abort();
       process.once("SIGINT", stop);
 
+      // Library mode: honor external abort signal (e.g., consumer break)
+      const externalSignal = (this.process as { abortSignal?: AbortSignal })
+        ?.abortSignal;
+      if (externalSignal) {
+        externalSignal.addEventListener("abort", stop, { once: true });
+      }
+
       let isFirstRender = true;
 
       try {

src/commands/log/list.ts

@@ -256,6 +256,11 @@ type FollowGeneratorConfig<T extends LogLike> = {
    * Use this to seed dedup state (e.g., tracking seen log IDs).
    */
   onInitialLogs?: (logs: T[]) => void;
+  /**
+   * External abort signal (library mode). When aborted, the follow
+   * generator stops on the next poll cycle. Complements SIGINT handling.
+   */
+  abortSignal?: AbortSignal;
 };
 
 /** Find the highest timestamp_precise in a batch, or undefined if none have it. */
@@ -343,11 +348,16 @@ async function* generateFollowLogs<T extends LogLike>(
   // timestamp_precise is nanoseconds; Date.now() is milliseconds → convert
   let lastTimestamp = Date.now() * 1_000_000;
 
-  // AbortController for clean SIGINT handling
+  // AbortController for clean SIGINT handling + library mode abort
   const controller = new AbortController();
   const stop = () => controller.abort();
   process.once("SIGINT", stop);
 
+  // Library mode: honor external abort signal (e.g., consumer break)
+  if (config.abortSignal) {
+    config.abortSignal.addEventListener("abort", stop, { once: true });
+  }
+
   try {
     // Initial fetch
     const initialLogs = await config.fetch("1m");
@@ -695,6 +705,8 @@ export const listCommand = buildListCommand(
           const generator = generateFollowLogs({
             flags,
             onDiagnostic: (msg) => logger.warn(msg),
+            abortSignal: (this.process as { abortSignal?: AbortSignal })
+              ?.abortSignal,
             fetch: (statsPeriod) =>
               listTraceLogs(org, traceId, {
                 query: flags.query,
@@ -757,6 +769,8 @@ export const listCommand = buildListCommand(
           const generator = generateFollowLogs({
             flags,
             onDiagnostic: (msg) => logger.warn(msg),
+            abortSignal: (this.process as { abortSignal?: AbortSignal })
+              ?.abortSignal,
             fetch: (statsPeriod, afterTimestamp) =>
               listLogs(org, project, {
                 query: flags.query,

src/index.ts

@@ -25,6 +25,7 @@
 import { buildInvoker, buildRunner } from "./lib/sdk-invoke.js";
 import { createSDKMethods } from "./sdk.generated.js";
 
+export type { AsyncChannel } from "./lib/async-channel.js";
 // Re-export public types and error class from the shared module.
 // These re-exports exist to break a circular dependency between
 // index.ts ↔ sdk-invoke.ts. SentryError and SentryOptions live