diff --git a/nodejs/src/client.ts b/nodejs/src/client.ts
index 65784569cc..a146874530 100644
--- a/nodejs/src/client.ts
+++ b/nodejs/src/client.ts
@@ -85,6 +85,7 @@ import type {
     TypedSessionLifecycleHandler,
 } from "./types.js";
 import { defaultJoinSessionPermissionHandler } from "./types.js";
+import type { WorkflowHandle } from "./workflow.js";
 
 /**
  * Minimum protocol version this SDK can communicate with.
@@ -1657,6 +1658,23 @@ export class CopilotClient {
      * ```
      */
     async resumeSession(sessionId: string, config: ResumeSessionConfig): Promise<CopilotSession> {
+        return this.resumeSessionInternal(sessionId, config);
+    }
+
+    /** @internal */
+    async resumeSessionForExtension(
+        sessionId: string,
+        config: ResumeSessionConfig,
+        workflows?: WorkflowHandle[]
+    ): Promise<CopilotSession> {
+        return this.resumeSessionInternal(sessionId, config, workflows);
+    }
+
+    private async resumeSessionInternal(
+        sessionId: string,
+        config: ResumeSessionConfig,
+        workflows?: WorkflowHandle[]
+    ): Promise<CopilotSession> {
         if (!this.connection) {
             await this.start();
         }
@@ -1673,6 +1691,7 @@ export class CopilotClient {
         session.registerTools(config.tools);
         session.registerCanvases(config.canvases);
         session.registerCommands(config.commands);
+        session.registerWorkflows(workflows);
         const {
             wireProvider: bearerWireProvider,
             wireProviders: bearerWireProviders,
@@ -1743,6 +1762,7 @@ export class CopilotClient {
                 })),
                 toolSearch: config.toolSearch,
                 canvases: config.canvases?.map((canvas) => canvas.declaration),
+                workflows: workflows?.map((workflow) => workflow.meta),
                 requestCanvasRenderer: config.requestCanvasRenderer,
                 requestExtensions: config.requestExtensions,
                 extensionSdkPath: config.extensionSdkPath,
diff --git a/nodejs/src/extension.ts b/nodejs/src/extension.ts
index 72bde93bde..0d55379f5c 100644
--- a/nodejs/src/extension.ts
+++ b/nodejs/src/extension.ts
@@ -6,10 +6,10 @@ import { CopilotClient } from "./client.js";
 import type { CopilotSession } from "./session.js";
 import {
     defaultJoinSessionPermissionHandler,
-    type ExtensionInfo,
     type PermissionHandler,
     type ResumeSessionConfig,
 } from "./types.js";
+import type { WorkflowHandle } from "./workflow.js";
 
 export {
     Canvas,
@@ -27,9 +27,23 @@ export type JoinSessionConfig = Omit<
     "onPermissionRequest" | "extensionSdkPath"
 > & {
     onPermissionRequest?: PermissionHandler;
+    workflows?: WorkflowHandle[];
 };
 
-export type { ExtensionInfo };
+export type { ExtensionInfo, WorkflowLimits, WorkflowMeta } from "./types.js";
+export {
+    defineWorkflow,
+    WorkflowRunError,
+    type RunOptions,
+    type SessionWorkflowApi,
+    type WorkflowAgentOptions,
+    type WorkflowContext,
+    type WorkflowDefinition,
+    type WorkflowHandle,
+    type WorkflowJsonSchema,
+    type WorkflowPipelineStage,
+    type WorkflowStepOptions,
+} from "./workflow.js";
 
 /**
  * Joins the current foreground session.
@@ -58,14 +72,22 @@ export async function joinSession(config: JoinSessionConfig = {}): Promise<Copil
     // at the type level — untyped (JS) callers can still slip it through, and
     // honoring it here would be misleading since the extension subprocess has
     // already been forked by the host with the SDK the host chose.
-    const { extensionSdkPath: _stripped, ...rest } = config as JoinSessionConfig & {
+    const {
+        extensionSdkPath: _stripped,
+        workflows,
+        ...rest
+    } = config as JoinSessionConfig & {
         extensionSdkPath?: string;
     };
     void _stripped;
 
-    return client.resumeSession(sessionId, {
-        ...rest,
-        onPermissionRequest: config.onPermissionRequest ?? defaultJoinSessionPermissionHandler,
-        suppressResumeEvent: config.suppressResumeEvent ?? true,
-    });
+    return client.resumeSessionForExtension(
+        sessionId,
+        {
+            ...rest,
+            onPermissionRequest: config.onPermissionRequest ?? defaultJoinSessionPermissionHandler,
+            suppressResumeEvent: config.suppressResumeEvent ?? true,
+        },
+        workflows
+    );
 }
diff --git a/nodejs/src/generated/rpc.ts b/nodejs/src/generated/rpc.ts
index 2abba23d5d..77a099948c 100644
--- a/nodejs/src/generated/rpc.ts
+++ b/nodejs/src/generated/rpc.ts
@@ -2060,6 +2060,38 @@ export type UISessionLimitsExhaustedResponseAction =
   | "unset"
   /** Leave the limit unchanged and cancel the blocked model request. */
   | "cancel";
+/**
+ * Kind of workflow progress line.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowLogLineKind".
+ */
+/** @experimental */
+export type WorkflowLogLineKind =
+  /** A narrator log line. */
+  | "log"
+  /** A named workflow phase marker. */
+  | "phase";
+/**
+ * Current or terminal state of a workflow run.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowRunStatus".
+ */
+/** @experimental */
+export type WorkflowRunStatus =
+  /** The run was minted and is awaiting approval. */
+  | "pending"
+  /** The run is executing. */
+  | "running"
+  /** The run completed successfully. */
+  | "completed"
+  /** The run stopped after reaching a resource ceiling. */
+  | "halted"
+  /** The run was cancelled before completion. */
+  | "cancelled"
+  /** The workflow body failed. */
+  | "error";
 /**
  * Type of change represented by this file diff.
  *
@@ -10949,6 +10981,23 @@ export interface RemoteSessionRepository {
    */
   branch?: string;
 }
+/**
+ * Options controlling workflow invocation.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "RunOptions".
+ */
+/** @experimental */
+export interface RunOptions {
+  /**
+   * Whether to return once the approved run starts instead of awaiting its terminal result.
+   */
+  background?: boolean;
+  /**
+   * Run identifier whose journal and progress should seed this resumed run.
+   */
+  resumeFromRunId?: string;
+}
 /**
  * Resolved sandbox configuration.
  *
@@ -12255,10 +12304,6 @@ export interface SessionOpenOptions {
    * Denylist of tool names.
    */
   excludedTools?: string[];
-  /**
-   * Built-in subagent names to include in this session. When specified, only these built-ins are available, subject to runtime availability and exclusions. Custom agents with the same name remain available.
-   */
-  includedBuiltinAgents?: string[];
   /**
    * Built-in subagent names to exclude from this session. Excluded built-ins are hidden from agent discovery and cannot be dispatched unless a custom agent with the same name is available.
    */
@@ -13359,10 +13404,6 @@ export interface SessionUpdateOptionsParams {
    * Denylist of tool names for this session.
    */
   excludedTools?: string[];
-  /**
-   * Built-in subagent names to include in this session. When specified, only these built-ins are available, subject to runtime availability and exclusions. Custom agents with the same name remain available. Set to null to remove the allowlist restriction.
-   */
-  includedBuiltinAgents?: string[] | null;
   /**
    * Built-in subagent names to exclude from this session. Excluded built-ins are hidden from agent discovery and cannot be dispatched unless a custom agent with the same name is available.
    */
@@ -15445,6 +15486,330 @@ export interface VisibilitySetResult {
    */
   shareUrl?: string;
 }
+/**
+ * Parameters for cooperatively aborting a workflow body.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowAbortRequest".
+ */
+/** @experimental */
+export interface WorkflowAbortRequest {
+  /**
+   * Target session identifier
+   */
+  sessionId: string;
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+}
+/**
+ * Acknowledgement that a workflow request was accepted.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowAckResult".
+ */
+/** @experimental */
+export interface WorkflowAckResult {}
+/**
+ * Options for one workflow-scoped subagent call.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowAgentOptions".
+ */
+/** @experimental */
+export interface WorkflowAgentOptions {
+  /**
+   * Optional label distinguishing otherwise identical memoized agent calls.
+   */
+  label?: string;
+  /**
+   * Optional JSON Schema for structured agent output.
+   */
+  schema?: {
+    [k: string]: unknown | undefined;
+  };
+  /**
+   * Optional model identifier for the subagent.
+   */
+  model?: string;
+}
+/**
+ * Parameters for one workflow-scoped subagent call.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowAgentRequest".
+ */
+/** @experimental */
+export interface WorkflowAgentRequest {
+  /**
+   * Workflow run identifier that owns the subagent.
+   */
+  workflowRunId: string;
+  /**
+   * Prompt to send to the subagent.
+   */
+  prompt: string;
+  opts: WorkflowAgentOptions;
+}
+/**
+ * Result of one workflow-scoped subagent call.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowAgentResult".
+ */
+/** @experimental */
+export interface WorkflowAgentResult {
+  /**
+   * Agent result, omitted when the agent produced no result.
+   */
+  result?: {
+    [k: string]: unknown | undefined;
+  };
+}
+/**
+ * Parameters for cancelling a workflow run.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowCancelRequest".
+ */
+/** @experimental */
+export interface WorkflowCancelRequest {
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+}
+/**
+ * Parameters sent to the owning extension to execute a workflow closure.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowExecuteRequest".
+ */
+/** @experimental */
+export interface WorkflowExecuteRequest {
+  /**
+   * Target session identifier
+   */
+  sessionId: string;
+  /**
+   * Registered workflow name.
+   */
+  name: string;
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+  /**
+   * Workflow input value.
+   */
+  args: {
+    [k: string]: unknown | undefined;
+  };
+  /**
+   * Parent workflow run identifier for nested execution.
+   */
+  parentRunId?: string;
+}
+/**
+ * Result returned by an extension workflow closure.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowExecuteResult".
+ */
+/** @experimental */
+export interface WorkflowExecuteResult {
+  /**
+   * Workflow result value.
+   */
+  result: {
+    [k: string]: unknown | undefined;
+  };
+}
+/**
+ * Parameters for retrieving a workflow run.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowGetRunRequest".
+ */
+/** @experimental */
+export interface WorkflowGetRunRequest {
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+}
+/**
+ * Parameters for reading a workflow journal entry.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowJournalGetRequest".
+ */
+/** @experimental */
+export interface WorkflowJournalGetRequest {
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+  /**
+   * Namespaced journal key.
+   */
+  key: string;
+}
+/**
+ * Result of reading a workflow journal entry.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowJournalGetResult".
+ */
+/** @experimental */
+export interface WorkflowJournalGetResult {
+  /**
+   * Whether the journal contained the requested key.
+   */
+  hit: boolean;
+  /**
+   * Cached JSON result. The hit field distinguishes a cached JSON null from a miss.
+   */
+  resultJson?: {
+    [k: string]: unknown | undefined;
+  };
+}
+/**
+ * Parameters for storing a workflow journal entry.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowJournalPutRequest".
+ */
+/** @experimental */
+export interface WorkflowJournalPutRequest {
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+  /**
+   * Namespaced journal key.
+   */
+  key: string;
+  /**
+   * JSON result to memoize.
+   */
+  resultJson: {
+    [k: string]: unknown | undefined;
+  };
+}
+/**
+ * One ordered workflow progress line.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowLogLine".
+ */
+/** @experimental */
+export interface WorkflowLogLine {
+  /**
+   * Monotonic sequence number within the workflow run.
+   */
+  seq: number;
+  kind: WorkflowLogLineKind;
+  /**
+   * Progress text.
+   */
+  text: string;
+}
+/**
+ * Parameters for recording workflow progress.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowLogRequest".
+ */
+/** @experimental */
+export interface WorkflowLogRequest {
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+  /**
+   * Ordered progress lines to append.
+   */
+  lines: WorkflowLogLine[];
+}
+/**
+ * Parameters for invoking a nested workflow.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowRunNestedRequest".
+ */
+/** @experimental */
+export interface WorkflowRunNestedRequest {
+  /**
+   * Parent workflow run identifier.
+   */
+  parentRunId: string;
+  /**
+   * Registered child workflow name.
+   */
+  name: string;
+  /**
+   * Child workflow input value.
+   */
+  args: {
+    [k: string]: unknown | undefined;
+  };
+}
+/**
+ * Parameters for invoking a registered workflow.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowRunRequest".
+ */
+/** @experimental */
+export interface WorkflowRunRequest {
+  /**
+   * Registered workflow name.
+   */
+  name: string;
+  /**
+   * Workflow input value.
+   */
+  args: {
+    [k: string]: unknown | undefined;
+  };
+  options?: RunOptions;
+}
+/**
+ * Complete current or terminal workflow run envelope.
+ *
+ * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema
+ * via the `definition` "WorkflowRunResult".
+ */
+/** @experimental */
+export interface WorkflowRunResult {
+  /**
+   * Workflow run identifier.
+   */
+  runId: string;
+  status: WorkflowRunStatus;
+  /**
+   * Completed workflow result.
+   */
+  result?: {
+    [k: string]: unknown | undefined;
+  };
+  /**
+   * Error message for an errored run.
+   */
+  error?: string;
+  /**
+   * Reason for a halted or cancelled run.
+   */
+  reason?: string;
+  /**
+   * Partial journal and progress snapshot for a halted or cancelled run.
+   */
+  snapshot?: {
+    [k: string]: unknown | undefined;
+  };
+}
 /**
  * A single changed file and its unified diff.
  *
@@ -16555,6 +16920,84 @@ export function createSessionRpc(connection: MessageConnection, sessionId: strin
             },
         },
         /** @experimental */
+        workflow: {
+            /**
+             * Runs a registered workflow by name at the top level.
+             *
+             * @param params Parameters for invoking a registered workflow.
+             *
+             * @returns Complete current or terminal workflow run envelope.
+             */
+            run: async (params: WorkflowRunRequest): Promise<WorkflowRunResult> =>
+                connection.sendRequest("session.workflow.run", { sessionId, ...params }),
+            /**
+             * Runs a registered workflow as a child of an existing workflow run.
+             *
+             * @param params Parameters for invoking a nested workflow.
+             *
+             * @returns Complete current or terminal workflow run envelope.
+             */
+            runNested: async (params: WorkflowRunNestedRequest): Promise<WorkflowRunResult> =>
+                connection.sendRequest("session.workflow.runNested", { sessionId, ...params }),
+            /**
+             * Gets the current or settled envelope for a workflow run.
+             *
+             * @param params Parameters for retrieving a workflow run.
+             *
+             * @returns Complete current or terminal workflow run envelope.
+             */
+            getRun: async (params: WorkflowGetRunRequest): Promise<WorkflowRunResult> =>
+                connection.sendRequest("session.workflow.getRun", { sessionId, ...params }),
+            /**
+             * Requests cancellation of a workflow run and returns its run envelope.
+             *
+             * @param params Parameters for cancelling a workflow run.
+             *
+             * @returns Complete current or terminal workflow run envelope.
+             */
+            cancel: async (params: WorkflowCancelRequest): Promise<WorkflowRunResult> =>
+                connection.sendRequest("session.workflow.cancel", { sessionId, ...params }),
+            /**
+             * Records a batch of ordered workflow progress lines.
+             *
+             * @param params Parameters for recording workflow progress.
+             *
+             * @returns Acknowledgement that a workflow request was accepted.
+             */
+            log: async (params: WorkflowLogRequest): Promise<WorkflowAckResult> =>
+                connection.sendRequest("session.workflow.log", { sessionId, ...params }),
+            /**
+             * Runs one workflow-scoped subagent and returns its result.
+             *
+             * @param params Parameters for one workflow-scoped subagent call.
+             *
+             * @returns Result of one workflow-scoped subagent call.
+             */
+            agent: async (params: WorkflowAgentRequest): Promise<WorkflowAgentResult> =>
+                connection.sendRequest("session.workflow.agent", { sessionId, ...params }),
+            /** @experimental */
+            journal: {
+                /**
+                 * Reads a memoized workflow journal entry.
+                 *
+                 * @param params Parameters for reading a workflow journal entry.
+                 *
+                 * @returns Result of reading a workflow journal entry.
+                 */
+                get: async (params: WorkflowJournalGetRequest): Promise<WorkflowJournalGetResult> =>
+                    connection.sendRequest("session.workflow.journal.get", { sessionId, ...params }),
+                /**
+                 * Stores a memoized workflow journal entry.
+                 *
+                 * @param params Parameters for storing a workflow journal entry.
+                 *
+                 * @returns Acknowledgement that a workflow request was accepted.
+                 */
+                put: async (params: WorkflowJournalPutRequest): Promise<WorkflowAckResult> =>
+                    connection.sendRequest("session.workflow.journal.put", { sessionId, ...params }),
+            },
+        },
+        /** @experimental */
         model: {
             /**
              * Gets the currently selected model for the session.
@@ -18042,6 +18485,27 @@ export interface ProviderTokenHandler {
     getToken(params: ProviderTokenAcquireRequest): Promise<ProviderTokenAcquireResult>;
 }
 
+/** Handler for `workflow` client session API methods. */
+/** @experimental */
+export interface WorkflowHandler {
+    /**
+     * Asks the owning extension connection to execute a registered workflow closure.
+     *
+     * @param params Parameters sent to the owning extension to execute a workflow closure.
+     *
+     * @returns Result returned by an extension workflow closure.
+     */
+    execute(params: WorkflowExecuteRequest): Promise<WorkflowExecuteResult>;
+    /**
+     * Asks the owning extension connection to abort a running workflow cooperatively.
+     *
+     * @param params Parameters for cooperatively aborting a workflow body.
+     *
+     * @returns Acknowledgement that a workflow request was accepted.
+     */
+    abort(params: WorkflowAbortRequest): Promise<WorkflowAckResult>;
+}
+
 /** Handler for `sessionFs` client session API methods. */
 /** @experimental */
 export interface SessionFsHandler {
@@ -18173,6 +18637,7 @@ export interface CanvasHandler {
 /** All client session API handler groups. */
 export interface ClientSessionApiHandlers {
     providerToken?: ProviderTokenHandler;
+    workflow?: WorkflowHandler;
     sessionFs?: SessionFsHandler;
     canvas?: CanvasHandler;
 }
@@ -18192,6 +18657,16 @@ export function registerClientSessionApiHandlers(
         if (!handler) throw new Error(`No providerToken handler registered for session: ${params.sessionId}`);
         return handler.getToken(params);
     });
+    connection.onRequest("workflow.execute", async (params: WorkflowExecuteRequest) => {
+        const handler = getHandlers(params.sessionId).workflow;
+        if (!handler) throw new Error(`No workflow handler registered for session: ${params.sessionId}`);
+        return handler.execute(params);
+    });
+    connection.onRequest("workflow.abort", async (params: WorkflowAbortRequest) => {
+        const handler = getHandlers(params.sessionId).workflow;
+        if (!handler) throw new Error(`No workflow handler registered for session: ${params.sessionId}`);
+        return handler.abort(params);
+    });
     connection.onRequest("sessionFs.readFile", async (params: SessionFsReadFileRequest) => {
         const handler = getHandlers(params.sessionId).sessionFs;
         if (!handler) throw new Error(`No sessionFs handler registered for session: ${params.sessionId}`);
diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts
index 9d3bdcd7f0..2a4b5b06d6 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/index.ts
@@ -12,6 +12,7 @@ export { CopilotClient } from "./client.js";
 export { RuntimeConnection } from "./types.js";
 export { BuiltInTools, ToolSet } from "./toolSet.js";
 export { CopilotSession, type AssistantMessageEvent } from "./session.js";
+export { defineWorkflow, WorkflowRunError } from "./workflow.js";
 export {
     Canvas,
     CanvasError,
@@ -86,6 +87,8 @@ export type {
     LargeToolOutputConfig,
     MemoryConfiguration,
     UiInputOptions,
+    WorkflowLimits,
+    WorkflowMeta,
     MCPStdioServerConfig,
     MCPHTTPServerConfig,
     MCPServerConfig,
@@ -158,3 +161,14 @@ export type {
     TypedSessionLifecycleHandler,
     ZodSchema,
 } from "./types.js";
+export type {
+    RunOptions,
+    SessionWorkflowApi,
+    WorkflowAgentOptions,
+    WorkflowContext,
+    WorkflowDefinition,
+    WorkflowHandle,
+    WorkflowJsonSchema,
+    WorkflowPipelineStage,
+    WorkflowStepOptions,
+} from "./workflow.js";
diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts
index 1f71209de8..89f5f0de87 100644
--- a/nodejs/src/session.ts
+++ b/nodejs/src/session.ts
@@ -15,6 +15,8 @@ import type {
     CanvasActionInvokeResult,
     CurrentToolMetadata,
     McpOauthPendingRequestResponse,
+    WorkflowExecuteResult,
+    WorkflowLogLine,
 } from "./generated/rpc.js";
 import { type Canvas, CanvasError } from "./canvas.js";
 import type { OpenCanvasInstance } from "./generated/rpc.js";
@@ -60,6 +62,15 @@ import type {
     UserInputRequest,
     UserInputResponse,
 } from "./types.js";
+import {
+    getWorkflowDefinition,
+    WorkflowRunError,
+    type RunOptions,
+    type SessionWorkflowApi,
+    type WorkflowContext,
+    type WorkflowHandle,
+    type WorkflowStepOptions,
+} from "./workflow.js";
 
 /**
  * Convert a raw hook input received over the wire into its public-facing shape.
@@ -94,6 +105,157 @@ function isOpenCanvasInstance(value: unknown): value is OpenCanvasInstance {
     );
 }
 
+const WORKFLOW_LOG_FLUSH_DELAY_MS = 10;
+const MAX_WORKFLOW_FANOUT_ITEMS = 4096;
+
+function assertWorkflowFanoutSize(kind: "parallel" | "pipeline", size: number): void {
+    if (size > MAX_WORKFLOW_FANOUT_ITEMS) {
+        throw new Error(
+            `${kind}() accepts at most ${MAX_WORKFLOW_FANOUT_ITEMS} items; got ${size}.`
+        );
+    }
+}
+
+async function runWorkflowParallel<TResult>(
+    thunks: Array<() => Promise<TResult> | TResult>
+): Promise<Array<TResult | null>> {
+    if (!Array.isArray(thunks)) {
+        throw new Error(
+            "parallel() expects an array of functions, not promises. Wrap each call: () => agent(...)"
+        );
+    }
+    assertWorkflowFanoutSize("parallel", thunks.length);
+    if (thunks.some((thunk) => typeof thunk !== "function")) {
+        throw new Error(
+            "parallel() expects an array of functions, not promises. Wrap each call: () => agent(...)"
+        );
+    }
+    return Promise.all(
+        thunks.map((thunk) =>
+            Promise.resolve()
+                .then(() => thunk())
+                .catch(() => null)
+        )
+    );
+}
+
+async function runWorkflowPipeline(
+    items: unknown[],
+    ...stages: Array<
+        (previous: unknown, item: unknown, index: number) => Promise<unknown> | unknown
+    >
+): Promise<unknown[]> {
+    if (!Array.isArray(items)) {
+        throw new Error("pipeline(items, ...stages): items must be an array");
+    }
+    assertWorkflowFanoutSize("pipeline", items.length);
+    return Promise.all(
+        items.map(async (item, index) => {
+            let previous = item;
+            for (const stage of stages) {
+                try {
+                    previous = await stage(previous, item, index);
+                } catch {
+                    return null;
+                }
+            }
+            return previous;
+        })
+    );
+}
+
+class WorkflowProgressBuffer {
+    private nextSeq = 0;
+    private pending: WorkflowLogLine[] = [];
+    private flushTimer?: ReturnType<typeof setTimeout>;
+    private flushTail: Promise<void> = Promise.resolve();
+    private flushError: unknown;
+    private flushFailed = false;
+    private closed = false;
+
+    constructor(private readonly send: (lines: WorkflowLogLine[]) => Promise<void>) {}
+
+    enqueue(kind: WorkflowLogLine["kind"], text: string): void {
+        if (this.closed) {
+            throw new Error("Cannot log after the workflow run has settled");
+        }
+
+        this.pending.push({ seq: this.nextSeq++, kind, text });
+        this.scheduleFlush();
+    }
+
+    async flush(): Promise<void> {
+        this.clearFlushTimer();
+        const lines = this.pending.splice(0);
+        if (lines.length > 0) {
+            this.flushTail = this.flushTail.then(async () => {
+                try {
+                    await this.send(lines);
+                } catch (error) {
+                    if (!this.flushFailed) {
+                        this.flushFailed = true;
+                        this.flushError = error;
+                    }
+                }
+            });
+        }
+        await this.flushTail;
+        if (this.flushFailed) {
+            throw this.flushError;
+        }
+    }
+
+    async close(): Promise<void> {
+        this.closed = true;
+        await this.flush();
+    }
+
+    private scheduleFlush(): void {
+        if (this.flushTimer !== undefined) {
+            return;
+        }
+        this.flushTimer = setTimeout(() => {
+            this.flushTimer = undefined;
+            void this.flush().catch(() => {});
+        }, WORKFLOW_LOG_FLUSH_DELAY_MS);
+        this.flushTimer.unref?.();
+    }
+
+    private clearFlushTimer(): void {
+        if (this.flushTimer !== undefined) {
+            clearTimeout(this.flushTimer);
+            this.flushTimer = undefined;
+        }
+    }
+}
+
+async function awaitWorkflowOperation<TResult>(
+    operation: Promise<TResult>,
+    signal: AbortSignal
+): Promise<TResult> {
+    throwIfWorkflowAborted(signal);
+    let rejectAbort: ((reason?: unknown) => void) | undefined;
+    const onAbort = () =>
+        rejectAbort?.(signal.reason ?? new DOMException("Workflow run was aborted", "AbortError"));
+    try {
+        return await Promise.race([
+            operation,
+            new Promise<never>((_resolve, reject) => {
+                rejectAbort = reject;
+                signal.addEventListener("abort", onAbort, { once: true });
+            }),
+        ]);
+    } finally {
+        signal.removeEventListener("abort", onAbort);
+    }
+}
+
+function throwIfWorkflowAborted(signal: AbortSignal): void {
+    if (signal.aborted) {
+        throw signal.reason ?? new DOMException("Workflow run was aborted", "AbortError");
+    }
+}
+
 /** Assistant message event - the final response from the assistant. */
 export type AssistantMessageEvent = Extract<SessionEvent, { type: "assistant.message" }>;
 
@@ -138,6 +300,8 @@ export class CopilotSession {
     private canvases: Map<string, Canvas> = new Map();
     private bearerTokenProviders: Map<string, BearerTokenProvider> = new Map();
     private commandHandlers: Map<string, CommandHandler> = new Map();
+    private workflows = new Map<string, ReturnType<typeof getWorkflowDefinition>>();
+    private workflowAbortControllers = new Map<string, AbortController>();
     private permissionHandler?: PermissionHandler;
     private mcpAuthHandler?: McpAuthHandler;
     private userInputHandler?: UserInputHandler;
@@ -155,6 +319,38 @@ export class CopilotSession {
     /** @internal Client session API handlers, populated by CopilotClient during create/resume. */
     clientSessionApis: ClientSessionApiHandlers = {};
 
+    readonly workflow: SessionWorkflowApi = {
+        run: (async (
+            nameOrHandle: string | WorkflowHandle,
+            options?: RunOptions
+        ): Promise<unknown> => {
+            const name =
+                typeof nameOrHandle === "string"
+                    ? nameOrHandle
+                    : getWorkflowDefinition(nameOrHandle).meta.name;
+            const envelope = await this.rpc.workflow.run({
+                name,
+                args: (options?.args === undefined ? {} : options.args) as Parameters<
+                    typeof this.rpc.workflow.run
+                >[0]["args"],
+                options: {
+                    background: options?.background,
+                    resumeFromRunId: options?.resumeFromRunId,
+                },
+            });
+
+            if (options?.background) {
+                return envelope;
+            }
+            if (envelope.status !== "completed") {
+                throw new WorkflowRunError(envelope);
+            }
+            return envelope.result;
+        }) as SessionWorkflowApi["run"],
+        getRun: (runId) => this.rpc.workflow.getRun({ runId }),
+        cancel: (runId) => this.rpc.workflow.cancel({ runId }),
+    };
+
     /**
      * Creates a new CopilotSession instance.
      *
@@ -358,6 +554,11 @@ export class CopilotSession {
         this.autoModeSwitchHandler = undefined;
         this.commandHandlers.clear();
         this.canvases.clear();
+        this.workflows.clear();
+        for (const controller of this.workflowAbortControllers.values()) {
+            controller.abort();
+        }
+        this.workflowAbortControllers.clear();
         this.transformCallbacks?.clear();
     }
 
@@ -872,6 +1073,138 @@ export class CopilotSession {
         };
     }
 
+    /**
+     * Registers workflow closures and reverse-RPC handlers for this session.
+     *
+     * @param workflows - Workflow handles declared by the joining extension.
+     * @internal Called by the SDK when an extension joins a session.
+     */
+    registerWorkflows(workflows?: WorkflowHandle[]): void {
+        this.workflows.clear();
+        if (!workflows || workflows.length === 0) {
+            delete this.clientSessionApis.workflow;
+            return;
+        }
+
+        for (const handle of workflows) {
+            const definition = getWorkflowDefinition(handle);
+            this.workflows.set(definition.meta.name, definition);
+        }
+
+        const self = this;
+        this.clientSessionApis.workflow = {
+            async execute(params) {
+                const definition = self.workflows.get(params.name);
+                if (!definition) {
+                    const message = `No workflow registered with name "${params.name}"`;
+                    throw new ResponseError(ErrorCodes.InvalidParams, message, {
+                        code: "workflow_not_found",
+                        name: params.name,
+                    });
+                }
+
+                const controller = new AbortController();
+                self.workflowAbortControllers.set(params.runId, controller);
+                const progress = new WorkflowProgressBuffer(async (lines) => {
+                    await self.rpc.workflow.log({ runId: params.runId, lines });
+                });
+                try {
+                    const context: WorkflowContext = {
+                        args: params.args,
+                        session: self,
+                        signal: controller.signal,
+                        phase: (title: string) => {
+                            throwIfWorkflowAborted(controller.signal);
+                            progress.enqueue("phase", title);
+                        },
+                        log: (message: string) => {
+                            throwIfWorkflowAborted(controller.signal);
+                            progress.enqueue("log", message);
+                        },
+                        agent: async (prompt, options = {}) => {
+                            await progress.flush();
+                            const response = await awaitWorkflowOperation(
+                                self.rpc.workflow.agent({
+                                    workflowRunId: params.runId,
+                                    prompt,
+                                    opts: {
+                                        label: options.label,
+                                        schema: options.schema,
+                                        model: options.model,
+                                    },
+                                }),
+                                controller.signal
+                            );
+                            return response.result ?? null;
+                        },
+                        step: async <TResult>(
+                            key: string,
+                            producer: () => Promise<TResult> | TResult,
+                            options: WorkflowStepOptions = {}
+                        ): Promise<TResult> => {
+                            await progress.flush();
+                            if (options.volatile) {
+                                return producer();
+                            }
+                            const cached = await awaitWorkflowOperation(
+                                self.rpc.workflow.journal.get({
+                                    runId: params.runId,
+                                    key,
+                                }),
+                                controller.signal
+                            );
+                            if (cached.hit) {
+                                return cached.resultJson as TResult;
+                            }
+
+                            // Producers are best-effort at-least-once across crashes or
+                            // concurrent callers, so authors must make side effects idempotent.
+                            const result = await producer();
+                            const resultJson = JSON.stringify(result);
+                            if (resultJson === undefined) {
+                                throw new Error(
+                                    `step("${key}") returned a value that is not JSON-serializable`
+                                );
+                            }
+                            await awaitWorkflowOperation(
+                                self.rpc.workflow.journal.put({
+                                    runId: params.runId,
+                                    key,
+                                    resultJson: result as Parameters<
+                                        typeof self.rpc.workflow.journal.put
+                                    >[0]["resultJson"],
+                                }),
+                                controller.signal
+                            );
+                            return result;
+                        },
+                        parallel: runWorkflowParallel,
+                        pipeline: runWorkflowPipeline,
+                        workflow: async () => {
+                            throw new Error("nested workflows are not supported");
+                        },
+                    };
+                    const result = await definition.run(context);
+                    return { result } as WorkflowExecuteResult;
+                } finally {
+                    try {
+                        await progress.close();
+                    } finally {
+                        if (self.workflowAbortControllers.get(params.runId) === controller) {
+                            self.workflowAbortControllers.delete(params.runId);
+                        }
+                    }
+                }
+            },
+            async abort(params) {
+                self.workflowAbortControllers
+                    .get(params.runId)
+                    ?.abort(new DOMException("Workflow run was aborted", "AbortError"));
+                return {};
+            },
+        };
+    }
+
     /**
      * Registers per-provider {@link BearerTokenProvider} callbacks for BYOK providers
      * configured with managed-identity / on-demand bearer-token auth.
diff --git a/nodejs/src/types.ts b/nodejs/src/types.ts
index 12ab0860b0..f62ee77920 100644
--- a/nodejs/src/types.ts
+++ b/nodejs/src/types.ts
@@ -1821,6 +1821,28 @@ export interface CanvasProviderIdentity {
     name?: string;
 }
 
+/** Static resource ceilings declared by a workflow before it runs. */
+export interface WorkflowLimits {
+    /** Maximum number of workflow subagents that may run concurrently. Must be positive when present. */
+    maxConcurrentSubagents?: number;
+    /** Maximum total number of workflow subagents that may be spawned. Must be positive when present. */
+    maxTotalSubagents?: number;
+    /** Wall-clock timeout for the workflow run, in milliseconds. Must be positive when present. */
+    timeout?: number;
+}
+
+/** Registration metadata for an extension-authored workflow. */
+export interface WorkflowMeta {
+    /** Stable workflow name used for invocation. */
+    name: string;
+    /** Human-readable workflow description. */
+    description: string;
+    /** Display metadata for the progress phases the workflow may report. */
+    phases: Array<{ title: string; detail?: string }>;
+    /** Optional resource ceilings presented to the user before execution. */
+    limits?: WorkflowLimits;
+}
+
 /**
  * Provider-scoped options for the Copilot API (CAPI).
  *
diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts
new file mode 100644
index 0000000000..0306a0d4e1
--- /dev/null
+++ b/nodejs/src/workflow.ts
@@ -0,0 +1,185 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------------------------------------------*/
+
+import type { WorkflowRunResult } from "./generated/rpc.js";
+import type { CopilotSession } from "./session.js";
+import type { WorkflowMeta } from "./types.js";
+
+declare const workflowHandleBrand: unique symbol;
+
+/**
+ * Conservative JSON shape language accepted for structured workflow agent output.
+ *
+ * Supports `type`, `required`, `enum`, `const`, recursive `properties`/`items`,
+ * and `anyOf`/`oneOf`/`allOf`. Other JSON Schema keywords are ignored.
+ */
+export type WorkflowJsonSchema = Record<string, unknown>;
+
+/** Options for one workflow-scoped subagent call. */
+export interface WorkflowAgentOptions {
+    label?: string;
+    schema?: WorkflowJsonSchema;
+    model?: string;
+}
+
+/** Options for a durable workflow step. */
+export interface WorkflowStepOptions {
+    /** Skip the journal and always invoke the producer. */
+    volatile?: boolean;
+}
+
+/** One stage in a per-item workflow pipeline. */
+export type WorkflowPipelineStage<TInput = unknown, TResult = unknown> = (
+    previous: TInput,
+    item: unknown,
+    index: number
+) => Promise<TResult> | TResult;
+
+/** Context passed to an extension-authored workflow body. */
+export interface WorkflowContext<TArgs = unknown> {
+    /** Spawn and await one workflow-scoped subagent. */
+    agent(prompt: string, options?: WorkflowAgentOptions): Promise<unknown>;
+    /** Memoize an arbitrary producer under a stable author-supplied key. */
+    step<TResult>(
+        key: string,
+        producer: () => Promise<TResult> | TResult,
+        options?: WorkflowStepOptions
+    ): Promise<TResult>;
+    /** Run thunks concurrently, returning null for a thunk that throws. */
+    parallel<TResult>(
+        thunks: Array<() => Promise<TResult> | TResult>
+    ): Promise<Array<TResult | null>>;
+    /** Run each item through every stage without barriers between stages. */
+    pipeline(items: unknown[], ...stages: WorkflowPipelineStage[]): Promise<unknown[]>;
+    /** Start a named workflow progress phase. */
+    phase(title: string): void;
+    /** Emit a workflow progress line. */
+    log(message: string): void;
+    /** Reject because nested workflows are not supported. */
+    workflow(name: string, args?: unknown): Promise<unknown>;
+    /** Caller-supplied input, forwarded verbatim. */
+    args: TArgs;
+    /** The same full session instance returned by `joinSession`. */
+    session: CopilotSession;
+    /** Cooperative cancellation signal for the current workflow run. */
+    signal: AbortSignal;
+}
+
+/** Definition accepted by {@link defineWorkflow}. */
+export interface WorkflowDefinition<TArgs = unknown, TResult = unknown> {
+    meta: WorkflowMeta;
+    run(context: WorkflowContext<TArgs>): Promise<TResult>;
+}
+
+/** Opaque reusable reference to a defined workflow. */
+export interface WorkflowHandle<TArgs = unknown, TResult = unknown> {
+    readonly meta: WorkflowMeta;
+    readonly [workflowHandleBrand]: {
+        readonly args: TArgs;
+        readonly result: TResult;
+    };
+}
+
+/** Options for invoking a workflow. */
+export interface RunOptions<TArgs = unknown> {
+    /** Input surfaced as `context.args`. */
+    args?: TArgs;
+    /** Return once the approved run starts instead of awaiting completion. */
+    background?: boolean;
+    /** Prior run whose journal and progress should seed this run. */
+    resumeFromRunId?: string;
+}
+
+/** Friendly workflow API exposed on a session. */
+export interface SessionWorkflowApi {
+    run(name: string, options: RunOptions & { background: true }): Promise<WorkflowRunResult>;
+    run<TResult = unknown>(
+        name: string,
+        options?: RunOptions & { background?: false }
+    ): Promise<TResult>;
+    run<TResult = unknown>(
+        name: string,
+        options?: RunOptions
+    ): Promise<TResult | WorkflowRunResult>;
+    run<TArgs, TResult>(
+        workflow: WorkflowHandle<TArgs, TResult>,
+        options: RunOptions<TArgs> & { background: true }
+    ): Promise<WorkflowRunResult>;
+    run<TArgs, TResult>(
+        workflow: WorkflowHandle<TArgs, TResult>,
+        options?: RunOptions<TArgs> & { background?: false }
+    ): Promise<TResult>;
+    run<TArgs, TResult>(
+        workflow: WorkflowHandle<TArgs, TResult>,
+        options?: RunOptions<TArgs>
+    ): Promise<TResult | WorkflowRunResult>;
+    /** Read the latest durable envelope for a workflow run. */
+    getRun(runId: string): Promise<WorkflowRunResult>;
+    /** Cancel a workflow run and return its terminal envelope. */
+    cancel(runId: string): Promise<WorkflowRunResult>;
+}
+
+/** Error thrown when a foreground workflow run does not complete successfully. */
+export class WorkflowRunError extends Error {
+    constructor(public readonly envelope: WorkflowRunResult) {
+        super(
+            envelope.error ??
+                envelope.reason ??
+                `Workflow run "${envelope.runId}" ended with status "${envelope.status}"`
+        );
+        this.name = "WorkflowRunError";
+    }
+}
+
+interface StoredWorkflow {
+    meta: WorkflowMeta;
+    run(context: WorkflowContext<unknown>): Promise<unknown>;
+}
+
+const workflowHandles = new WeakMap<object, StoredWorkflow>();
+
+function validateLimits(meta: WorkflowMeta): void {
+    const limits = meta.limits;
+    if (!limits) {
+        return;
+    }
+
+    for (const field of ["maxConcurrentSubagents", "maxTotalSubagents"] as const) {
+        const value = limits[field];
+        if (value !== undefined && (!Number.isInteger(value) || value <= 0)) {
+            throw new Error(`Workflow limit "${field}" must be a positive integer`);
+        }
+    }
+
+    if (limits.timeout !== undefined && (!Number.isFinite(limits.timeout) || limits.timeout <= 0)) {
+        throw new Error('Workflow limit "timeout" must be a positive number of milliseconds');
+    }
+}
+
+/**
+ * Defines an extension-authored workflow and returns an opaque registration handle.
+ */
+export function defineWorkflow<TArgs = unknown, TResult = unknown>(
+    definition: WorkflowDefinition<TArgs, TResult>
+): WorkflowHandle<TArgs, TResult> {
+    validateLimits(definition.meta);
+
+    const stored: StoredWorkflow = {
+        meta: definition.meta,
+        run: definition.run as StoredWorkflow["run"],
+    };
+    const handle = Object.freeze({ meta: definition.meta }) as WorkflowHandle<TArgs, TResult>;
+
+    workflowHandles.set(handle, stored);
+    return handle;
+}
+
+/** @internal */
+export function getWorkflowDefinition(handle: WorkflowHandle): StoredWorkflow {
+    const definition = workflowHandles.get(handle);
+    if (!definition) {
+        throw new Error("Invalid workflow handle");
+    }
+    return definition;
+}
diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts
new file mode 100644
index 0000000000..a602d2eef3
--- /dev/null
+++ b/nodejs/test/workflow.test.ts
@@ -0,0 +1,844 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------------------------------------------*/
+
+import { afterEach, describe, expect, it, onTestFinished, vi } from "vitest";
+import { ResponseError } from "vscode-jsonrpc/node.js";
+import { CopilotClient } from "../src/client.js";
+import { joinSession } from "../src/extension.js";
+import { CopilotSession } from "../src/session.js";
+import {
+    defineWorkflow,
+    WorkflowRunError,
+    type WorkflowAgentOptions,
+    type WorkflowDefinition,
+} from "../src/workflow.js";
+
+async function stopClient(client: CopilotClient): Promise<void> {
+    await client.stop();
+}
+
+describe("workflows", () => {
+    const originalSessionId = process.env.SESSION_ID;
+
+    afterEach(() => {
+        if (originalSessionId === undefined) {
+            delete process.env.SESSION_ID;
+        } else {
+            process.env.SESSION_ID = originalSessionId;
+        }
+        vi.restoreAllMocks();
+    });
+
+    it("defines a stable handle and accepts omitted limits", async () => {
+        const meta = {
+            name: "no-limits",
+            description: "A workflow without resource limits",
+            phases: [],
+        };
+        const run = vi.fn(async ({ args }: { args: unknown }) => args);
+        const handle = defineWorkflow({ meta, run });
+
+        expect(handle.meta).toBe(meta);
+        expect(Object.isFrozen(handle)).toBe(true);
+
+        const session = new CopilotSession("session-1", {} as never);
+        session.registerWorkflows([handle]);
+        const result = await session.clientSessionApis.workflow!.execute({
+            sessionId: session.sessionId,
+            name: meta.name,
+            runId: "run-1",
+            args: { value: 42 },
+        });
+
+        expect(run).toHaveBeenCalledOnce();
+        expect(result).toEqual({ result: { value: 42 } });
+    });
+
+    it.each([
+        ["maxConcurrentSubagents", 0],
+        ["maxConcurrentSubagents", 1.5],
+        ["maxTotalSubagents", -1],
+        ["maxTotalSubagents", Number.POSITIVE_INFINITY],
+        ["timeout", 0],
+        ["timeout", Number.NaN],
+    ] as const)("rejects invalid %s limit %s", (field, value) => {
+        const definition = {
+            meta: {
+                name: `invalid-${field}-${String(value)}`,
+                description: "Invalid workflow",
+                phases: [],
+                limits: { [field]: value },
+            },
+            run: async () => null,
+        } as WorkflowDefinition;
+
+        expect(() => defineWorkflow(definition)).toThrow(/must be a positive/);
+    });
+
+    it("serializes only workflow metadata in the extension resume payload", async () => {
+        const client = new CopilotClient();
+        await client.start();
+        onTestFinished(() => stopClient(client));
+
+        const run = vi.fn(async () => ({ ok: true }));
+        const workflow = defineWorkflow({
+            meta: {
+                name: "registered",
+                description: "Registration test",
+                phases: [{ title: "Run" }],
+                limits: { maxTotalSubagents: 2 },
+            },
+            run,
+        });
+        const sendRequest = vi
+            .spyOn(
+                (client as never as { connection: { sendRequest: Function } }).connection,
+                "sendRequest"
+            )
+            .mockImplementation(async (method: string, params: Record<string, unknown>) => {
+                if (method === "session.resume") {
+                    const sessions = (client as never as { sessions: Map<string, CopilotSession> })
+                        .sessions;
+                    expect(
+                        sessions.get(params.sessionId as string)?.clientSessionApis.workflow
+                    ).toBeDefined();
+                    return { sessionId: params.sessionId };
+                }
+                throw new Error(`Unexpected method: ${method}`);
+            });
+
+        await client.resumeSessionForExtension(
+            "session-registration",
+            { onPermissionRequest: () => ({ kind: "approved" }) },
+            [workflow]
+        );
+
+        const payload = sendRequest.mock.calls.find(
+            ([method]) => method === "session.resume"
+        )![1] as {
+            workflows: unknown[];
+        };
+        expect(payload.workflows).toEqual([workflow.meta]);
+        expect(payload.workflows[0]).not.toHaveProperty("run");
+        expect(JSON.stringify(payload.workflows)).not.toContain("async");
+    });
+
+    it("passes workflows only through the extension join path", async () => {
+        process.env.SESSION_ID = "session-extension";
+        const workflow = defineWorkflow({
+            meta: {
+                name: "extension-only",
+                description: "Extension-only registration",
+                phases: [],
+            },
+            run: async () => ({ ok: true }),
+        });
+        const resumeSessionForExtension = vi
+            .spyOn(CopilotClient.prototype, "resumeSessionForExtension")
+            .mockResolvedValue({} as CopilotSession);
+
+        await joinSession({ workflows: [workflow] });
+
+        expect(resumeSessionForExtension).toHaveBeenCalledWith(
+            "session-extension",
+            expect.objectContaining({ suppressResumeEvent: true }),
+            [workflow]
+        );
+    });
+
+    it("builds the workflow context with the unrestricted joined session identity", async () => {
+        process.env.SESSION_ID = "session-context";
+        const sendRequest = vi.fn(async (method: string) => {
+            if (method === "session.workflow.log") {
+                return {};
+            }
+            if (method === "session.tasks.list") {
+                return { tasks: [] };
+            }
+            throw new Error(`Unexpected method: ${method}`);
+        });
+        const joinedSession = new CopilotSession("session-context", { sendRequest } as never);
+        const contextSeen = Promise.withResolvers<{
+            args: unknown;
+            session: CopilotSession;
+            signal: AbortSignal;
+        }>();
+        const workflow = defineWorkflow({
+            meta: {
+                name: "context",
+                description: "Context test",
+                phases: [],
+            },
+            run: async (context) => {
+                contextSeen.resolve(context);
+                context.phase("A");
+                context.log("hi");
+                const tasks = await context.session.rpc.tasks.list();
+                return { ok: true, taskCount: tasks.tasks.length };
+            },
+        });
+        vi.spyOn(CopilotClient.prototype, "resumeSessionForExtension").mockImplementation(
+            async (_sessionId, _config, workflows) => {
+                joinedSession.registerWorkflows(workflows);
+                return joinedSession;
+            }
+        );
+
+        const joinSessionResult = await joinSession({ workflows: [workflow] });
+        const executeResult = await joinSessionResult.clientSessionApis.workflow!.execute({
+            sessionId: joinSessionResult.sessionId,
+            name: "context",
+            runId: "run-context",
+            args: { value: 42 },
+        });
+        const context = await contextSeen.promise;
+
+        expect(context.args).toEqual({ value: 42 });
+        expect(context.session).toBe(joinSessionResult);
+        expect(context.session.rpc).toBe(joinSessionResult.rpc);
+        expect(context.signal).toBeInstanceOf(AbortSignal);
+        expect(executeResult).toEqual({ result: { ok: true, taskCount: 0 } });
+        expect(sendRequest).toHaveBeenCalledWith("session.tasks.list", {
+            sessionId: joinSessionResult.sessionId,
+        });
+        expect(sendRequest).toHaveBeenCalledWith("session.workflow.log", {
+            sessionId: joinSessionResult.sessionId,
+            runId: "run-context",
+            lines: [
+                { seq: 0, kind: "phase", text: "A" },
+                { seq: 1, kind: "log", text: "hi" },
+            ],
+        });
+    });
+
+    it("rejects nested workflows without forwarding a runNested request", async () => {
+        const sendRequest = vi.fn(async () => {
+            throw new Error("Unexpected forward request");
+        });
+        const session = new CopilotSession("session-no-nesting", { sendRequest } as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "no-nesting",
+                description: "Nested workflow rejection test",
+                phases: [],
+            },
+            run: async (context) => context.workflow("nested", { value: 42 }),
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "no-nesting",
+                runId: "run-no-nesting",
+                args: {},
+            })
+        ).rejects.toThrow("nested workflows are not supported");
+        expect(sendRequest).not.toHaveBeenCalled();
+    });
+
+    it("flushes progress incrementally while a workflow body is awaiting", async () => {
+        const sendRequest = vi.fn(async () => ({}));
+        const session = new CopilotSession("session-live-progress", { sendRequest } as never);
+        const body = Promise.withResolvers<void>();
+        const workflow = defineWorkflow({
+            meta: {
+                name: "live-progress",
+                description: "Incremental progress test",
+                phases: [],
+            },
+            run: async ({ log }) => {
+                log("before await");
+                await body.promise;
+                return "done";
+            },
+        });
+        session.registerWorkflows([workflow]);
+
+        const execution = session.clientSessionApis.workflow!.execute({
+            sessionId: session.sessionId,
+            name: "live-progress",
+            runId: "run-live-progress",
+            args: {},
+        });
+        await vi.waitFor(() => {
+            expect(sendRequest).toHaveBeenCalledWith("session.workflow.log", {
+                sessionId: session.sessionId,
+                runId: "run-live-progress",
+                lines: [{ seq: 0, kind: "log", text: "before await" }],
+            });
+        });
+
+        body.resolve();
+        await expect(execution).resolves.toEqual({ result: "done" });
+    });
+
+    it("calls workflow.agent with the current run id and returns its text", async () => {
+        const sendRequest = vi.fn(async (method: string) => {
+            if (method === "session.workflow.agent") {
+                return { result: "pong" };
+            }
+            throw new Error(`Unexpected method: ${method}`);
+        });
+        const session = new CopilotSession("session-agent", { sendRequest } as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "agent",
+                description: "Agent context test",
+                phases: [],
+            },
+            run: async ({ agent }) =>
+                agent("Reply with pong", {
+                    label: "Pong helper",
+                    model: "gpt-test",
+                    schema: { type: "string" },
+                    effort: "high",
+                } as WorkflowAgentOptions),
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "agent",
+                runId: "run-agent",
+                args: {},
+            })
+        ).resolves.toEqual({ result: "pong" });
+        expect(sendRequest).toHaveBeenCalledWith("session.workflow.agent", {
+            sessionId: session.sessionId,
+            workflowRunId: "run-agent",
+            prompt: "Reply with pong",
+            opts: {
+                label: "Pong helper",
+                model: "gpt-test",
+                schema: { type: "string" },
+            },
+        });
+    });
+
+    it("runs a durable step once, serves cached null, and does not cache failures", async () => {
+        const journal = new Map<string, unknown>();
+        const sendRequest = vi.fn(
+            async (method: string, params: { key?: string; resultJson?: unknown }) => {
+                if (method === "session.workflow.journal.get") {
+                    return journal.has(params.key!)
+                        ? { hit: true, resultJson: journal.get(params.key!) }
+                        : { hit: false };
+                }
+                if (method === "session.workflow.journal.put") {
+                    journal.set(params.key!, params.resultJson);
+                    return {};
+                }
+                throw new Error(`Unexpected method: ${method}`);
+            }
+        );
+        const session = new CopilotSession("session-step", { sendRequest } as never);
+        let cachedProducerCalls = 0;
+        let failingProducerCalls = 0;
+        const workflow = defineWorkflow({
+            meta: {
+                name: "step",
+                description: "Durable step context test",
+                phases: [],
+            },
+            run: async ({ step }) => {
+                const first = await step("cached-null", async () => {
+                    cachedProducerCalls++;
+                    return null;
+                });
+                const second = await step("cached-null", async () => {
+                    cachedProducerCalls++;
+                    return "wrong";
+                });
+                const failed = await step("retry", async () => {
+                    failingProducerCalls++;
+                    throw new Error("transient");
+                }).catch(() => "failed");
+                const retried = await step("retry", async () => {
+                    failingProducerCalls++;
+                    return "recovered";
+                });
+                return { first, second, failed, retried };
+            },
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "step",
+                runId: "run-step",
+                args: {},
+            })
+        ).resolves.toEqual({
+            result: { first: null, second: null, failed: "failed", retried: "recovered" },
+        });
+        expect(cachedProducerCalls).toBe(1);
+        expect(failingProducerCalls).toBe(2);
+        expect(
+            sendRequest.mock.calls.filter(([method]) => method === "session.workflow.journal.put")
+        ).toHaveLength(2);
+    });
+
+    it("exposes workflow getRun and forwards the run id", async () => {
+        const envelope = { runId: "run-read", status: "error", error: "failed" };
+        const sendRequest = vi.fn(async () => envelope);
+        const session = new CopilotSession("session-read", { sendRequest } as never);
+
+        await expect(session.workflow.getRun("run-read")).resolves.toEqual(envelope);
+        expect(sendRequest).toHaveBeenCalledWith("session.workflow.getRun", {
+            sessionId: session.sessionId,
+            runId: "run-read",
+        });
+    });
+
+    it("exposes workflow cancel and forwards the run id", async () => {
+        const envelope = { runId: "run-cancel", status: "cancelled", reason: "cancelled" };
+        const sendRequest = vi.fn(async () => envelope);
+        const session = new CopilotSession("session-cancel", { sendRequest } as never);
+
+        await expect(session.workflow.cancel("run-cancel")).resolves.toEqual(envelope);
+        expect(sendRequest).toHaveBeenCalledWith("session.workflow.cancel", {
+            sessionId: session.sessionId,
+            runId: "run-cancel",
+        });
+    });
+
+    it("runs parallel as a barrier and maps a throwing thunk to null", async () => {
+        const first = Promise.withResolvers<string>();
+        const second = Promise.withResolvers<string>();
+        const started: string[] = [];
+        const session = new CopilotSession("session-parallel", {} as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "parallel",
+                description: "Parallel combinator test",
+                phases: [],
+            },
+            run: async ({ parallel }) =>
+                parallel([
+                    async () => {
+                        started.push("first");
+                        return first.promise;
+                    },
+                    async () => {
+                        started.push("second");
+                        return second.promise;
+                    },
+                    async () => {
+                        started.push("throwing");
+                        throw new Error("expected");
+                    },
+                ]),
+        });
+        session.registerWorkflows([workflow]);
+
+        let settled = false;
+        const execution = session.clientSessionApis
+            .workflow!.execute({
+                sessionId: session.sessionId,
+                name: "parallel",
+                runId: "run-parallel",
+                args: {},
+            })
+            .finally(() => {
+                settled = true;
+            });
+        await vi.waitFor(() => expect(started).toEqual(["first", "second", "throwing"]));
+
+        second.resolve("second");
+        await Promise.resolve();
+        expect(settled).toBe(false);
+
+        first.resolve("first");
+        await expect(execution).resolves.toEqual({ result: ["first", "second", null] });
+    });
+
+    it("rejects already-invoked promises passed to parallel with a clear diagnostic", async () => {
+        const session = new CopilotSession("session-parallel-promises", {} as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "parallel-promises",
+                description: "Parallel misuse diagnostic",
+                phases: [],
+            },
+            run: async ({ parallel }) =>
+                parallel([Promise.resolve("already running")] as unknown as Array<
+                    () => Promise<string>
+                >),
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "parallel-promises",
+                runId: "run-parallel-promises",
+                args: {},
+            })
+        ).rejects.toThrow(
+            "parallel() expects an array of functions, not promises. Wrap each call: () => agent(...)"
+        );
+    });
+
+    it("flows pipeline items independently and drops only the item whose stage throws", async () => {
+        const releaseFirstItem = Promise.withResolvers<void>();
+        const secondStageStarted = Promise.withResolvers<void>();
+        const finalStageItems: string[] = [];
+        const session = new CopilotSession("session-pipeline", {} as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "pipeline",
+                description: "Pipeline combinator test",
+                phases: [],
+            },
+            run: async ({ pipeline }) =>
+                pipeline(
+                    ["slow", "fast", "throw"],
+                    async (_previous, item) => {
+                        if (item === "slow") {
+                            await releaseFirstItem.promise;
+                        }
+                        if (item === "throw") {
+                            throw new Error("expected");
+                        }
+                        return `${item}-stage-1`;
+                    },
+                    async (previous, item) => {
+                        if (item === "fast") {
+                            secondStageStarted.resolve();
+                        }
+                        finalStageItems.push(item as string);
+                        return `${previous}-stage-2`;
+                    }
+                ),
+        });
+        session.registerWorkflows([workflow]);
+
+        const execution = session.clientSessionApis.workflow!.execute({
+            sessionId: session.sessionId,
+            name: "pipeline",
+            runId: "run-pipeline",
+            args: {},
+        });
+        await secondStageStarted.promise;
+        expect(finalStageItems).toEqual(["fast"]);
+
+        releaseFirstItem.resolve();
+        await expect(execution).resolves.toEqual({
+            result: ["slow-stage-1-stage-2", "fast-stage-1-stage-2", null],
+        });
+        expect(finalStageItems).toEqual(["fast", "slow"]);
+    });
+
+    it("enforces the 4096-item cap for parallel and pipeline", async () => {
+        const session = new CopilotSession("session-fanout-cap", {} as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "fanout-cap",
+                description: "Fan-out cap test",
+                phases: [],
+            },
+            run: async ({ parallel, pipeline }) => {
+                const tooManyItems = Array.from({ length: 4097 }, () => null);
+                const parallelError = await parallel(
+                    tooManyItems.map(() => async () => null)
+                ).catch((error: unknown) => error);
+                const pipelineError = await pipeline(tooManyItems).catch((error: unknown) => error);
+                return {
+                    parallel: (parallelError as Error).message,
+                    pipeline: (pipelineError as Error).message,
+                };
+            },
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "fanout-cap",
+                runId: "run-fanout-cap",
+                args: {},
+            })
+        ).resolves.toEqual({
+            result: {
+                parallel: "parallel() accepts at most 4096 items; got 4097.",
+                pipeline: "pipeline() accepts at most 4096 items; got 4097.",
+            },
+        });
+    });
+
+    it("does not deadlock nested combinators when only leaf agents use a one-slot limiter", async () => {
+        let active = 0;
+        let maxActive = 0;
+        let tail = Promise.resolve();
+        const sendRequest = vi.fn(
+            async (method: string, params: { prompt: string }): Promise<{ result: string }> => {
+                if (method !== "session.workflow.agent") {
+                    throw new Error(`Unexpected method: ${method}`);
+                }
+                const previous = tail;
+                const done = Promise.withResolvers<void>();
+                tail = done.promise;
+                await previous;
+                active++;
+                maxActive = Math.max(maxActive, active);
+                await Promise.resolve();
+                active--;
+                done.resolve();
+                return { result: params.prompt };
+            }
+        );
+        const session = new CopilotSession("session-nested-combinators", {
+            sendRequest,
+        } as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "nested-combinators",
+                description: "Nested combinator deadlock regression",
+                phases: [],
+            },
+            run: async ({ agent, parallel, pipeline }) =>
+                parallel([
+                    () => parallel([() => agent("a"), () => agent("b")]),
+                    () => pipeline(["c"], (_previous, item) => agent(item as string)),
+                ]),
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "nested-combinators",
+                runId: "run-nested-combinators",
+                args: {},
+            })
+        ).resolves.toEqual({ result: [["a", "b"], ["c"]] });
+        expect(maxActive).toBe(1);
+        expect(sendRequest).toHaveBeenCalledTimes(3);
+    });
+
+    it("flushes buffered progress in finally when the workflow body throws", async () => {
+        const sendRequest = vi.fn(async () => ({}));
+        const session = new CopilotSession("session-throw-progress", { sendRequest } as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "throw-progress",
+                description: "Throwing progress test",
+                phases: [],
+            },
+            run: async ({ log }) => {
+                log("before throw");
+                throw new Error("body failed");
+            },
+        });
+        session.registerWorkflows([workflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "throw-progress",
+                runId: "run-throw-progress",
+                args: {},
+            })
+        ).rejects.toThrow("body failed");
+        expect(sendRequest).toHaveBeenCalledWith("session.workflow.log", {
+            sessionId: session.sessionId,
+            runId: "run-throw-progress",
+            lines: [{ seq: 0, kind: "log", text: "before throw" }],
+        });
+    });
+
+    it("surfaces the per-run abort signal on the workflow context", async () => {
+        const session = new CopilotSession("session-abort-signal", {} as never);
+        const signalSeen = Promise.withResolvers<AbortSignal>();
+        const workflow = defineWorkflow({
+            meta: {
+                name: "abort-signal",
+                description: "Abort signal test",
+                phases: [],
+            },
+            run: async ({ signal }) => {
+                signalSeen.resolve(signal);
+                await new Promise<void>((resolve) =>
+                    signal.addEventListener("abort", () => resolve(), { once: true })
+                );
+                return signal.aborted;
+            },
+        });
+        session.registerWorkflows([workflow]);
+
+        const execution = session.clientSessionApis.workflow!.execute({
+            sessionId: session.sessionId,
+            name: "abort-signal",
+            runId: "run-abort-signal",
+            args: {},
+        });
+        const signal = await signalSeen.promise;
+        expect(signal.aborted).toBe(false);
+
+        await session.clientSessionApis.workflow!.abort({
+            sessionId: session.sessionId,
+            runId: "run-abort-signal",
+        });
+
+        expect(signal.aborted).toBe(true);
+        await expect(execution).resolves.toEqual({ result: true });
+    });
+
+    it("rejects an in-flight runtime-backed await when workflow.abort trips the signal", async () => {
+        const agentResponse = Promise.withResolvers<{ result: string }>();
+        const sendRequest = vi.fn(async (method: string) => {
+            if (method === "session.workflow.agent") {
+                return agentResponse.promise;
+            }
+            return {};
+        });
+        const session = new CopilotSession("session-abort-await", { sendRequest } as never);
+        const workflow = defineWorkflow({
+            meta: {
+                name: "abort-await",
+                description: "Abort an in-flight workflow await",
+                phases: [],
+            },
+            run: async ({ agent }) => agent("wait forever"),
+        });
+        session.registerWorkflows([workflow]);
+
+        const execution = session.clientSessionApis.workflow!.execute({
+            sessionId: session.sessionId,
+            name: "abort-await",
+            runId: "run-abort-await",
+            args: {},
+        });
+        await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledWith("session.workflow.agent", expect.anything()));
+
+        await session.clientSessionApis.workflow!.abort({
+            sessionId: session.sessionId,
+            runId: "run-abort-await",
+        });
+
+        await expect(execution).rejects.toMatchObject({ name: "AbortError" });
+        agentResponse.resolve({ result: "late" });
+    });
+
+    it("dispatches workflow.execute to the registered workflow selected by name", async () => {
+        const firstRun = vi.fn(async () => ({ selected: "first" }));
+        const secondRun = vi.fn(async ({ args, log }) => {
+            log("executing");
+            return { selected: "second", echoed: args };
+        });
+        const firstWorkflow = defineWorkflow({
+            meta: {
+                name: "first",
+                description: "First workflow",
+                phases: [],
+            },
+            run: firstRun,
+        });
+        const secondWorkflow = defineWorkflow({
+            meta: {
+                name: "second",
+                description: "Second workflow",
+                phases: [],
+            },
+            run: secondRun,
+        });
+        const session = new CopilotSession("session-execute", {
+            sendRequest: vi.fn(async () => ({})),
+        } as never);
+        session.registerWorkflows([firstWorkflow, secondWorkflow]);
+
+        await expect(
+            session.clientSessionApis.workflow!.execute({
+                sessionId: session.sessionId,
+                name: "second",
+                runId: "run-echo",
+                args: { message: "hello" },
+            })
+        ).resolves.toEqual({
+            result: { selected: "second", echoed: { message: "hello" } },
+        });
+        expect(firstRun).not.toHaveBeenCalled();
+        expect(secondRun).toHaveBeenCalledOnce();
+
+        const error = await session.clientSessionApis
+            .workflow!.execute({
+                sessionId: session.sessionId,
+                name: "missing",
+                runId: "run-missing",
+                args: {},
+            })
+            .catch((caught: unknown) => caught);
+        expect(error).toBeInstanceOf(ResponseError);
+        expect((error as ResponseError<{ code: string; name: string }>).data).toEqual({
+            code: "workflow_not_found",
+            name: "missing",
+        });
+    });
+
+    it("runs workflows by name or handle and unwraps only foreground results", async () => {
+        const workflow = defineWorkflow({
+            meta: {
+                name: "friendly-run",
+                description: "Friendly run wrapper",
+                phases: [],
+            },
+            run: async () => ({ unused: true }),
+        });
+        const sendRequest = vi.fn(
+            async (
+                _method: string,
+                params: { name: string; options?: { background?: boolean } }
+            ) =>
+                params.options?.background
+                    ? { runId: "run-background", status: "running" }
+                    : {
+                          runId: "run-foreground",
+                          status: "completed",
+                          result: { name: params.name },
+                      }
+        );
+        const session = new CopilotSession("session-run", { sendRequest } as never);
+
+        await expect(session.workflow.run("by-name", { args: { value: 1 } })).resolves.toEqual({
+            name: "by-name",
+        });
+        await expect(session.workflow.run(workflow)).resolves.toEqual({ name: "friendly-run" });
+        await expect(session.workflow.run("background", { background: true })).resolves.toEqual({
+            runId: "run-background",
+            status: "running",
+        });
+
+        expect(sendRequest).toHaveBeenNthCalledWith(1, "session.workflow.run", {
+            sessionId: session.sessionId,
+            name: "by-name",
+            args: { value: 1 },
+            options: { background: undefined, resumeFromRunId: undefined },
+        });
+        expect(sendRequest).toHaveBeenNthCalledWith(2, "session.workflow.run", {
+            sessionId: session.sessionId,
+            name: "friendly-run",
+            args: {},
+            options: { background: undefined, resumeFromRunId: undefined },
+        });
+    });
+
+    it("throws WorkflowRunError with the full foreground envelope", async () => {
+        const envelope = {
+            runId: "run-error",
+            status: "error" as const,
+            error: "workflow failed",
+            snapshot: { completed: 1 },
+        };
+        const session = new CopilotSession("session-error", {
+            sendRequest: vi.fn(async () => envelope),
+        } as never);
+
+        const error = await session.workflow.run("failing").catch((caught: unknown) => caught);
+        expect(error).toBeInstanceOf(WorkflowRunError);
+        expect((error as WorkflowRunError).envelope).toBe(envelope);
+    });
+});