diff --git a/nodejs/src/client.ts b/nodejs/src/client.ts index 61f4a9941..6d99ce49e 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 { FactoryHandle } from "./factory.js"; /** * Minimum protocol version this SDK can communicate with. @@ -1663,6 +1664,23 @@ export class CopilotClient { * ``` */ async resumeSession(sessionId: string, config: ResumeSessionConfig): Promise { + return this.resumeSessionInternal(sessionId, config); + } + + /** @internal */ + async resumeSessionForExtension( + sessionId: string, + config: ResumeSessionConfig, + factories?: FactoryHandle[] + ): Promise { + return this.resumeSessionInternal(sessionId, config, factories); + } + + private async resumeSessionInternal( + sessionId: string, + config: ResumeSessionConfig, + factories?: FactoryHandle[] + ): Promise { if (!this.connection) { await this.start(); } @@ -1679,6 +1697,7 @@ export class CopilotClient { session.registerTools(config.tools); session.registerCanvases(config.canvases); session.registerCommands(config.commands); + session.registerFactories(factories); const { wireProvider: bearerWireProvider, wireProviders: bearerWireProviders, @@ -1750,6 +1769,7 @@ export class CopilotClient { })), toolSearch: config.toolSearch, canvases: config.canvases?.map((canvas) => canvas.declaration), + factories: factories?.map((factory) => factory.meta), requestCanvasRenderer: config.requestCanvasRenderer, requestExtensions: config.requestExtensions, extensionSdkPath: config.extensionSdkPath, diff --git a/nodejs/src/extension.ts b/nodejs/src/extension.ts index 72bde93bd..8553d7d72 100644 --- a/nodejs/src/extension.ts +++ b/nodejs/src/extension.ts @@ -6,10 +6,10 @@ import { CopilotClient } from "./client.js"; import type { CopilotSession } from "./session.js"; import { defaultJoinSessionPermissionHandler, - type ExtensionInfo, type PermissionHandler, type ResumeSessionConfig, } from "./types.js"; +import type { FactoryHandle } from "./factory.js"; export { Canvas, @@ -27,9 +27,30 @@ export type JoinSessionConfig = Omit< "onPermissionRequest" | "extensionSdkPath" > & { onPermissionRequest?: PermissionHandler; + /** + * Factory handles to register when the extension joins the session. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ + factories?: FactoryHandle[]; }; -export type { ExtensionInfo }; +export type { ExtensionInfo, FactoryLimits, FactoryMeta } from "./types.js"; +export { + defineFactory, + FactoryRunError, + type RunOptions, + type SessionFactoryApi, + type FactoryAgentOptions, + type FactoryContext, + type FactoryDefinition, + type FactoryHandle, + type FactoryJsonSchema, + type FactoryPipelineStage, + type FactoryStepOptions, + type FactoryRunResult, +} from "./factory.js"; /** * Joins the current foreground session. @@ -58,14 +79,22 @@ export async function joinSession(config: JoinSessionConfig = {}): Promise; + +/** + * Options for one factory-scoped subagent call. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryAgentOptions { + label?: string; + schema?: FactoryJsonSchema; + model?: string; +} + +/** + * Options for a durable factory step. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryStepOptions { + /** Skip the journal and always invoke the producer. */ + volatile?: boolean; +} + +/** + * One stage in a per-item factory pipeline. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export type FactoryPipelineStage = ( + previous: TInput, + item: unknown, + index: number +) => Promise | TResult; + +/** + * Context passed to an extension-authored factory body. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryContext { + /** Spawn and await one factory-scoped subagent. */ + agent(prompt: string, options?: FactoryAgentOptions): Promise; + /** Memoize an arbitrary producer under a stable author-supplied key. */ + step( + key: string, + producer: () => Promise | TResult, + options?: FactoryStepOptions + ): Promise; + /** Run thunks concurrently, returning null for a thunk that throws. */ + parallel( + thunks: Array<() => Promise | TResult> + ): Promise>; + /** Run each item through every stage without barriers between stages. */ + pipeline(items: unknown[], ...stages: FactoryPipelineStage[]): Promise; + /** Start a named factory progress phase. */ + phase(title: string): void; + /** Emit a factory progress line. */ + log(message: string): void; + /** Reject because nested factories are not supported. */ + factory(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 factory run. */ + signal: AbortSignal; +} + +/** + * Definition accepted by {@link defineFactory}. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryDefinition { + meta: FactoryMeta; + run(context: FactoryContext): Promise; +} + +/** + * Opaque reusable reference to a defined factory. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryHandle { + readonly meta: FactoryMeta; + readonly [factoryHandleBrand]: { + readonly args: TArgs; + readonly result: TResult; + }; +} + +/** + * Options for invoking a factory. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +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 factory API exposed on a session. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface SessionFactoryApi { + run(name: string, options: RunOptions & { background: true }): Promise; + run( + name: string, + options?: RunOptions & { background?: false } + ): Promise; + run(name: string, options?: RunOptions): Promise; + run( + factory: FactoryHandle, + options: RunOptions & { background: true } + ): Promise; + run( + factory: FactoryHandle, + options?: RunOptions & { background?: false } + ): Promise; + run( + factory: FactoryHandle, + options?: RunOptions + ): Promise; + /** Read the latest durable envelope for a factory run. */ + getRun(runId: string): Promise; + /** Cancel a factory run and return its terminal envelope. */ + cancel(runId: string): Promise; +} + +/** + * Error thrown when a foreground factory run does not complete successfully. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export class FactoryRunError extends Error { + constructor(public readonly envelope: FactoryRunResult) { + super( + envelope.error ?? + envelope.reason ?? + `Factory run "${envelope.runId}" ended with status "${envelope.status}"` + ); + this.name = "FactoryRunError"; + } +} + +interface StoredFactory { + meta: FactoryMeta; + run(context: FactoryContext): Promise; +} + +const factoryHandles = new WeakMap(); + +/** + * Maximum accepted factory timeout in milliseconds (2^31-1). Node clamps + * `setTimeout` delays above this to ~1ms, so a larger value would invert the + * declared timeout into an immediate halt. + */ +const MAX_FACTORY_TIMEOUT_MS = 2_147_483_647; + +function validateLimits(meta: FactoryMeta): 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(`Factory limit "${field}" must be a positive integer`); + } + } + + if (limits.timeout !== undefined && (!Number.isFinite(limits.timeout) || limits.timeout <= 0)) { + throw new Error('Factory limit "timeout" must be a positive number of milliseconds'); + } + // Node clamps setTimeout delays above 2^31-1 ms to ~1ms, which would make a + // large timeout halt the run almost immediately. Reject it up front. + if (limits.timeout !== undefined && limits.timeout > MAX_FACTORY_TIMEOUT_MS) { + throw new Error( + `Factory limit "timeout" must not exceed ${MAX_FACTORY_TIMEOUT_MS} milliseconds (~24.8 days)` + ); + } +} + +/** + * Defines an extension-authored factory and returns an opaque registration handle. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export function defineFactory( + definition: FactoryDefinition +): FactoryHandle { + validateLimits(definition.meta); + + const stored: StoredFactory = { + meta: definition.meta, + run: definition.run as StoredFactory["run"], + }; + const handle = Object.freeze({ meta: definition.meta }) as FactoryHandle; + + factoryHandles.set(handle, stored); + return handle; +} + +/** @internal */ +export function getFactoryDefinition(handle: FactoryHandle): StoredFactory { + const definition = factoryHandles.get(handle); + if (!definition) { + throw new Error("Invalid factory handle"); + } + return definition; +} diff --git a/nodejs/src/generated/rpc.ts b/nodejs/src/generated/rpc.ts index 255a769d5..361e0f259 100644 --- a/nodejs/src/generated/rpc.ts +++ b/nodejs/src/generated/rpc.ts @@ -509,6 +509,38 @@ export type ExternalToolTextResultForLlmContentResourceLinkIconTheme = export type ExternalToolTextResultForLlmContentResourceDetails = | EmbeddedTextResourceContents | EmbeddedBlobResourceContents; +/** + * Kind of factory progress line. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryLogLineKind". + */ +/** @experimental */ +export type FactoryLogLineKind = + /** A narrator log line. */ + | "log" + /** A named factory phase marker. */ + | "phase"; +/** + * Current or terminal state of a factory run. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryRunStatus". + */ +/** @experimental */ +export type FactoryRunStatus = + /** 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 factory body failed. */ + | "error"; /** * Content filtering mode to apply to all tools, or a map of tool name to content filtering mode. * @@ -540,6 +572,8 @@ export type HookType = | "postToolUseFailure" /** Runs after the user submits a prompt. */ | "userPromptSubmitted" + /** Runs after the runtime transforms the submitted prompt for the model, before it is added to session history. */ + | "userPromptTransformed" /** Runs when a session starts. */ | "sessionStart" /** Runs when a session ends. */ @@ -4831,6 +4865,320 @@ export interface ExternalToolTextResultForLlmContentResource { type: "resource"; resource: ExternalToolTextResultForLlmContentResourceDetails; } +/** + * Parameters for cooperatively aborting a factory body. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryAbortRequest". + */ +/** @experimental */ +export interface FactoryAbortRequest { + /** + * Target session identifier + */ + sessionId: string; + /** + * Factory run identifier. + */ + runId: string; +} +/** + * Acknowledgement that a factory request was accepted. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryAckResult". + */ +/** @experimental */ +export interface FactoryAckResult {} +/** + * Options for one factory-scoped subagent call. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryAgentOptions". + */ +/** @experimental */ +export interface FactoryAgentOptions { + /** + * 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 factory-scoped subagent call. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryAgentRequest". + */ +/** @experimental */ +export interface FactoryAgentRequest { + /** + * Factory run identifier that owns the subagent. + */ + factoryRunId: string; + /** + * Prompt to send to the subagent. + */ + prompt: string; + opts: FactoryAgentOptions; +} +/** + * Result of one factory-scoped subagent call. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryAgentResult". + */ +/** @experimental */ +export interface FactoryAgentResult { + /** + * Agent result, omitted when the agent produced no result. + */ + result?: { + [k: string]: unknown | undefined; + }; +} +/** + * Parameters for cancelling a factory run. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryCancelRequest". + */ +/** @experimental */ +export interface FactoryCancelRequest { + /** + * Factory run identifier. + */ + runId: string; +} +/** + * Parameters sent to the owning extension to execute a factory closure. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryExecuteRequest". + */ +/** @experimental */ +export interface FactoryExecuteRequest { + /** + * Target session identifier + */ + sessionId: string; + /** + * Registered factory name. + */ + name: string; + /** + * Factory run identifier. + */ + runId: string; + /** + * Factory input value. + */ + args: { + [k: string]: unknown | undefined; + }; +} +/** + * Result returned by an extension factory closure. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryExecuteResult". + */ +/** @experimental */ +export interface FactoryExecuteResult { + /** + * Factory result value. + */ + result: { + [k: string]: unknown | undefined; + }; +} +/** + * Parameters for retrieving a factory run. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryGetRunRequest". + */ +/** @experimental */ +export interface FactoryGetRunRequest { + /** + * Factory run identifier. + */ + runId: string; +} +/** + * Parameters for reading a factory journal entry. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryJournalGetRequest". + */ +/** @experimental */ +export interface FactoryJournalGetRequest { + /** + * Factory run identifier. + */ + runId: string; + /** + * Namespaced journal key. + */ + key: string; +} +/** + * Result of reading a factory journal entry. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryJournalGetResult". + */ +/** @experimental */ +export interface FactoryJournalGetResult { + /** + * 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 factory journal entry. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryJournalPutRequest". + */ +/** @experimental */ +export interface FactoryJournalPutRequest { + /** + * Factory run identifier. + */ + runId: string; + /** + * Namespaced journal key. + */ + key: string; + /** + * JSON result to memoize. + */ + resultJson: { + [k: string]: unknown | undefined; + }; +} +/** + * One ordered factory progress line. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryLogLine". + */ +/** @experimental */ +export interface FactoryLogLine { + /** + * Monotonic sequence number within the factory run. + */ + seq: number; + kind: FactoryLogLineKind; + /** + * Progress text. + */ + text: string; +} +/** + * Parameters for recording factory progress. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryLogRequest". + */ +/** @experimental */ +export interface FactoryLogRequest { + /** + * Factory run identifier. + */ + runId: string; + /** + * Ordered progress lines to append. + */ + lines: FactoryLogLine[]; +} +/** + * Parameters for invoking a registered factory. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryRunRequest". + */ +/** @experimental */ +export interface FactoryRunRequest { + /** + * Registered factory name. + */ + name: string; + /** + * Factory input value. + */ + args: { + [k: string]: unknown | undefined; + }; + options?: RunOptions; +} +/** + * Options controlling factory 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; +} +/** + * Complete current or terminal factory run envelope. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "FactoryRunResult". + */ +/** @experimental */ +export interface FactoryRunResult { + /** + * Factory run identifier. + */ + runId: string; + status: FactoryRunStatus; + /** + * Completed factory 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; + }; +} /** * Optional user prompt to combine with the fleet orchestration instructions. * @@ -16669,6 +17017,75 @@ export function createSessionRpc(connection: MessageConnection, sessionId: strin }, }, /** @experimental */ + factory: { + /** + * Runs a registered factory by name at the top level. + * + * @param params Parameters for invoking a registered factory. + * + * @returns Complete current or terminal factory run envelope. + */ + run: async (params: FactoryRunRequest): Promise => + connection.sendRequest("session.factory.run", { sessionId, ...params }), + /** + * Gets the current or settled envelope for a factory run. + * + * @param params Parameters for retrieving a factory run. + * + * @returns Complete current or terminal factory run envelope. + */ + getRun: async (params: FactoryGetRunRequest): Promise => + connection.sendRequest("session.factory.getRun", { sessionId, ...params }), + /** + * Requests cancellation of a factory run and returns its run envelope. + * + * @param params Parameters for cancelling a factory run. + * + * @returns Complete current or terminal factory run envelope. + */ + cancel: async (params: FactoryCancelRequest): Promise => + connection.sendRequest("session.factory.cancel", { sessionId, ...params }), + /** + * Records a batch of ordered factory progress lines. + * + * @param params Parameters for recording factory progress. + * + * @returns Acknowledgement that a factory request was accepted. + */ + log: async (params: FactoryLogRequest): Promise => + connection.sendRequest("session.factory.log", { sessionId, ...params }), + /** + * Runs one factory-scoped subagent and returns its result. + * + * @param params Parameters for one factory-scoped subagent call. + * + * @returns Result of one factory-scoped subagent call. + */ + agent: async (params: FactoryAgentRequest): Promise => + connection.sendRequest("session.factory.agent", { sessionId, ...params }), + /** @experimental */ + journal: { + /** + * Reads a memoized factory journal entry. + * + * @param params Parameters for reading a factory journal entry. + * + * @returns Result of reading a factory journal entry. + */ + get: async (params: FactoryJournalGetRequest): Promise => + connection.sendRequest("session.factory.journal.get", { sessionId, ...params }), + /** + * Stores a memoized factory journal entry. + * + * @param params Parameters for storing a factory journal entry. + * + * @returns Acknowledgement that a factory request was accepted. + */ + put: async (params: FactoryJournalPutRequest): Promise => + connection.sendRequest("session.factory.journal.put", { sessionId, ...params }), + }, + }, + /** @experimental */ model: { /** * Gets the currently selected model for the session. @@ -18156,6 +18573,27 @@ export interface ProviderTokenHandler { getToken(params: ProviderTokenAcquireRequest): Promise; } +/** Handler for `factory` client session API methods. */ +/** @experimental */ +export interface FactoryHandler { + /** + * Asks the owning extension connection to execute a registered factory closure. + * + * @param params Parameters sent to the owning extension to execute a factory closure. + * + * @returns Result returned by an extension factory closure. + */ + execute(params: FactoryExecuteRequest): Promise; + /** + * Asks the owning extension connection to abort a running factory cooperatively. + * + * @param params Parameters for cooperatively aborting a factory body. + * + * @returns Acknowledgement that a factory request was accepted. + */ + abort(params: FactoryAbortRequest): Promise; +} + /** Handler for `sessionFs` client session API methods. */ /** @experimental */ export interface SessionFsHandler { @@ -18287,6 +18725,7 @@ export interface CanvasHandler { /** All client session API handler groups. */ export interface ClientSessionApiHandlers { providerToken?: ProviderTokenHandler; + factory?: FactoryHandler; sessionFs?: SessionFsHandler; canvas?: CanvasHandler; } @@ -18306,6 +18745,16 @@ export function registerClientSessionApiHandlers( if (!handler) throw new Error(`No providerToken handler registered for session: ${params.sessionId}`); return handler.getToken(params); }); + connection.onRequest("factory.execute", async (params: FactoryExecuteRequest) => { + const handler = getHandlers(params.sessionId).factory; + if (!handler) throw new Error(`No factory handler registered for session: ${params.sessionId}`); + return handler.execute(params); + }); + connection.onRequest("factory.abort", async (params: FactoryAbortRequest) => { + const handler = getHandlers(params.sessionId).factory; + if (!handler) throw new Error(`No factory 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 9d3bdcd7f..16c512dff 100644 --- a/nodejs/src/index.ts +++ b/nodejs/src/index.ts @@ -12,6 +12,7 @@ export { CopilotClient } from "./client.js"; export { RuntimeConnection } from "./types.js"; export { BuiltInTools, ToolSet } from "./toolSet.js"; export { CopilotSession, type AssistantMessageEvent } from "./session.js"; +export { defineFactory, FactoryRunError } from "./factory.js"; export { Canvas, CanvasError, @@ -86,6 +87,8 @@ export type { LargeToolOutputConfig, MemoryConfiguration, UiInputOptions, + FactoryLimits, + FactoryMeta, MCPStdioServerConfig, MCPHTTPServerConfig, MCPServerConfig, @@ -158,3 +161,15 @@ export type { TypedSessionLifecycleHandler, ZodSchema, } from "./types.js"; +export type { + RunOptions, + SessionFactoryApi, + FactoryAgentOptions, + FactoryContext, + FactoryDefinition, + FactoryHandle, + FactoryJsonSchema, + FactoryPipelineStage, + FactoryStepOptions, + FactoryRunResult, +} from "./factory.js"; diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 1f71209de..6e5fae21d 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -15,6 +15,8 @@ import type { CanvasActionInvokeResult, CurrentToolMetadata, McpOauthPendingRequestResponse, + FactoryExecuteResult, + FactoryLogLine, } from "./generated/rpc.js"; import { type Canvas, CanvasError } from "./canvas.js"; import type { OpenCanvasInstance } from "./generated/rpc.js"; @@ -60,6 +62,15 @@ import type { UserInputRequest, UserInputResponse, } from "./types.js"; +import { + getFactoryDefinition, + FactoryRunError, + type RunOptions, + type SessionFactoryApi, + type FactoryContext, + type FactoryHandle, + type FactoryStepOptions, +} from "./factory.js"; /** * Convert a raw hook input received over the wire into its public-facing shape. @@ -94,6 +105,184 @@ function isOpenCanvasInstance(value: unknown): value is OpenCanvasInstance { ); } +const FACTORY_LOG_FLUSH_DELAY_MS = 10; +const MAX_FACTORY_FANOUT_ITEMS = 4096; + +function assertFactoryFanoutSize(kind: "parallel" | "pipeline", size: number): void { + if (size > MAX_FACTORY_FANOUT_ITEMS) { + throw new Error( + `${kind}() accepts at most ${MAX_FACTORY_FANOUT_ITEMS} items; got ${size}.` + ); + } +} + +async function runFactoryParallel( + thunks: Array<() => Promise | TResult> +): Promise> { + if (!Array.isArray(thunks)) { + throw new Error( + "parallel() expects an array of functions, not promises. Wrap each call: () => agent(...)" + ); + } + assertFactoryFanoutSize("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((error) => { + // Cancellation must propagate out of the combinator rather + // than be mapped to a successful `null`; otherwise an aborted + // run could be reported as completed. + if (isFactoryAbortError(error)) { + throw error; + } + return null; + }) + ) + ); +} + +async function runFactoryPipeline( + 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"); + } + assertFactoryFanoutSize("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 (error) { + // Propagate cancellation instead of mapping it to `null`, so + // an aborted stage does not let the run report success. + if (isFactoryAbortError(error)) { + throw error; + } + return null; + } + } + return previous; + }) + ); +} + +class FactoryProgressBuffer { + private nextSeq = 0; + private pending: FactoryLogLine[] = []; + private flushTimer?: ReturnType; + private flushTail: Promise = Promise.resolve(); + private flushError: unknown; + private flushFailed = false; + private closed = false; + + constructor(private readonly send: (lines: FactoryLogLine[]) => Promise) {} + + enqueue(kind: FactoryLogLine["kind"], text: string): void { + if (this.closed) { + throw new Error("Cannot log after the factory 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(() => {}); + }, FACTORY_LOG_FLUSH_DELAY_MS); + this.flushTimer.unref?.(); + } + + private clearFlushTimer(): void { + if (this.flushTimer !== undefined) { + clearTimeout(this.flushTimer); + this.flushTimer = undefined; + } + } +} + +async function awaitFactoryOperation( + operation: Promise, + signal: AbortSignal +): Promise { + throwIfFactoryAborted(signal); + let rejectAbort: ((reason?: unknown) => void) | undefined; + const onAbort = () => + rejectAbort?.(signal.reason ?? new DOMException("Factory 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 throwIfFactoryAborted(signal: AbortSignal): void { + if (signal.aborted) { + throw signal.reason ?? new DOMException("Factory run was aborted", "AbortError"); + } +} + +/** + * Whether an error represents factory run cancellation (an `AbortError`-shaped + * rejection from {@link awaitFactoryOperation}). Cancellation must bubble out of + * `parallel`/`pipeline` rather than being flattened into a `null` result. + */ +function isFactoryAbortError(error: unknown): boolean { + return ( + typeof error === "object" && + error !== null && + "name" in error && + (error as { name?: unknown }).name === "AbortError" + ); +} + /** Assistant message event - the final response from the assistant. */ export type AssistantMessageEvent = Extract; @@ -138,6 +327,8 @@ export class CopilotSession { private canvases: Map = new Map(); private bearerTokenProviders: Map = new Map(); private commandHandlers: Map = new Map(); + private factories = new Map>(); + private factoryAbortControllers = new Map(); private permissionHandler?: PermissionHandler; private mcpAuthHandler?: McpAuthHandler; private userInputHandler?: UserInputHandler; @@ -155,6 +346,44 @@ export class CopilotSession { /** @internal Client session API handlers, populated by CopilotClient during create/resume. */ clientSessionApis: ClientSessionApiHandlers = {}; + /** + * Friendly factory API for running registered factories by name or handle. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ + readonly factory: SessionFactoryApi = { + run: (async ( + nameOrHandle: string | FactoryHandle, + options?: RunOptions + ): Promise => { + const name = + typeof nameOrHandle === "string" + ? nameOrHandle + : getFactoryDefinition(nameOrHandle).meta.name; + const envelope = await this.rpc.factory.run({ + name, + args: (options?.args === undefined ? {} : options.args) as Parameters< + typeof this.rpc.factory.run + >[0]["args"], + options: { + background: options?.background, + resumeFromRunId: options?.resumeFromRunId, + }, + }); + + if (options?.background) { + return envelope; + } + if (envelope.status !== "completed") { + throw new FactoryRunError(envelope); + } + return envelope.result; + }) as SessionFactoryApi["run"], + getRun: (runId) => this.rpc.factory.getRun({ runId }), + cancel: (runId) => this.rpc.factory.cancel({ runId }), + }; + /** * Creates a new CopilotSession instance. * @@ -358,6 +587,11 @@ export class CopilotSession { this.autoModeSwitchHandler = undefined; this.commandHandlers.clear(); this.canvases.clear(); + this.factories.clear(); + for (const controller of this.factoryAbortControllers.values()) { + controller.abort(); + } + this.factoryAbortControllers.clear(); this.transformCallbacks?.clear(); } @@ -872,6 +1106,143 @@ export class CopilotSession { }; } + /** + * Registers factory closures and reverse-RPC handlers for this session. + * + * @param factories - Factory handles declared by the joining extension. + * @internal Called by the SDK when an extension joins a session. + */ + registerFactories(factories?: FactoryHandle[]): void { + this.factories.clear(); + if (!factories || factories.length === 0) { + delete this.clientSessionApis.factory; + return; + } + + for (const handle of factories) { + const definition = getFactoryDefinition(handle); + if (this.factories.has(definition.meta.name)) { + throw new Error( + `Duplicate factory name "${definition.meta.name}". Factory names must be unique within a joinSession call.` + ); + } + this.factories.set(definition.meta.name, definition); + } + + const self = this; + this.clientSessionApis.factory = { + async execute(params) { + const definition = self.factories.get(params.name); + if (!definition) { + const message = `No factory registered with name "${params.name}"`; + throw new ResponseError(ErrorCodes.InvalidParams, message, { + code: "factory_not_found", + name: params.name, + }); + } + + const controller = new AbortController(); + self.factoryAbortControllers.set(params.runId, controller); + const progress = new FactoryProgressBuffer(async (lines) => { + await self.rpc.factory.log({ runId: params.runId, lines }); + }); + try { + const context: FactoryContext = { + args: params.args, + session: self, + signal: controller.signal, + phase: (title: string) => { + throwIfFactoryAborted(controller.signal); + progress.enqueue("phase", title); + }, + log: (message: string) => { + throwIfFactoryAborted(controller.signal); + progress.enqueue("log", message); + }, + agent: async (prompt, options = {}) => { + await progress.flush(); + const response = await awaitFactoryOperation( + self.rpc.factory.agent({ + factoryRunId: params.runId, + prompt, + opts: { + label: options.label, + schema: options.schema, + model: options.model, + }, + }), + controller.signal + ); + return response.result ?? null; + }, + step: async ( + key: string, + producer: () => Promise | TResult, + options: FactoryStepOptions = {} + ): Promise => { + await progress.flush(); + if (options.volatile) { + return producer(); + } + const cached = await awaitFactoryOperation( + self.rpc.factory.journal.get({ + runId: params.runId, + key, + }), + controller.signal + ); + if (cached.hit) { + return cached.resultJson as TResult; + } + + // Producers are best-effort at-least-once across crashes or + // concurrent callers, so authors must make side effects idempotent. + const result = await producer(); + const resultJson = JSON.stringify(result); + if (resultJson === undefined) { + throw new Error( + `step("${key}") returned a value that is not JSON-serializable` + ); + } + await awaitFactoryOperation( + self.rpc.factory.journal.put({ + runId: params.runId, + key, + resultJson: result as Parameters< + typeof self.rpc.factory.journal.put + >[0]["resultJson"], + }), + controller.signal + ); + return result; + }, + parallel: runFactoryParallel, + pipeline: runFactoryPipeline, + factory: async () => { + throw new Error("nested factories are not supported"); + }, + }; + const result = await definition.run(context); + return { result } as FactoryExecuteResult; + } finally { + try { + await progress.close(); + } finally { + if (self.factoryAbortControllers.get(params.runId) === controller) { + self.factoryAbortControllers.delete(params.runId); + } + } + } + }, + async abort(params) { + self.factoryAbortControllers + .get(params.runId) + ?.abort(new DOMException("Factory run was aborted", "AbortError")); + return {}; + }, + }; + } + /** * Registers per-provider {@link BearerTokenProvider} callbacks for BYOK providers * configured with managed-identity / on-demand bearer-token auth. diff --git a/nodejs/src/types.ts b/nodejs/src/types.ts index fc0fa51d0..0311c70b6 100644 --- a/nodejs/src/types.ts +++ b/nodejs/src/types.ts @@ -1836,6 +1836,38 @@ export interface CanvasProviderIdentity { name?: string; } +/** + * Static resource ceilings declared by a factory before it runs. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryLimits { + /** Maximum number of factory subagents that may run concurrently. Must be positive when present. */ + maxConcurrentSubagents?: number; + /** Maximum total number of factory subagents that may be spawned. Must be positive when present. */ + maxTotalSubagents?: number; + /** Wall-clock timeout for the factory run, in milliseconds. Must be positive when present. */ + timeout?: number; +} + +/** + * Registration metadata for an extension-authored factory. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryMeta { + /** Stable factory name used for invocation. */ + name: string; + /** Human-readable factory description. */ + description: string; + /** Display metadata for the progress phases the factory may report. */ + phases: Array<{ title: string; detail?: string }>; + /** Optional resource ceilings presented to the user before execution. */ + limits?: FactoryLimits; +} + /** * Provider-scoped options for the Copilot API (CAPI). * diff --git a/nodejs/test/extension.test.ts b/nodejs/test/extension.test.ts index 1baa83a3a..e94ad2204 100644 --- a/nodejs/test/extension.test.ts +++ b/nodejs/test/extension.test.ts @@ -18,13 +18,13 @@ describe("joinSession", () => { it("defaults onPermissionRequest to no-result", async () => { process.env.SESSION_ID = "session-123"; - const resumeSession = vi - .spyOn(CopilotClient.prototype, "resumeSession") + const resumeForExtension = vi + .spyOn(CopilotClient.prototype, "resumeSessionForExtension") .mockResolvedValue({} as any); await joinSession({ tools: [] }); - const [, config] = resumeSession.mock.calls[0]!; + const [, config] = resumeForExtension.mock.calls[0]!; expect(config.onPermissionRequest).toBeDefined(); expect(config.onPermissionRequest).toBe(defaultJoinSessionPermissionHandler); const result = await Promise.resolve( @@ -36,13 +36,13 @@ describe("joinSession", () => { it("preserves an explicit onPermissionRequest handler", async () => { process.env.SESSION_ID = "session-123"; - const resumeSession = vi - .spyOn(CopilotClient.prototype, "resumeSession") + const resumeForExtension = vi + .spyOn(CopilotClient.prototype, "resumeSessionForExtension") .mockResolvedValue({} as any); await joinSession({ onPermissionRequest: approveAll, suppressResumeEvent: false }); - const [, config] = resumeSession.mock.calls[0]!; + const [, config] = resumeForExtension.mock.calls[0]!; expect(config.onPermissionRequest).toBe(approveAll); expect(config.suppressResumeEvent).toBe(false); }); diff --git a/nodejs/test/factory.test.ts b/nodejs/test/factory.test.ts new file mode 100644 index 000000000..1b05c88d9 --- /dev/null +++ b/nodejs/test/factory.test.ts @@ -0,0 +1,919 @@ +/*--------------------------------------------------------------------------------------------- + * 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 { + defineFactory, + FactoryRunError, + type FactoryAgentOptions, + type FactoryDefinition, +} from "../src/factory.js"; + +async function stopClient(client: CopilotClient): Promise { + await client.stop(); +} + +describe("factories", () => { + 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 factory without resource limits", + phases: [], + }; + const run = vi.fn(async ({ args }: { args: unknown }) => args); + const handle = defineFactory({ meta, run }); + + expect(handle.meta).toBe(meta); + expect(Object.isFrozen(handle)).toBe(true); + + const session = new CopilotSession("session-1", {} as never); + session.registerFactories([handle]); + const result = await session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: meta.name, + runId: "run-1", + args: { value: 42 }, + }); + + expect(run).toHaveBeenCalledOnce(); + expect(result).toEqual({ result: { value: 42 } }); + }); + + it("rejects duplicate factory names within a single registration", () => { + const run = async () => null; + const first = defineFactory({ + meta: { name: "dup", description: "first", phases: [] }, + run, + }); + const second = defineFactory({ + meta: { name: "dup", description: "second", phases: [] }, + run, + }); + + const session = new CopilotSession("session-dup", {} as never); + expect(() => session.registerFactories([first, second])).toThrow( + /Duplicate factory name "dup"/ + ); + }); + + 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 factory", + phases: [], + limits: { [field]: value }, + }, + run: async () => null, + } as FactoryDefinition; + + expect(() => defineFactory(definition)).toThrow(/must be a positive/); + }); + + it("rejects a timeout above the Node setTimeout ceiling", () => { + const definition = { + meta: { + name: "oversized-timeout", + description: "Factory with an out-of-range timeout", + phases: [], + limits: { timeout: 2_147_483_648 }, + }, + run: async () => null, + } as FactoryDefinition; + + expect(() => defineFactory(definition)).toThrow(/must not exceed/); + }); + + it("serializes only factory 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 factory = defineFactory({ + 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.factory + ).toBeDefined(); + return { sessionId: params.sessionId }; + } + throw new Error(`Unexpected method: ${method}`); + }); + + await client.resumeSessionForExtension( + "session-registration", + { onPermissionRequest: () => ({ kind: "approved" }) }, + [factory] + ); + + const payload = sendRequest.mock.calls.find( + ([method]) => method === "session.resume" + )![1] as { + factories: unknown[]; + }; + expect(payload.factories).toEqual([factory.meta]); + expect(payload.factories[0]).not.toHaveProperty("run"); + expect(JSON.stringify(payload.factories)).not.toContain("async"); + }); + + it("passes factories only through the extension join path", async () => { + process.env.SESSION_ID = "session-extension"; + const factory = defineFactory({ + 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({ factories: [factory] }); + + expect(resumeSessionForExtension).toHaveBeenCalledWith( + "session-extension", + expect.objectContaining({ suppressResumeEvent: true }), + [factory] + ); + }); + + it("builds the factory context with the unrestricted joined session identity", async () => { + process.env.SESSION_ID = "session-context"; + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.log") { + return {}; + } + if (method === "session.tasks.list") { + return { tasks: [] }; + } + throw new Error(`Unexpected method: ${method}`); + }); + const joinedSession = new CopilotSession("session-context", { sendRequest } as never); + const contextSeen = Promise.withResolvers<{ + args: unknown; + session: CopilotSession; + signal: AbortSignal; + }>(); + const factory = defineFactory({ + meta: { + name: "context", + description: "Context test", + phases: [], + }, + run: async (context) => { + contextSeen.resolve(context); + context.phase("A"); + context.log("hi"); + const tasks = await context.session.rpc.tasks.list(); + return { ok: true, taskCount: tasks.tasks.length }; + }, + }); + vi.spyOn(CopilotClient.prototype, "resumeSessionForExtension").mockImplementation( + async (_sessionId, _config, factories) => { + joinedSession.registerFactories(factories); + return joinedSession; + } + ); + + const joinSessionResult = await joinSession({ factories: [factory] }); + const executeResult = await joinSessionResult.clientSessionApis.factory!.execute({ + sessionId: joinSessionResult.sessionId, + name: "context", + runId: "run-context", + args: { value: 42 }, + }); + const context = await contextSeen.promise; + + expect(context.args).toEqual({ value: 42 }); + expect(context.session).toBe(joinSessionResult); + expect(context.session.rpc).toBe(joinSessionResult.rpc); + expect(context.signal).toBeInstanceOf(AbortSignal); + expect(executeResult).toEqual({ result: { ok: true, taskCount: 0 } }); + expect(sendRequest).toHaveBeenCalledWith("session.tasks.list", { + sessionId: joinSessionResult.sessionId, + }); + expect(sendRequest).toHaveBeenCalledWith("session.factory.log", { + sessionId: joinSessionResult.sessionId, + runId: "run-context", + lines: [ + { seq: 0, kind: "phase", text: "A" }, + { seq: 1, kind: "log", text: "hi" }, + ], + }); + }); + + it("rejects nested factories 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 factory = defineFactory({ + meta: { + name: "no-nesting", + description: "Nested factory rejection test", + phases: [], + }, + run: async (context) => context.factory("nested", { value: 42 }), + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "no-nesting", + runId: "run-no-nesting", + args: {}, + }) + ).rejects.toThrow("nested factories are not supported"); + expect(sendRequest).not.toHaveBeenCalled(); + }); + + it("flushes progress incrementally while a factory body is awaiting", async () => { + const sendRequest = vi.fn(async () => ({})); + const session = new CopilotSession("session-live-progress", { sendRequest } as never); + const body = Promise.withResolvers(); + const factory = defineFactory({ + meta: { + name: "live-progress", + description: "Incremental progress test", + phases: [], + }, + run: async ({ log }) => { + log("before await"); + await body.promise; + return "done"; + }, + }); + session.registerFactories([factory]); + + const execution = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "live-progress", + runId: "run-live-progress", + args: {}, + }); + await vi.waitFor(() => { + expect(sendRequest).toHaveBeenCalledWith("session.factory.log", { + sessionId: session.sessionId, + runId: "run-live-progress", + lines: [{ seq: 0, kind: "log", text: "before await" }], + }); + }); + + body.resolve(); + await expect(execution).resolves.toEqual({ result: "done" }); + }); + + it("calls factory.agent with the current run id and returns its text", async () => { + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.agent") { + return { result: "pong" }; + } + throw new Error(`Unexpected method: ${method}`); + }); + const session = new CopilotSession("session-agent", { sendRequest } as never); + const factory = defineFactory({ + 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 FactoryAgentOptions), + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "agent", + runId: "run-agent", + args: {}, + }) + ).resolves.toEqual({ result: "pong" }); + expect(sendRequest).toHaveBeenCalledWith("session.factory.agent", { + sessionId: session.sessionId, + factoryRunId: "run-agent", + prompt: "Reply with pong", + opts: { + label: "Pong helper", + model: "gpt-test", + schema: { type: "string" }, + }, + }); + }); + + it("runs a durable step once, serves cached null, and does not cache failures", async () => { + const journal = new Map(); + const sendRequest = vi.fn( + async (method: string, params: { key?: string; resultJson?: unknown }) => { + if (method === "session.factory.journal.get") { + return journal.has(params.key!) + ? { hit: true, resultJson: journal.get(params.key!) } + : { hit: false }; + } + if (method === "session.factory.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 factory = defineFactory({ + 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.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.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.factory.journal.put") + ).toHaveLength(2); + }); + + it("exposes factory 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.factory.getRun("run-read")).resolves.toEqual(envelope); + expect(sendRequest).toHaveBeenCalledWith("session.factory.getRun", { + sessionId: session.sessionId, + runId: "run-read", + }); + }); + + it("exposes factory 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.factory.cancel("run-cancel")).resolves.toEqual(envelope); + expect(sendRequest).toHaveBeenCalledWith("session.factory.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(); + const started: string[] = []; + const session = new CopilotSession("session-parallel", {} as never); + const factory = defineFactory({ + 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.registerFactories([factory]); + + let settled = false; + const execution = session.clientSessionApis + .factory!.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 factory = defineFactory({ + meta: { + name: "parallel-promises", + description: "Parallel misuse diagnostic", + phases: [], + }, + run: async ({ parallel }) => + parallel([Promise.resolve("already running")] as unknown as Array< + () => Promise + >), + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.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 factory = defineFactory({ + 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.registerFactories([factory]); + + const execution = session.clientSessionApis.factory!.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 factory = defineFactory({ + 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.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.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.factory.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 factory = defineFactory({ + 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.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.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 factory body throws", async () => { + const sendRequest = vi.fn(async () => ({})); + const session = new CopilotSession("session-throw-progress", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "throw-progress", + description: "Throwing progress test", + phases: [], + }, + run: async ({ log }) => { + log("before throw"); + throw new Error("body failed"); + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "throw-progress", + runId: "run-throw-progress", + args: {}, + }) + ).rejects.toThrow("body failed"); + expect(sendRequest).toHaveBeenCalledWith("session.factory.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 factory context", async () => { + const session = new CopilotSession("session-abort-signal", {} as never); + const signalSeen = Promise.withResolvers(); + const factory = defineFactory({ + 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.registerFactories([factory]); + + const execution = session.clientSessionApis.factory!.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.factory!.abort({ + sessionId: session.sessionId, + runId: "run-abort-signal", + }); + + expect(signal.aborted).toBe(true); + await expect(execution).resolves.toEqual({ result: true }); + }); + + it("rejects an in-flight runtime-backed await when factory.abort trips the signal", async () => { + const agentResponse = Promise.withResolvers<{ result: string }>(); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.agent") { + return agentResponse.promise; + } + return {}; + }); + const session = new CopilotSession("session-abort-await", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "abort-await", + description: "Abort an in-flight factory await", + phases: [], + }, + run: async ({ agent }) => agent("wait forever"), + }); + session.registerFactories([factory]); + + const execution = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "abort-await", + runId: "run-abort-await", + args: {}, + }); + await vi.waitFor(() => + expect(sendRequest).toHaveBeenCalledWith("session.factory.agent", expect.anything()) + ); + + await session.clientSessionApis.factory!.abort({ + sessionId: session.sessionId, + runId: "run-abort-await", + }); + + await expect(execution).rejects.toMatchObject({ name: "AbortError" }); + agentResponse.resolve({ result: "late" }); + }); + + it("propagates cancellation out of parallel/pipeline instead of mapping it to null", async () => { + const agentResponse = Promise.withResolvers<{ result: string }>(); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.agent") { + return agentResponse.promise; + } + return {}; + }); + const session = new CopilotSession("session-abort-parallel", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "abort-parallel", + description: "Cancellation must bubble out of a combinator", + phases: [], + }, + // If the combinator swallowed the AbortError to null, this run would + // resolve successfully with [null] despite the run being cancelled. + run: async ({ agent, parallel }) => parallel([() => agent("wait forever")]), + }); + session.registerFactories([factory]); + + const execution = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "abort-parallel", + runId: "run-abort-parallel", + args: {}, + }); + await vi.waitFor(() => + expect(sendRequest).toHaveBeenCalledWith("session.factory.agent", expect.anything()) + ); + + await session.clientSessionApis.factory!.abort({ + sessionId: session.sessionId, + runId: "run-abort-parallel", + }); + + await expect(execution).rejects.toMatchObject({ name: "AbortError" }); + agentResponse.resolve({ result: "late" }); + }); + + it("dispatches factory.execute to the registered factory selected by name", async () => { + const firstRun = vi.fn(async () => ({ selected: "first" })); + const secondRun = vi.fn(async ({ args, log }) => { + log("executing"); + return { selected: "second", echoed: args }; + }); + const firstFactory = defineFactory({ + meta: { + name: "first", + description: "First factory", + phases: [], + }, + run: firstRun, + }); + const secondFactory = defineFactory({ + meta: { + name: "second", + description: "Second factory", + phases: [], + }, + run: secondRun, + }); + const session = new CopilotSession("session-execute", { + sendRequest: vi.fn(async () => ({})), + } as never); + session.registerFactories([firstFactory, secondFactory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "second", + runId: "run-echo", + args: { message: "hello" }, + }) + ).resolves.toEqual({ + result: { selected: "second", echoed: { message: "hello" } }, + }); + expect(firstRun).not.toHaveBeenCalled(); + expect(secondRun).toHaveBeenCalledOnce(); + + const error = await session.clientSessionApis + .factory!.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: "factory_not_found", + name: "missing", + }); + }); + + it("runs factories by name or handle and unwraps only foreground results", async () => { + const factory = defineFactory({ + 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.factory.run("by-name", { args: { value: 1 } })).resolves.toEqual({ + name: "by-name", + }); + await expect(session.factory.run(factory)).resolves.toEqual({ + name: "friendly-run", + }); + await expect(session.factory.run("background", { background: true })).resolves.toEqual({ + runId: "run-background", + status: "running", + }); + + expect(sendRequest).toHaveBeenNthCalledWith(1, "session.factory.run", { + sessionId: session.sessionId, + name: "by-name", + args: { value: 1 }, + options: { background: undefined, resumeFromRunId: undefined }, + }); + expect(sendRequest).toHaveBeenNthCalledWith(2, "session.factory.run", { + sessionId: session.sessionId, + name: "friendly-run", + args: {}, + options: { background: undefined, resumeFromRunId: undefined }, + }); + }); + + it("throws FactoryRunError with the full foreground envelope", async () => { + const envelope = { + runId: "run-error", + status: "error" as const, + error: "factory failed", + snapshot: { completed: 1 }, + }; + const session = new CopilotSession("session-error", { + sendRequest: vi.fn(async () => envelope), + } as never); + + const error = await session.factory.run("failing").catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(FactoryRunError); + expect((error as FactoryRunError).envelope).toBe(envelope); + }); +});