From c4cc32259cca19df618ddcb48120ffd850c74fd1 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Wed, 15 Jul 2026 11:58:44 -0700 Subject: [PATCH 1/6] 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 2/6] 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 3/6] 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 4/6] 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 5/6] 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 6/6] 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);