From c4cc32259cca19df618ddcb48120ffd850c74fd1 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 11:58:44 -0700 Subject: [PATCH 01/11] Phase 1: Contract skeleton and feature flag Regenerate SDK wire types (nodejs/src/generated/rpc.ts) from the updated runtime schema for the new workflow.* methods and DTOs, and mirror WorkflowMeta/WorkflowLimits in the SDK public types (nodejs/src/types.ts) with exports via index.ts/extension.ts. No behavior wired (contract skeleton only). Verification: cd nodejs && npm run build (Node 22) exit 0. Deviations: none. --- nodejs/src/extension.ts | 3 +- nodejs/src/generated/rpc.ts | 491 +++++++++++++++++++++++++++++++++++- nodejs/src/index.ts | 2 + nodejs/src/types.ts | 22 ++ 4 files changed, 508 insertions(+), 10 deletions(-) diff --git a/nodejs/src/extension.ts b/nodejs/src/extension.ts index 72bde93bde..6cec2674f7 100644 --- a/nodejs/src/extension.ts +++ b/nodejs/src/extension.ts @@ -6,7 +6,6 @@ import { CopilotClient } from "./client.js"; import type { CopilotSession } from "./session.js"; import { defaultJoinSessionPermissionHandler, - type ExtensionInfo, type PermissionHandler, type ResumeSessionConfig, } from "./types.js"; @@ -29,7 +28,7 @@ export type JoinSessionConfig = Omit< onPermissionRequest?: PermissionHandler; }; -export type { ExtensionInfo }; +export type { ExtensionInfo, WorkflowLimits, WorkflowMeta } from "./types.js"; /** * Joins the current foreground session. 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 => + 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 => + 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 => + 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 => + 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 => + 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 => + 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 => + 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 => + 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; } +/** 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; + /** + * 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; +} + /** 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..5ae3f4f1f5 100644 --- a/nodejs/src/index.ts +++ b/nodejs/src/index.ts @@ -86,6 +86,8 @@ export type { LargeToolOutputConfig, MemoryConfiguration, UiInputOptions, + WorkflowLimits, + WorkflowMeta, MCPStdioServerConfig, MCPHTTPServerConfig, MCPServerConfig, 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). * From 594c68815e2a495323595385a5a9aca0be94326c Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 12:25:55 -0700 Subject: [PATCH 02/11] Phase 2: SDK authoring surface (defineWorkflow + registration) Add the SDK-side Dynamic Workflows authoring surface: defineWorkflow (opaque frozen WorkflowHandle, module-local name->{meta,run} map, optional limits that reject non-positive values), workflows?: WorkflowHandle[] on the extension JoinSessionConfig only (serializing just WorkflowMeta on the wire, never the run closure), the workflow.execute reverse handler (dispatch-by-name with a stub ctx of args+log-noop+return, structured error on unknown name) and workflow.abort handler (per-run AbortController keyed by runId), both registered before session.resume, and the session.workflow.run friendly wrapper (name/handle overloads; foreground unwraps completed result and throws exported WorkflowRunError on non-completed; background returns the envelope). Exports from index.ts and extension.ts. SDK-only; runtime untouched; no generated files edited. Verification (Node 22, exit 0): npm run build; npm test -- test/workflow.test.ts (12 passed); typecheck; targeted ESLint. Deviations: none. --- nodejs/src/client.ts | 20 +++ nodejs/src/extension.ts | 35 ++++- nodejs/src/index.ts | 12 ++ nodejs/src/session.ts | 97 ++++++++++++++ nodejs/src/workflow.ts | 176 +++++++++++++++++++++++++ nodejs/test/workflow.test.ts | 247 +++++++++++++++++++++++++++++++++++ 6 files changed, 581 insertions(+), 6 deletions(-) create mode 100644 nodejs/src/workflow.ts create mode 100644 nodejs/test/workflow.test.ts 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 { + return this.resumeSessionInternal(sessionId, config); + } + + /** @internal */ + async resumeSessionForExtension( + sessionId: string, + config: ResumeSessionConfig, + workflows?: WorkflowHandle[] + ): Promise { + return this.resumeSessionInternal(sessionId, config, workflows); + } + + private async resumeSessionInternal( + sessionId: string, + config: ResumeSessionConfig, + workflows?: WorkflowHandle[] + ): Promise { 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 6cec2674f7..0d55379f5c 100644 --- a/nodejs/src/extension.ts +++ b/nodejs/src/extension.ts @@ -9,6 +9,7 @@ import { type PermissionHandler, type ResumeSessionConfig, } from "./types.js"; +import type { WorkflowHandle } from "./workflow.js"; export { Canvas, @@ -26,9 +27,23 @@ export type JoinSessionConfig = Omit< "onPermissionRequest" | "extensionSdkPath" > & { onPermissionRequest?: PermissionHandler; + workflows?: WorkflowHandle[]; }; 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. @@ -57,14 +72,22 @@ export async function joinSession(config: JoinSessionConfig = {}): Promise = new Map(); private bearerTokenProviders: Map = new Map(); private commandHandlers: Map = new Map(); + private workflows = new Map>(); + private workflowAbortControllers = new Map(); private permissionHandler?: PermissionHandler; private mcpAuthHandler?: McpAuthHandler; private userInputHandler?: UserInputHandler; @@ -155,6 +166,36 @@ 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 => { + 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"], + }; + /** * Creates a new CopilotSession instance. * @@ -358,6 +399,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 +918,57 @@ 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); + try { + const result = await definition.run({ + args: params.args, + log: (_message: string) => {}, + } as WorkflowContext); + return { result } as WorkflowExecuteResult; + } finally { + if (self.workflowAbortControllers.get(params.runId) === controller) { + self.workflowAbortControllers.delete(params.runId); + } + } + }, + async abort(params) { + self.workflowAbortControllers.get(params.runId)?.abort(); + return {}; + }, + }; + } + /** * Registers per-provider {@link BearerTokenProvider} callbacks for BYOK providers * configured with managed-identity / on-demand bearer-token auth. diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts new file mode 100644 index 0000000000..d68cfdc92b --- /dev/null +++ b/nodejs/src/workflow.ts @@ -0,0 +1,176 @@ +/*--------------------------------------------------------------------------------------------- + * 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; + +/** JSON Schema accepted for structured workflow agent output. */ +export type WorkflowJsonSchema = Record; + +/** 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 = ( + previous: TInput, + item: unknown, + index: number +) => Promise | TResult; + +/** Context passed to an extension-authored workflow body. */ +export interface WorkflowContext { + /** Spawn and await one workflow-scoped subagent. */ + agent(prompt: string, options?: WorkflowAgentOptions): Promise; + /** Memoize an arbitrary producer under a stable author-supplied key. */ + step( + key: string, + producer: () => Promise | TResult, + options?: WorkflowStepOptions + ): Promise; + /** Run thunks concurrently, returning null for a thunk that throws. */ + parallel(thunks: Array<() => Promise>): Promise>; + /** Run each item through every stage without barriers between stages. */ + pipeline(items: unknown[], ...stages: WorkflowPipelineStage[]): Promise; + /** Start a named workflow progress phase. */ + phase(title: string): void; + /** Emit a workflow progress line. */ + log(message: string): void; + /** Invoke another registered workflow as a child run. */ + workflow(name: string, args?: unknown): Promise; + /** 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 { + meta: WorkflowMeta; + run(context: WorkflowContext): Promise; +} + +/** Opaque reusable reference to a defined workflow. */ +export interface WorkflowHandle { + readonly meta: WorkflowMeta; + readonly [workflowHandleBrand]: { + readonly args: TArgs; + readonly result: TResult; + }; +} + +/** Options for invoking a workflow. */ +export interface RunOptions { + /** 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( + workflow: WorkflowHandle, + options: RunOptions & { background: true } + ): Promise; + run( + workflow: WorkflowHandle, + options?: RunOptions & { background?: false } + ): Promise; + run( + workflow: WorkflowHandle, + options?: RunOptions + ): Promise; + run(name: string, options: RunOptions & { background: true }): Promise; + run( + name: string, + options?: RunOptions & { background?: false } + ): Promise; + run( + name: string, + options?: RunOptions + ): Promise; +} + +/** 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): Promise; +} + +const workflowDefinitions = new Map(); +const workflowHandles = new WeakMap(); + +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( + definition: WorkflowDefinition +): WorkflowHandle { + validateLimits(definition.meta); + + const stored: StoredWorkflow = { + meta: definition.meta, + run: definition.run as StoredWorkflow["run"], + }; + const handle = Object.freeze({ meta: definition.meta }) as WorkflowHandle; + + workflowDefinitions.set(definition.meta.name, stored); + 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..0fe5388ca8 --- /dev/null +++ b/nodejs/test/workflow.test.ts @@ -0,0 +1,247 @@ +/*--------------------------------------------------------------------------------------------- + * 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 WorkflowDefinition } from "../src/workflow.js"; + +async function stopClient(client: CopilotClient): Promise { + 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) => { + if (method === "session.resume") { + const sessions = (client as never as { sessions: Map }) + .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("dispatches workflow.execute by name and returns a structured unknown-name error", async () => { + const run = vi.fn(async ({ args, log }) => { + log("ignored in phase 2"); + return { echoed: args }; + }); + const workflow = defineWorkflow({ + meta: { + name: "echo", + description: "Echo arguments", + phases: [], + }, + run, + }); + const session = new CopilotSession("session-execute", {} as never); + session.registerWorkflows([workflow]); + + await expect( + session.clientSessionApis.workflow!.execute({ + sessionId: session.sessionId, + name: "echo", + runId: "run-echo", + args: { message: "hello" }, + }) + ).resolves.toEqual({ result: { echoed: { message: "hello" } } }); + + 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); + }); +}); From f29ad5720c5496af164c4875a99c38dc8238c118 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 14:05:18 -0700 Subject: [PATCH 03/11] Phase 4: Host API context, part 1 (args, log, phase, return) Build the real run() WorkflowContext in the SDK workflow.execute handler (replacing the Phase 2 stub): populate args; set ctx.session to the identical CopilotSession instance joinSession returned (strict === identity, no wrapper); expose ctx.signal as the per-run AbortController's AbortSignal; and implement log()/phase() as synchronous calls that buffer each line with a monotonic seq and flush incrementally over workflow.log (at await boundaries, a short unref'd timer, and a guaranteed finally-flush so no buffered narration is lost on a throwing/aborted body). SDK side of Phase 4; runtime workflow.log handler is the companion commit. Verification (Node 22, exit 0): npm run build; npm test -- test/workflow.test.ts (16 passed); typecheck; lint. Deviations: none. --- nodejs/src/session.ts | 97 ++++++++++++++++++-- nodejs/test/workflow.test.ts | 167 ++++++++++++++++++++++++++++++++++- 2 files changed, 257 insertions(+), 7 deletions(-) diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 52c6ac9b81..54e00ee65b 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -16,6 +16,7 @@ import type { CurrentToolMetadata, McpOauthPendingRequestResponse, WorkflowExecuteResult, + WorkflowLogLine, } from "./generated/rpc.js"; import { type Canvas, CanvasError } from "./canvas.js"; import type { OpenCanvasInstance } from "./generated/rpc.js"; @@ -103,6 +104,72 @@ function isOpenCanvasInstance(value: unknown): value is OpenCanvasInstance { ); } +const WORKFLOW_LOG_FLUSH_DELAY_MS = 10; + +class WorkflowProgressBuffer { + private nextSeq = 0; + private pending: WorkflowLogLine[] = []; + private flushTimer?: ReturnType; + private flushTail: Promise = Promise.resolve(); + private flushError: unknown; + private flushFailed = false; + private closed = false; + + constructor(private readonly send: (lines: WorkflowLogLine[]) => Promise) {} + + 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 { + 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 { + 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; + } + } +} + /** Assistant message event - the final response from the assistant. */ export type AssistantMessageEvent = Extract; @@ -950,15 +1017,35 @@ export class CopilotSession { 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 result = await definition.run({ + const unsupported = async (method: string): Promise => { + await progress.flush(); + throw new Error(`${method} is not yet supported`); + }; + const context: WorkflowContext = { args: params.args, - log: (_message: string) => {}, - } as WorkflowContext); + session: self, + signal: controller.signal, + phase: (title: string) => progress.enqueue("phase", title), + log: (message: string) => progress.enqueue("log", message), + agent: async () => unsupported("workflow.agent"), + step: async () => unsupported("workflow.step"), + parallel: async () => unsupported("workflow.parallel"), + pipeline: async () => unsupported("workflow.pipeline"), + workflow: async () => unsupported("workflow.runNested"), + }; + const result = await definition.run(context); return { result } as WorkflowExecuteResult; } finally { - if (self.workflowAbortControllers.get(params.runId) === controller) { - self.workflowAbortControllers.delete(params.runId); + try { + await progress.close(); + } finally { + if (self.workflowAbortControllers.get(params.runId) === controller) { + self.workflowAbortControllers.delete(params.runId); + } } } }, diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 0fe5388ca8..06a055b650 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -142,9 +142,170 @@ describe("workflows", () => { ); }); + it("builds the workflow context with args, progress, signal, and the joined session identity", async () => { + process.env.SESSION_ID = "session-context"; + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.workflow.log") { + return {}; + } + 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"); + return { ok: true }; + }, + }); + 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.signal).toBeInstanceOf(AbortSignal); + expect(executeResult).toEqual({ result: { ok: true } }); + 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("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(); + 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("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(); + const workflow = defineWorkflow({ + meta: { + name: "abort-signal", + description: "Abort signal test", + phases: [], + }, + run: async ({ signal }) => { + signalSeen.resolve(signal); + await new Promise((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("dispatches workflow.execute by name and returns a structured unknown-name error", async () => { const run = vi.fn(async ({ args, log }) => { - log("ignored in phase 2"); + log("executing"); return { echoed: args }; }); const workflow = defineWorkflow({ @@ -155,7 +316,9 @@ describe("workflows", () => { }, run, }); - const session = new CopilotSession("session-execute", {} as never); + const session = new CopilotSession("session-execute", { + sendRequest: vi.fn(async () => ({})), + } as never); session.registerWorkflows([workflow]); await expect( From 4ab5967eb39021be2826f71b584374090a6e7dc2 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 15:51:41 -0700 Subject: [PATCH 04/11] Phase 5: The one-agent executor (`agent()` via in-session subagents) Implement ctx.agent in the SDK run() context: it calls workflow.agent carrying the closed-over workflowRunId and the model, returns the subagent result text, and trims opts to label/schema/model only (effort/isolation/agentType/tool-narrowing are not v1). Schema handling arrives in Phase 6. Verification (Node 22, exit 0): npm run build; npm test -- test/workflow.test.ts (17 passed, asserts effort is dropped). Deviations: none. --- nodejs/src/session.ts | 14 +++++++++- nodejs/test/workflow.test.ts | 51 +++++++++++++++++++++++++++++++++++- 2 files changed, 63 insertions(+), 2 deletions(-) diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 54e00ee65b..62e4c82f98 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -1031,7 +1031,19 @@ export class CopilotSession { signal: controller.signal, phase: (title: string) => progress.enqueue("phase", title), log: (message: string) => progress.enqueue("log", message), - agent: async () => unsupported("workflow.agent"), + agent: async (prompt, options = {}) => { + await progress.flush(); + const response = await self.rpc.workflow.agent({ + workflowRunId: params.runId, + prompt, + opts: { + label: options.label, + schema: options.schema, + model: options.model, + }, + }); + return response.result ?? null; + }, step: async () => unsupported("workflow.step"), parallel: async () => unsupported("workflow.parallel"), pipeline: async () => unsupported("workflow.pipeline"), diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 06a055b650..3f4fafe46f 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -7,7 +7,12 @@ 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 WorkflowDefinition } from "../src/workflow.js"; +import { + defineWorkflow, + WorkflowRunError, + type WorkflowAgentOptions, + type WorkflowDefinition, +} from "../src/workflow.js"; async function stopClient(client: CopilotClient): Promise { await client.stop(); @@ -235,6 +240,50 @@ describe("workflows", () => { 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("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); From cc6e45b32ac14d689019b218b0478b95e49fc3aa Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 16:43:29 -0700 Subject: [PATCH 05/11] Phase 6: Structured output (`agent({ schema })`) and retry Document the supported schema subset on the WorkflowContext agent() schema option (examon validator subset, not arbitrary JSON Schema) so authors do not expect unsupported keywords. Companion to the runtime Rust parse/validate. Verification (Node 22, exit 0): npm run build; npm run typecheck; npm test -- test/workflow.test.ts. Deviations: none. --- nodejs/src/workflow.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts index d68cfdc92b..0475a6b43a 100644 --- a/nodejs/src/workflow.ts +++ b/nodejs/src/workflow.ts @@ -8,7 +8,12 @@ import type { WorkflowMeta } from "./types.js"; declare const workflowHandleBrand: unique symbol; -/** JSON Schema accepted for structured workflow agent output. */ +/** + * 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; /** Options for one workflow-scoped subagent call. */ From 209d9ae2072ab92647854c4674865113351d3f53 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 18:27:42 -0700 Subject: [PATCH 06/11] Phase 7: Combinators (`parallel`, `pipeline`) and the per-run limiter Implement ctx.parallel(thunks) as a barrier fan-out (awaits all; throwing thunk -> null) and ctx.pipeline(items, ...stages) as per-item staged flow with no barrier (throwing stage nulls that item and skips its remaining stages). Both are pure SDK control flow holding no slot; concurrency is enforced by the runtime per-run limiter (each leaf ctx.agent is a workflow.agent RPC that queues until admitted). Enforce the non-configurable 4096 per-fan-out item cap, and validate element types up front so passing already-invoked promises surfaces a clear pass-functions-not-promises diagnostic instead of silently resolving to all-nulls. Verification (Node 22, exit 0): npm run build; npm test -- test/workflow.test.ts (22 passed); typecheck. Deviations: none. --- nodejs/src/session.ts | 61 +++++++++- nodejs/src/workflow.ts | 4 +- nodejs/test/workflow.test.ts | 214 +++++++++++++++++++++++++++++++++++ 3 files changed, 276 insertions(+), 3 deletions(-) diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 62e4c82f98..4a4f4482d2 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -105,6 +105,63 @@ 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( + thunks: Array<() => Promise | TResult> +): Promise> { + 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 + > +): Promise { + 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; @@ -1045,8 +1102,8 @@ export class CopilotSession { return response.result ?? null; }, step: async () => unsupported("workflow.step"), - parallel: async () => unsupported("workflow.parallel"), - pipeline: async () => unsupported("workflow.pipeline"), + parallel: runWorkflowParallel, + pipeline: runWorkflowPipeline, workflow: async () => unsupported("workflow.runNested"), }; const result = await definition.run(context); diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts index 0475a6b43a..3680f91fcd 100644 --- a/nodejs/src/workflow.ts +++ b/nodejs/src/workflow.ts @@ -47,7 +47,9 @@ export interface WorkflowContext { options?: WorkflowStepOptions ): Promise; /** Run thunks concurrently, returning null for a thunk that throws. */ - parallel(thunks: Array<() => Promise>): Promise>; + parallel( + thunks: Array<() => Promise | TResult> + ): Promise>; /** Run each item through every stage without barriers between stages. */ pipeline(items: unknown[], ...stages: WorkflowPipelineStage[]): Promise; /** Start a named workflow progress phase. */ diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 3f4fafe46f..60c04c223a 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -284,6 +284,220 @@ describe("workflows", () => { }); }); + it("runs parallel as a barrier and maps a throwing thunk to null", async () => { + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + 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 + >), + }); + 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(); + const secondStageStarted = Promise.withResolvers(); + 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(); + 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); From 2cc0f0b270eacc6065919aa07b36aa71998d5a22 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 20:39:58 -0700 Subject: [PATCH 07/11] Phase 8: Durable `step()` and execution state in SQL SDK-side durable step(): ctx.step(key, producer) does get-then-put across the RPC gap (workflow.journal.get -> run producer on miss -> workflow.journal.put), with a cached JSON null served as a hit and producer failures left un-journaled (retried on resume). Adds getRun() forwarding and resume integration wiring. Pairs with the runtime SQL journal/run store. Verified (workflows-scoped, exit 0): npm run build; test/workflow.test.ts (24 passed, incl. step-once + cached-null + failures-not-cached + getRun forwarding). Deviations: none. --- nodejs/src/session.ts | 38 +++++++++++++++++- nodejs/src/workflow.ts | 2 + nodejs/test/workflow.test.ts | 76 ++++++++++++++++++++++++++++++++++++ 3 files changed, 115 insertions(+), 1 deletion(-) diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 4a4f4482d2..ccdab1a06a 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -69,6 +69,7 @@ import { type SessionWorkflowApi, type WorkflowContext, type WorkflowHandle, + type WorkflowStepOptions, } from "./workflow.js"; /** @@ -318,6 +319,7 @@ export class CopilotSession { } return envelope.result; }) as SessionWorkflowApi["run"], + getRun: (runId) => this.rpc.workflow.getRun({ runId }), }; /** @@ -1101,7 +1103,41 @@ export class CopilotSession { }); return response.result ?? null; }, - step: async () => unsupported("workflow.step"), + step: async ( + key: string, + producer: () => Promise | TResult, + options: WorkflowStepOptions = {} + ): Promise => { + await progress.flush(); + if (options.volatile) { + return producer(); + } + const cached = await self.rpc.workflow.journal.get({ + runId: params.runId, + key, + }); + 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 self.rpc.workflow.journal.put({ + runId: params.runId, + key, + resultJson: result as Parameters< + typeof self.rpc.workflow.journal.put + >[0]["resultJson"], + }); + return result; + }, parallel: runWorkflowParallel, pipeline: runWorkflowPipeline, workflow: async () => unsupported("workflow.runNested"), diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts index 3680f91fcd..0c60d09b4a 100644 --- a/nodejs/src/workflow.ts +++ b/nodejs/src/workflow.ts @@ -114,6 +114,8 @@ export interface SessionWorkflowApi { name: string, options?: RunOptions ): Promise; + /** Read the latest durable envelope for a workflow run. */ + getRun(runId: string): Promise; } /** Error thrown when a foreground workflow run does not complete successfully. */ diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 60c04c223a..961e997b03 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -284,6 +284,82 @@ describe("workflows", () => { }); }); + it("runs a durable step once, serves cached null, and does not cache failures", async () => { + const journal = new Map(); + 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("runs parallel as a barrier and maps a throwing thunk to null", async () => { const first = Promise.withResolvers(); const second = Promise.withResolvers(); From dcdeaf591a62ef41efcb61b4f2127ba6b98acf1d Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 22:56:27 -0700 Subject: [PATCH 08/11] Phase 9: Limits enforcement and the pre-run approval gate SDK-side cancellation support for the Phase-9 limits/timeout/cancel machinery: workflow.abort trips the per-run AbortController so in-flight ctx awaits reject with an AbortError and a well-behaved body unwinds cooperatively; session.workflow.cancel forwards the runId. Pairs with the runtime enforcement, approval/feedback gate, and halt->continue flow. Verified (workflows-scoped, exit 0): npm run build; test/workflow.test.ts (both new Phase-9 tests pass: cancel forwards runId; workflow.abort rejects the in-flight await with AbortError). The lone unrelated failure is the documented win32-arm64 platform-package env wrinkle. Deviations: none. --- nodejs/src/session.ts | 92 +++++++++++++++++++++++++++--------- nodejs/src/workflow.ts | 2 + nodejs/test/workflow.test.ts | 48 +++++++++++++++++++ 3 files changed, 119 insertions(+), 23 deletions(-) diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index ccdab1a06a..638e56a241 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -179,6 +179,7 @@ class WorkflowProgressBuffer { if (this.closed) { throw new Error("Cannot log after the workflow run has settled"); } + this.pending.push({ seq: this.nextSeq++, kind, text }); this.scheduleFlush(); } @@ -228,6 +229,33 @@ class WorkflowProgressBuffer { } } +async function awaitWorkflowOperation( + operation: Promise, + signal: AbortSignal +): Promise { + 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((_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; @@ -320,6 +348,7 @@ export class CopilotSession { return envelope.result; }) as SessionWorkflowApi["run"], getRun: (runId) => this.rpc.workflow.getRun({ runId }), + cancel: (runId) => this.rpc.workflow.cancel({ runId }), }; /** @@ -1088,19 +1117,28 @@ export class CopilotSession { args: params.args, session: self, signal: controller.signal, - phase: (title: string) => progress.enqueue("phase", title), - log: (message: string) => progress.enqueue("log", message), + 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 self.rpc.workflow.agent({ - workflowRunId: params.runId, - prompt, - opts: { - label: options.label, - schema: options.schema, - model: options.model, - }, - }); + 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 ( @@ -1112,10 +1150,13 @@ export class CopilotSession { if (options.volatile) { return producer(); } - const cached = await self.rpc.workflow.journal.get({ - runId: params.runId, - key, - }); + const cached = await awaitWorkflowOperation( + self.rpc.workflow.journal.get({ + runId: params.runId, + key, + }), + controller.signal + ); if (cached.hit) { return cached.resultJson as TResult; } @@ -1129,13 +1170,16 @@ export class CopilotSession { `step("${key}") returned a value that is not JSON-serializable` ); } - await self.rpc.workflow.journal.put({ - runId: params.runId, - key, - resultJson: result as Parameters< - typeof self.rpc.workflow.journal.put - >[0]["resultJson"], - }); + 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, @@ -1155,7 +1199,9 @@ export class CopilotSession { } }, async abort(params) { - self.workflowAbortControllers.get(params.runId)?.abort(); + self.workflowAbortControllers + .get(params.runId) + ?.abort(new DOMException("Workflow run was aborted", "AbortError")); return {}; }, }; diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts index 0c60d09b4a..9d8b2a4feb 100644 --- a/nodejs/src/workflow.ts +++ b/nodejs/src/workflow.ts @@ -116,6 +116,8 @@ export interface SessionWorkflowApi { ): Promise; /** Read the latest durable envelope for a workflow run. */ getRun(runId: string): Promise; + /** Cancel a workflow run and return its terminal envelope. */ + cancel(runId: string): Promise; } /** Error thrown when a foreground workflow run does not complete successfully. */ diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 961e997b03..451ecc8ae9 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -360,6 +360,18 @@ describe("workflows", () => { }); }); + 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(); const second = Promise.withResolvers(); @@ -642,6 +654,42 @@ describe("workflows", () => { 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 by name and returns a structured unknown-name error", async () => { const run = vi.fn(async ({ args, log }) => { log("executing"); From c5128a7106910b0914b92097708c25329c9cc3fe Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 23:27:26 -0700 Subject: [PATCH 09/11] Phase 10: Full session and Node power (the trust-tier payoff) Strengthens the SDK workflow-context unit coverage to lock the full-session trust-tier guarantee: asserts ctx.session is identical to the joinSession result (context.session === joinSessionResult and context.session.rpc === joinSessionResult.rpc, proving no facade/pruning) and that session RPC is reachable from the context (invokes context.session.rpc.tasks.list()). Test-only; pairs with the runtime E2E. Verified (workflows-scoped, exit 0): npm run build; test/workflow.test.ts 26/26. Deviations: none. --- nodejs/test/workflow.test.ts | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 451ecc8ae9..409522a6f0 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -147,12 +147,15 @@ describe("workflows", () => { ); }); - it("builds the workflow context with args, progress, signal, and the joined session identity", async () => { + 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); @@ -171,7 +174,8 @@ describe("workflows", () => { contextSeen.resolve(context); context.phase("A"); context.log("hi"); - return { ok: true }; + const tasks = await context.session.rpc.tasks.list(); + return { ok: true, taskCount: tasks.tasks.length }; }, }); vi.spyOn(CopilotClient.prototype, "resumeSessionForExtension").mockImplementation( @@ -192,8 +196,12 @@ describe("workflows", () => { 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 } }); + 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", From 9571e1938298b835d4adb645f82f37182dcd2937 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Thu, 16 Jul 2026 00:35:33 -0700 Subject: [PATCH 10/11] Phase 11: Invocation ergonomics: the `run_workflow` tool, commands, name identity Makes name-based resolution the primary path in session.workflow.run (the workflow-handle overload becomes local sugar), matching how tools/commands/subagents resolve by name, and removes a now-dead module-global workflowDefinitions map (workflow.execute resolves via the session-scoped registry instead). Adds unit coverage that name resolution selects the right registered workflow and that run-by-name forwards {name, args}. Duplicate-name rejection continues to rely on the Phase-3 registration policy. Verified (workflows-scoped, exit 0): npm run build; test/workflow.test.ts (26 passed). Deviations: none. --- nodejs/src/workflow.ts | 20 +++++++++----------- nodejs/test/workflow.test.ts | 33 +++++++++++++++++++++++---------- 2 files changed, 32 insertions(+), 21 deletions(-) diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts index 9d8b2a4feb..52a2bf2982 100644 --- a/nodejs/src/workflow.ts +++ b/nodejs/src/workflow.ts @@ -93,6 +93,15 @@ export interface RunOptions { /** Friendly workflow API exposed on a session. */ export interface SessionWorkflowApi { + run(name: string, options: RunOptions & { background: true }): Promise; + run( + name: string, + options?: RunOptions & { background?: false } + ): Promise; + run( + name: string, + options?: RunOptions + ): Promise; run( workflow: WorkflowHandle, options: RunOptions & { background: true } @@ -105,15 +114,6 @@ export interface SessionWorkflowApi { workflow: WorkflowHandle, options?: RunOptions ): Promise; - run(name: string, options: RunOptions & { background: true }): Promise; - run( - name: string, - options?: RunOptions & { background?: false } - ): Promise; - run( - name: string, - options?: RunOptions - ): Promise; /** Read the latest durable envelope for a workflow run. */ getRun(runId: string): Promise; /** Cancel a workflow run and return its terminal envelope. */ @@ -137,7 +137,6 @@ interface StoredWorkflow { run(context: WorkflowContext): Promise; } -const workflowDefinitions = new Map(); const workflowHandles = new WeakMap(); function validateLimits(meta: WorkflowMeta): void { @@ -172,7 +171,6 @@ export function defineWorkflow( }; const handle = Object.freeze({ meta: definition.meta }) as WorkflowHandle; - workflowDefinitions.set(definition.meta.name, stored); workflowHandles.set(handle, stored); return handle; } diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index 409522a6f0..bf5ebd61dc 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -698,32 +698,45 @@ describe("workflows", () => { agentResponse.resolve({ result: "late" }); }); - it("dispatches workflow.execute by name and returns a structured unknown-name error", async () => { - const run = vi.fn(async ({ args, log }) => { + 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 { echoed: args }; + return { selected: "second", echoed: args }; }); - const workflow = defineWorkflow({ + const firstWorkflow = defineWorkflow({ meta: { - name: "echo", - description: "Echo arguments", + name: "first", + description: "First workflow", phases: [], }, - run, + 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([workflow]); + session.registerWorkflows([firstWorkflow, secondWorkflow]); await expect( session.clientSessionApis.workflow!.execute({ sessionId: session.sessionId, - name: "echo", + name: "second", runId: "run-echo", args: { message: "hello" }, }) - ).resolves.toEqual({ result: { echoed: { message: "hello" } } }); + ).resolves.toEqual({ + result: { selected: "second", echoed: { message: "hello" } }, + }); + expect(firstRun).not.toHaveBeenCalled(); + expect(secondRun).toHaveBeenCalledOnce(); const error = await session.clientSessionApis .workflow!.execute({ From d88b9c812a92f2bf07bca6005b4f6739e59dc9c1 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Thu, 16 Jul 2026 00:55:55 -0700 Subject: [PATCH 11/11] Phase 12: Forbid workflow nesting Nested workflows are not supported: ctx.workflow(name, args) now throws a clear "nested workflows are not supported" error synchronously at the call site and forwards NO workflow.runNested request (no RPC leaves the SDK). The parentRunId/rootRunId contract scaffolding is untouched. Pairs with the runtime workflow.runNested structured rejection. (Phase reshaped from "Nested workflows" to "Forbid workflow nesting" per a human-authorized plan amendment.) Verified (workflows-scoped, exit 0): npm run build; test/workflow.test.ts 27/27 (incl. ctx.workflow-throws-without-forwarding test asserting sendRequest is not called). Deviations: none. --- nodejs/src/session.ts | 8 +++----- nodejs/src/workflow.ts | 2 +- nodejs/test/workflow.test.ts | 26 ++++++++++++++++++++++++++ 3 files changed, 30 insertions(+), 6 deletions(-) diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 638e56a241..89f5f0de87 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -1109,10 +1109,6 @@ export class CopilotSession { await self.rpc.workflow.log({ runId: params.runId, lines }); }); try { - const unsupported = async (method: string): Promise => { - await progress.flush(); - throw new Error(`${method} is not yet supported`); - }; const context: WorkflowContext = { args: params.args, session: self, @@ -1184,7 +1180,9 @@ export class CopilotSession { }, parallel: runWorkflowParallel, pipeline: runWorkflowPipeline, - workflow: async () => unsupported("workflow.runNested"), + workflow: async () => { + throw new Error("nested workflows are not supported"); + }, }; const result = await definition.run(context); return { result } as WorkflowExecuteResult; diff --git a/nodejs/src/workflow.ts b/nodejs/src/workflow.ts index 52a2bf2982..0306a0d4e1 100644 --- a/nodejs/src/workflow.ts +++ b/nodejs/src/workflow.ts @@ -56,7 +56,7 @@ export interface WorkflowContext { phase(title: string): void; /** Emit a workflow progress line. */ log(message: string): void; - /** Invoke another registered workflow as a child run. */ + /** Reject because nested workflows are not supported. */ workflow(name: string, args?: unknown): Promise; /** Caller-supplied input, forwarded verbatim. */ args: TArgs; diff --git a/nodejs/test/workflow.test.ts b/nodejs/test/workflow.test.ts index bf5ebd61dc..a602d2eef3 100644 --- a/nodejs/test/workflow.test.ts +++ b/nodejs/test/workflow.test.ts @@ -212,6 +212,32 @@ describe("workflows", () => { }); }); + 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);