From 76a7fa803e6603efc3d49985c086a7838959dc92 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Thu, 3 Sep 2026 12:48:34 -0700 Subject: [PATCH 1/3] [Core] Add Factory Pause Checkpoints Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> From bed241edb4f9e8f84a63a77353d194f90c2637ff Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Thu, 3 Sep 2026 13:00:19 -0700 Subject: [PATCH 2/3] [Core] Expose SDK Pause Checkpoints Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- nodejs/src/factory.ts | 37 ++++- nodejs/src/index.ts | 1 + nodejs/src/session.ts | 79 ++++++++-- nodejs/test/factory.test.ts | 296 ++++++++++++++++++++++++++++++++++++ 4 files changed, 398 insertions(+), 15 deletions(-) diff --git a/nodejs/src/factory.ts b/nodejs/src/factory.ts index 6212f462b4..dfef30b9c4 100644 --- a/nodejs/src/factory.ts +++ b/nodejs/src/factory.ts @@ -47,13 +47,15 @@ export type FactoryRunsPage = FactoryListRunsResult; /** * Run statuses a factory run can no longer move away from. * - * A run is either still in flight (`pending`, `running`) or settled into one of - * these four. Terminal state is final: once written it is never reopened, so a - * caller that observes one of these can stop watching the run. + * A run is either still in flight (`pending`, `running`) or its current attempt + * has settled into one of these states. A paused run can later start a new + * attempt under the same run ID, but callers waiting on the current attempt can + * stop watching once they observe it. */ const FACTORY_TERMINAL_STATUSES: ReadonlySet = new Set([ "completed", "halted", + "paused", "cancelled", "error", ]); @@ -139,6 +141,22 @@ export interface FactoryStepOptions { volatile?: boolean; } +/** + * Per-invocation factory resource ceiling overrides. + * + * An omitted field preserves the existing/default ceiling, a number replaces + * it, and `null` explicitly makes that dimension unlimited. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface FactoryLimitOverrides { + maxConcurrentSubagents?: number | null; + maxTotalSubagents?: number | null; + maxAiCredits?: number | null; + timeoutSeconds?: number | null; +} + /** * One stage in a per-item factory pipeline. * @@ -168,6 +186,13 @@ export interface FactoryContext { producer: () => Promise | JsonValue, options?: FactoryStepOptions ): Promise; + /** + * Pause this run at a durable, one-shot checkpoint. + * + * The first attempt to reach a key pauses and aborts cooperatively. A + * resumed attempt returns from the same key and continues. + */ + pause(key: string): Promise; /** * Run thunks concurrently and await all of them. * @@ -259,7 +284,7 @@ export interface RunOptions { /** Input surfaced as `context.args`. */ args?: TArgs; /** Optional per-invocation resource ceiling overrides. */ - limits?: FactoryLimits; + limits?: FactoryLimitOverrides; /** Whether to notify the originating session when the factory completes. */ notifyOnComplete?: boolean; /** Whether to emit factory phase names to the session transcript. */ @@ -280,7 +305,7 @@ export interface RunOptions { */ export interface ResumeOptions { /** Optional per-invocation resource ceiling overrides. */ - limits?: FactoryLimits; + limits?: FactoryLimitOverrides; /** Whether to notify the originating session when the factory completes. */ notifyOnComplete?: boolean; /** Whether to emit factory phase names to the session transcript. */ @@ -375,6 +400,8 @@ export interface SessionFactoryApi { runId: string, options?: Omit ): Promise; + /** Pause a running factory and return its settled envelope. */ + pause(runId: string): Promise; /** Cancel a factory run and return its terminal envelope. */ cancel(runId: string): Promise; } diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts index 6251df4fc7..8a5a730b5a 100644 --- a/nodejs/src/index.ts +++ b/nodejs/src/index.ts @@ -203,6 +203,7 @@ export type { export type { RunOptions, ResumeOptions, + FactoryLimitOverrides, FactoryResumeErrorCode, SessionFactoryApi, FactoryAgentOptions, diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 4c2be14299..187dcaea7c 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -10,7 +10,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import type { MessageConnection } from "vscode-jsonrpc/node.js"; import { ConnectionError, ErrorCodes, ResponseError } from "vscode-jsonrpc/node.js"; -import { createSessionRpc } from "./generated/rpc.js"; +import { createInternalSessionRpc, createSessionRpc } from "./generated/rpc.js"; import type { ClientSessionApiHandlers, CanvasActionInvokeResult, @@ -108,16 +108,32 @@ function copyDefinedFactoryAgentOption( } } -const factoryExecutionStore = new AsyncLocalStorage<{ active: boolean }>(); +type FactoryExecutionContext = { + active: boolean; + helperScope?: "parallel" | "pipeline"; +}; + +const factoryExecutionStore = new AsyncLocalStorage(); function throwIfFactoryExecutionIsActive(): void { if (factoryExecutionStore.getStore()?.active) { throw new Error( - "factory.run and factory.resume are not allowed while a factory body is running on this call path." + "factory.run and factory.resume, and factory.pause are not allowed while a factory body is running on this call path." ); } } +function runInFactoryHelperScope( + helperScope: "parallel" | "pipeline", + callback: () => Promise | TResult +): Promise | TResult { + const current = factoryExecutionStore.getStore(); + return factoryExecutionStore.run( + { active: current?.active ?? false, helperScope }, + callback + ); +} + /** * Convert a raw hook input received over the wire into its public-facing shape. * This deserializes the numeric Unix-ms `timestamp` field on BaseHookInput @@ -188,7 +204,7 @@ async function runFactoryParallel( return Promise.all( thunks.map((thunk) => Promise.resolve() - .then(() => thunk()) + .then(() => runInFactoryHelperScope("parallel", thunk)) .catch((error) => { // Cancellation and hard runtime failures must propagate out // of the combinator rather than be mapped to a successful @@ -220,7 +236,9 @@ async function runFactoryPipeline( let previous = item; for (const stage of stages) { try { - previous = await stage(previous, item, index); + previous = await runInFactoryHelperScope("pipeline", () => + stage(previous, item, index) + ); } catch (error) { // Propagate cancellation and hard runtime failures instead // of mapping them to `null`, so an aborted stage — or one @@ -437,6 +455,7 @@ export class CopilotSession { private hooks?: SessionHooks; private transformCallbacks?: Map; private _rpc: ReturnType | null = null; + private _internalRpc: ReturnType | null = null; private traceContextProvider?: TraceContextProvider; private readonly managedSettingsEnabled: boolean; private _capabilities: SessionCapabilities = {}; @@ -517,6 +536,10 @@ export class CopilotSession { getRunDetail: (runId) => this.rpc.factory.getRunDetail({ runId }), getRunProgress: (runId, options = {}) => this.rpc.factory.getRunProgress({ runId, ...options }), + pause: async (runId) => { + throwIfFactoryExecutionIsActive(); + return this.rpc.factory.pause({ runId }); + }, cancel: async (runId) => this.rpc.factory.cancel({ runId }), }; @@ -654,6 +677,14 @@ export class CopilotSession { return this._rpc; } + /** @internal */ + private get internalRpc(): ReturnType { + if (!this._internalRpc) { + this._internalRpc = createInternalSessionRpc(this.connection, this.sessionId); + } + return this._internalRpc; + } + /** * Path to the session workspace directory when infinite sessions are enabled. * Contains checkpoints/, plan.md, and files/ subdirectories. @@ -1561,6 +1592,36 @@ export class CopilotSession { ); return result; }, + pause: async (key: string): Promise => { + if (typeof key !== "string" || key.length === 0) { + throw new Error("Factory pause checkpoint key must not be empty"); + } + const helperScope = factoryExecutionStore.getStore()?.helperScope; + if (helperScope !== undefined) { + throw new Error( + `Factory pause checkpoints are not allowed inside ${helperScope}() branches` + ); + } + await progress.flush(); + const response = await awaitFactoryOperation( + () => + self.internalRpc.factory.pauseAtCheckpoint({ + runId: params.runId, + executionToken: params.executionToken, + key, + }), + controller.signal + ); + switch (response.action) { + case "continue": + return; + case "pause": + await awaitFactoryOperation( + () => new Promise(() => {}), + controller.signal + ); + } + }, parallel: runFactoryParallel, pipeline: runFactoryPipeline, factory: async () => { @@ -1596,11 +1657,9 @@ export class CopilotSession { }, async abort(params) { const controllersForRun = self.factoryAbortControllers.get(params.runId); - if (controllersForRun !== undefined) { - const reason = new DOMException("Factory run was aborted", "AbortError"); - for (const controller of controllersForRun.values()) { - controller.abort(reason); - } + const controller = controllersForRun?.get(params.executionToken); + if (controller !== undefined) { + controller.abort(new DOMException("Factory run was aborted", "AbortError")); } return {}; }, diff --git a/nodejs/test/factory.test.ts b/nodejs/test/factory.test.ts index c9b8f65074..4270f6fd70 100644 --- a/nodejs/test/factory.test.ts +++ b/nodejs/test/factory.test.ts @@ -1407,6 +1407,7 @@ describe("factories", () => { await session.clientSessionApis.factory!.abort({ sessionId: session.sessionId, runId, + executionToken: "execution-token", }); await step( "volatile", @@ -1571,6 +1572,197 @@ describe("factories", () => { }); }); + it("exposes guarded public pause and prevents factory bodies from bypassing ctx.pause", async () => { + const paused = { runId: "run-pause", status: "paused" as const }; + const sendRequest = vi.fn(async () => paused); + const session = new CopilotSession("session-pause", { sendRequest } as never); + + await expect(session.factory.pause("run-pause")).resolves.toEqual(paused); + expect(sendRequest).toHaveBeenCalledWith("session.factory.pause", { + sessionId: session.sessionId, + runId: "run-pause", + }); + + const factory = defineFactory({ + meta: { + name: "pause-bypass", + description: "Public pause cannot bypass context restrictions", + phases: [], + }, + run: ({ runId, session: factorySession }) => factorySession.factory.pause(runId), + }); + session.registerFactories([factory]); + sendRequest.mockClear(); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "pause-bypass", + runId: "run-pause-bypass", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toThrow("factory.pause"); + expect(sendRequest).not.toHaveBeenCalled(); + }); + + it("rejects an empty pause checkpoint key before RPC", async () => { + const sendRequest = vi.fn(); + const session = new CopilotSession("session-empty-pause-key", { + sendRequest, + } as never); + const factory = defineFactory({ + meta: { + name: "empty-pause-key", + description: "Empty pause key rejection", + phases: [], + }, + run: ({ pause }) => pause(""), + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "empty-pause-key", + runId: "run-empty-pause-key", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toThrow("must not be empty"); + expect(sendRequest).not.toHaveBeenCalled(); + }); + + it("waits for cooperative abort when a pause checkpoint returns pause", async () => { + const checkpointRequested = Promise.withResolvers(); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.pauseAtCheckpoint") { + checkpointRequested.resolve(); + return { action: "pause" }; + } + throw new Error(`Unexpected method: ${method}`); + }); + const session = new CopilotSession("session-checkpoint-pause", { + sendRequest, + } as never); + const factory = defineFactory({ + meta: { + name: "checkpoint-pause", + description: "Pause checkpoint abort behavior", + phases: [], + }, + run: ({ pause }) => pause("review-ready"), + }); + session.registerFactories([factory]); + + let settled = false; + const execution = session.clientSessionApis.factory! + .execute({ + sessionId: session.sessionId, + name: "checkpoint-pause", + runId: "run-checkpoint-pause", + executionToken: "execution-token", + args: {}, + }) + .finally(() => { + settled = true; + }); + await checkpointRequested.promise; + await Promise.resolve(); + expect(settled).toBe(false); + + await session.clientSessionApis.factory!.abort({ + sessionId: session.sessionId, + runId: "run-checkpoint-pause", + executionToken: "execution-token", + }); + await expect(execution).rejects.toMatchObject({ name: "AbortError" }); + }); + + it("returns void and continues when a pause checkpoint returns continue", async () => { + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.pauseAtCheckpoint") { + return { action: "continue" }; + } + throw new Error(`Unexpected method: ${method}`); + }); + const session = new CopilotSession("session-checkpoint-continue", { + sendRequest, + } as never); + const factory = defineFactory({ + meta: { + name: "checkpoint-continue", + description: "Continue checkpoint behavior", + phases: [], + }, + run: async ({ pause }) => { + const result = await pause("review-ready"); + return result === undefined ? "continued" : "unexpected"; + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "checkpoint-continue", + runId: "run-checkpoint-continue", + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({ result: "continued" }); + expect(sendRequest).toHaveBeenCalledWith("session.factory.pauseAtCheckpoint", { + sessionId: session.sessionId, + runId: "run-checkpoint-continue", + executionToken: "execution-token", + key: "review-ready", + }); + }); + + it.each(["parallel", "pipeline"] as const)( + "rejects pause checkpoints inside %s helper branches before RPC", + async (helper) => { + const sendRequest = vi.fn(); + const session = new CopilotSession(`session-pause-${helper}`, { + sendRequest, + } as never); + const factory = defineFactory({ + meta: { + name: `pause-${helper}`, + description: "Pause helper-scope rejection", + phases: [], + }, + run: async ({ pause, parallel, pipeline }) => { + const attempt = async () => { + try { + await pause("review-ready"); + return "unexpected"; + } catch (error) { + return (error as Error).message; + } + }; + return helper === "parallel" + ? parallel([attempt]) + : pipeline(["item"], attempt); + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: `pause-${helper}`, + runId: `run-pause-${helper}`, + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({ + result: [`Factory pause checkpoints are not allowed inside ${helper}() branches`], + }); + expect(sendRequest).not.toHaveBeenCalled(); + } + ); + it("runs parallel as a barrier and maps a throwing thunk to null", async () => { const first = Promise.withResolvers(); const second = Promise.withResolvers(); @@ -1981,6 +2173,7 @@ describe("factories", () => { await session.clientSessionApis.factory!.abort({ sessionId: session.sessionId, runId: "run-abort-signal", + executionToken: "execution-token", }); expect(signal.aborted).toBe(true); @@ -2020,12 +2213,75 @@ describe("factories", () => { await session.clientSessionApis.factory!.abort({ sessionId: session.sessionId, runId: "run-abort-await", + executionToken: "execution-token", }); await expect(execution).rejects.toMatchObject({ name: "AbortError" }); agentResponse.resolve({ result: "late" }); }); + it("ignores a late abort for an older execution token with the same run id", async () => { + const oldAgent = Promise.withResolvers<{ result: string }>(); + const currentAgent = Promise.withResolvers<{ result: string }>(); + const sendRequest = vi.fn( + async (method: string, params: { executionToken?: string }) => { + if (method !== "session.factory.agent") { + return {}; + } + return params.executionToken === "old-token" + ? oldAgent.promise + : currentAgent.promise; + } + ); + const session = new CopilotSession("session-token-scoped-abort", { + sendRequest, + } as never); + const signals: AbortSignal[] = []; + const factory = defineFactory({ + meta: { + name: "token-scoped-abort", + description: "Abort only the matching execution attempt", + phases: [], + }, + run: async ({ agent, signal }) => { + signals.push(signal); + return agent("wait"); + }, + }); + session.registerFactories([factory]); + + const oldExecution = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "token-scoped-abort", + runId: "shared-run", + executionToken: "old-token", + args: {}, + }); + const currentExecution = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "token-scoped-abort", + runId: "shared-run", + executionToken: "current-token", + args: {}, + }); + await vi.waitFor(() => + expect(sendRequest).toHaveBeenCalledTimes(2) + ); + + await session.clientSessionApis.factory!.abort({ + sessionId: session.sessionId, + runId: "shared-run", + executionToken: "old-token", + }); + expect(signals[0].aborted).toBe(true); + expect(signals[1].aborted).toBe(false); + await expect(oldExecution).rejects.toMatchObject({ name: "AbortError" }); + + currentAgent.resolve({ result: "current completed" }); + await expect(currentExecution).resolves.toEqual({ result: "current completed" }); + oldAgent.resolve({ result: "late old result" }); + }); + it.each(["parallel", "pipeline"] as const)( "propagates cancellation out of %s instead of mapping it to null", async (combinator) => { @@ -2066,6 +2322,7 @@ describe("factories", () => { await session.clientSessionApis.factory!.abort({ sessionId: session.sessionId, runId: `run-abort-${combinator}`, + executionToken: "execution-token", }); await expect(execution).rejects.toMatchObject({ name: "AbortError" }); @@ -2213,6 +2470,44 @@ describe("factories", () => { }); }); + it("preserves omitted, numeric, and explicit unlimited invocation limit overrides", async () => { + const sendRequest = vi.fn(async (method: string) => + method === "session.factory.resume" + ? { + factoryName: "stored-name", + run: { runId: "run-limits", status: "completed" }, + } + : { runId: "run-limits", status: "completed" } + ); + const session = new CopilotSession("session-limit-overrides", { + sendRequest, + } as never); + + await session.factory.run("omitted"); + await session.factory.run("numeric", { + limits: { maxTotalSubagents: 12 }, + }); + await session.factory.run("unlimited", { + limits: { maxTotalSubagents: null }, + }); + await session.factory.resume("run-limits", { + limits: { timeoutSeconds: null }, + }); + + expect(sendRequest.mock.calls[0][1]).toMatchObject({ + options: { limits: undefined }, + }); + expect(sendRequest.mock.calls[1][1]).toMatchObject({ + options: { limits: { maxTotalSubagents: 12 } }, + }); + expect(sendRequest.mock.calls[2][1]).toMatchObject({ + options: { limits: { maxTotalSubagents: null } }, + }); + expect(sendRequest.mock.calls[3][1]).toMatchObject({ + limits: { timeoutSeconds: null }, + }); + }); + it("returns the full envelope for a failed foreground run", async () => { const envelope = { runId: "run-error", @@ -2289,6 +2584,7 @@ describe("factory run settlement", () => { ["completed", true], ["error", true], ["halted", true], + ["paused", true], ["cancelled", true], ["pending", false], ["running", false], From 295bd38fad975829aefddc7227c29e19904764a2 Mon Sep 17 00:00:00 2001 From: Matthew Rayermann Date: Thu, 3 Sep 2026 14:27:15 -0700 Subject: [PATCH 3/3] [Core] Clean Up Factory Pause Checkpoints Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- nodejs/src/factory.ts | 2 +- nodejs/src/session.ts | 5 +---- nodejs/test/factory.test.ts | 22 ++++++++-------------- 3 files changed, 10 insertions(+), 19 deletions(-) diff --git a/nodejs/src/factory.ts b/nodejs/src/factory.ts index dfef30b9c4..61051f30e8 100644 --- a/nodejs/src/factory.ts +++ b/nodejs/src/factory.ts @@ -14,7 +14,7 @@ import type { } from "./generated/rpc.js"; import type { ContextTier } from "./generated/session-events.js"; import type { CopilotSession } from "./session.js"; -import type { FactoryLimits, FactoryMeta } from "./types.js"; +import type { FactoryMeta } from "./types.js"; export type { FactoryRunResult }; export type { diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 187dcaea7c..4336c9272d 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -128,10 +128,7 @@ function runInFactoryHelperScope( callback: () => Promise | TResult ): Promise | TResult { const current = factoryExecutionStore.getStore(); - return factoryExecutionStore.run( - { active: current?.active ?? false, helperScope }, - callback - ); + return factoryExecutionStore.run({ active: current?.active ?? false, helperScope }, callback); } /** diff --git a/nodejs/test/factory.test.ts b/nodejs/test/factory.test.ts index 4270f6fd70..519bee30ea 100644 --- a/nodejs/test/factory.test.ts +++ b/nodejs/test/factory.test.ts @@ -1656,8 +1656,8 @@ describe("factories", () => { session.registerFactories([factory]); let settled = false; - const execution = session.clientSessionApis.factory! - .execute({ + const execution = session.clientSessionApis + .factory!.execute({ sessionId: session.sessionId, name: "checkpoint-pause", runId: "run-checkpoint-pause", @@ -2223,16 +2223,12 @@ describe("factories", () => { it("ignores a late abort for an older execution token with the same run id", async () => { const oldAgent = Promise.withResolvers<{ result: string }>(); const currentAgent = Promise.withResolvers<{ result: string }>(); - const sendRequest = vi.fn( - async (method: string, params: { executionToken?: string }) => { - if (method !== "session.factory.agent") { - return {}; - } - return params.executionToken === "old-token" - ? oldAgent.promise - : currentAgent.promise; + const sendRequest = vi.fn(async (method: string, params: { executionToken?: string }) => { + if (method !== "session.factory.agent") { + return {}; } - ); + return params.executionToken === "old-token" ? oldAgent.promise : currentAgent.promise; + }); const session = new CopilotSession("session-token-scoped-abort", { sendRequest, } as never); @@ -2264,9 +2260,7 @@ describe("factories", () => { executionToken: "current-token", args: {}, }); - await vi.waitFor(() => - expect(sendRequest).toHaveBeenCalledTimes(2) - ); + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(2)); await session.clientSessionApis.factory!.abort({ sessionId: session.sessionId,