diff --git a/nodejs/docs/factories.md b/nodejs/docs/factories.md index 23b9d0fed3..f07230a839 100644 --- a/nodejs/docs/factories.md +++ b/nodejs/docs/factories.md @@ -62,19 +62,20 @@ Validation covers the model's `run_factory` path only. An extension calling `ses The `run()` context provides: -- `ctx.runId`: Stable ID reused across resumed attempts. -- `ctx.args`: Invocation arguments, forwarded verbatim. When the caller omits `args`, this is `{}` rather than `undefined`. -- `ctx.agent(prompt, options?)`: Runs one factory-owned subagent. Options are exactly `label`, `schema`, `model`, `agent`, `reasoningEffort`, and `contextTier`. See [Subagent calls](#subagent-calls). -- `ctx.parallel(thunks)`: Runs thunks concurrently and awaits all of them (a barrier). A thunk that throws becomes `null` in the result array, so one failed item does not lose the rest. Cancellation and hard runtime failures (`ResponseError`, `ConnectionError`) are the exception — those propagate and reject the whole call, because they mean the run itself is in trouble rather than one item having failed. Handle them at run level; do not assume every failure arrives as a `null`. Rejects above 4096 items. -- `ctx.pipeline(items, ...stages)`: Flows each item through every stage without a barrier between stages, so one item can be in a later stage while another is still in an earlier one. Each stage is called as `(previous, item, index)`, where `previous` is the prior stage's result and `item` is the original input. A stage that throws drops that item to `null` and skips its remaining stages, with the same exception for cancellation and hard runtime failures. Rejects above 4096 items. -- `ctx.phase(title)`: Starts a named progress phase. This sets a single run-global value, so calling it from inside concurrent `parallel`/`pipeline` stages races. Call it at run-level transitions and distinguish concurrent work by `label` instead. -- `ctx.log(message)`: Appends a progress line. When a factory bounds its own coverage (top-N, sampling), log what was dropped. -- `ctx.step(key, producer, options?)`: Journals the producer's JSON result under a stable key so a resume replays it without re-running the producer. A journaled (default) producer must return a JSON-serializable value; `undefined` or a non-JSON value is rejected. Pass `{ volatile: true }` to bypass the journal and run the producer every time. +* `ctx.runId`: Stable ID reused across resumed attempts. +* `ctx.args`: Invocation arguments, forwarded verbatim. When the caller omits `args`, this is `{}` rather than `undefined`. +* `ctx.agent(prompt, options?)`: Runs one factory-owned subagent. Options are exactly `label`, `schema`, `model`, `agent`, `reasoningEffort`, and `contextTier`. See [Subagent calls](#subagent-calls). +* `ctx.parallel(thunks)`: Runs thunks concurrently and awaits all of them (a barrier). A thunk that throws becomes `null` in the result array, so one failed item does not lose the rest. Cancellation and hard runtime failures (`ResponseError`, `ConnectionError`) are the exception — those propagate and reject the whole call, because they mean the run itself is in trouble rather than one item having failed. Handle them at run level; do not assume every failure arrives as a `null`. Rejects above 4096 items. +* `ctx.pipeline(items, ...stages)`: Flows each item through every stage without a barrier between stages, so one item can be in a later stage while another is still in an earlier one. Each stage is called as `(previous, item, index)`, where `previous` is the prior stage's result and `item` is the original input. A stage that throws drops that item to `null` and skips its remaining stages, with the same exception for cancellation and hard runtime failures. Rejects above 4096 items. +* `ctx.phase(title)`: Starts a named progress phase. This sets a single run-global value, so calling it from inside concurrent `parallel`/`pipeline` stages races. Call it at run-level transitions and distinguish concurrent work by `label` instead. +* `ctx.log(message)`: Appends a progress line. When a factory bounds its own coverage (top-N, sampling), log what was dropped. +* `ctx.step(key, producer, options?)`: Journals the producer's JSON result under a stable key so a resume replays it without re-running the producer. A journaled (default) producer must return a JSON-serializable value; `undefined` or a non-JSON value is rejected. Pass `{ volatile: true }` to bypass the journal and run the producer every time. The key is the *sole* identity: neither the producer body nor its inputs contribute to it. A resume replays the cached value for a matching key even if the producer has since changed, so version the key (`"scan-v2"`) whenever its inputs or meaning change. Journaled producers are best-effort at-least-once and may run again across crashes or concurrent same-key callers, so keep side effects idempotent. -- `ctx.session`: The session returned by `joinSession`. It refuses calls that start or resume a factory run. Call `extensions_manage` with `operation: "guide"` to read more about the session APIs. -- `ctx.signal`: Cooperative cancellation signal for extension work and subprocesses. -- `ctx.factory(...)`: Always rejects because nested factories are not supported. +* `ctx.pause(key)`: Pauses at a durable, one-shot checkpoint. The first attempt records the checkpoint, pauses, and throws `AbortError` after cooperative cancellation. When the run resumes, the factory starts again and the same checkpoint returns so execution can continue. Call it only from the main factory flow, not inside `ctx.parallel()` or `ctx.pipeline()`. +* `ctx.session`: The session returned by `joinSession`. It refuses calls that start, resume, or pause a factory run. Call `extensions_manage` with `operation: "guide"` to read more about the session APIs. +* `ctx.signal`: Cooperative cancellation signal for extension work and subprocesses. +* `ctx.factory(...)`: Always rejects because nested factories are not supported. Factory-owned subagents are intentionally hidden from `read_agent` and `write_agent`. Use the factory observability APIs instead. @@ -157,7 +158,7 @@ session.factory.run( name: string, options?: { args?: JsonValue; - limits?: FactoryLimits; + limits?: FactoryLimitOverrides; notifyOnComplete?: boolean; logPhaseNames?: boolean; }, @@ -180,7 +181,7 @@ The signature is: session.factory.resume( runId: string, options?: { - limits?: FactoryLimits; + limits?: FactoryLimitOverrides; notifyOnComplete?: boolean; logPhaseNames?: boolean; }, @@ -189,15 +190,31 @@ session.factory.resume( Set `notifyOnComplete` to `true` for factories that are likely to be invoked by an agent, so the originating session is notified when the factory completes. Set it to `false` for factories intended to be invoked programmatically, where the caller awaits the result directly. Set `logPhaseNames` to emit factory phase names to the session transcript. Both options apply to new and resumed runs. -Both resolve with the run envelope (`FactoryRunResult`) for **every** outcome — `completed`, `error`, `halted`, and `cancelled` alike. Inspect `status` and read `result` only when the run completed; a limit breach carries a typed `failure`. SDK-initiated `run` and `resume` do not request permission, so they have no declined outcome. The model's `run_factory` tool requests permission before the durable row exists; declining it creates no run row. An SDK-initiated run is refused only when the session already has its maximum number of active top-level runs. Pre-execution resume failures throw `FactoryResumeError`, whose `code` is one of `not_found`, `non_resumable`, `already_active`, `factory_already_running`, `factory_limits_invalid`, `factory_session_disposed`, `factory_storage_unavailable`, or `factory_storage_corrupt`. +Both resolve with the run envelope (`FactoryRunResult`) for **every** outcome—`completed`, `error`, `halted`, `paused`, and `cancelled` alike. Inspect `status` and read `result` only when the run completed; a limit breach carries a typed `failure`. A `paused` envelope means that the current attempt settled, not that the durable run is permanently finished. Resume the same run ID to start another attempt with its journal and accounting intact. SDK-initiated `run` and `resume` do not request permission, so they have no declined outcome. The model's `run_factory` tool requests permission before the durable row exists; declining it creates no run row. An SDK-initiated run is refused only when the session already has its maximum number of active top-level runs. Pre-execution resume failures throw `FactoryResumeError`, whose `code` is one of `not_found`, `non_resumable`, `already_active`, `factory_already_running`, `factory_limits_invalid`, `factory_session_disposed`, `factory_storage_unavailable`, or `factory_storage_corrupt`. An agent that no longer has a prior run's ID in context can recover it with `factories_manage` and `operation: "runs"`, which lists the session's factory runs with their IDs and statuses. This matters for resume: a run that reached a limit keeps its journal, so resuming it replays completed work for free, while restarting it from scratch pays for that work twice. +Pause a running attempt from outside its factory body: + +```ts +const paused = await session.factory.pause(runId); +``` + +Inside a factory body, use a durable checkpoint instead: + +```ts +await ctx.step("prepare", prepareInput); +await ctx.pause("review-ready"); +await ctx.agent("Review the prepared input"); +``` + +The first attempt pauses at `"review-ready"` and ends through cooperative cancellation. On resume, the factory starts from the beginning, reuses the journaled step, returns from the checkpoint, and continues. + The agent-facing `run_factory` tool has exactly two input branches: ```ts -{ name: string; args?: JsonValue; limits?: FactoryLimits } -{ resumeFromRunId: string; limits?: FactoryLimits } +{ name: string; args?: JsonValue; limits?: FactoryLimitOverrides } +{ resumeFromRunId: string; limits?: FactoryLimitOverrides } ``` ## Authoring a factory from inside a session @@ -253,9 +270,9 @@ const progressPage = await session.factory.getRunProgress(runId, { - `getRunDetail(runId)` returns phases, prompt-safe agent summaries, and the latest progress page. - `getRunProgress(runId, options?)` pages progress forward, backward, by phase, or from the latest tail. -`getRun(runId)` reads the latest run envelope, and `cancel(runId)` cancels a run and returns its terminal envelope. +`getRun(runId)` reads the latest run envelope. `pause(runId)` pauses a running attempt and returns its `paused` envelope. `cancel(runId)` cancels a run and returns its terminal envelope. -`waitForRun(runId, options?)` resolves with the terminal envelope once the run settles into `completed`, `error`, `halted`, or `cancelled`, and resolves immediately when it has already settled: +`waitForRun(runId, options?)` resolves with the current attempt's envelope once it settles into `completed`, `error`, `halted`, `paused`, or `cancelled`. It resolves immediately when the current attempt has already settled: ```ts const settled = await session.factory.waitForRun(runId); @@ -272,7 +289,7 @@ setTimeout(() => controller.abort(), 30_000); const settled = await session.factory.waitForRun(runId, { signal: controller.signal }); ``` -Aborting rejects the wait and has no effect on the run, which keeps executing — use `cancel(runId)` to actually stop it. Because a terminal envelope is final, the resolved value never changes afterwards. `isFactoryRunTerminal(status)` exposes the same terminal-status test for callers driving their own loop. +Aborting rejects the wait and has no effect on the run, which keeps executing—use `pause(runId)` or `cancel(runId)` to stop it. The resolved object is a snapshot of that settled attempt. If its status is `paused`, a later resume updates the durable envelope under the same run ID. Call `getRun(runId)` to read the latest envelope. `isFactoryRunTerminal(status)` exposes the same current-attempt settlement test for callers driving their own loop. Listen for the ephemeral `factory.run_updated` event. Its `{ runId, revision }` payload is an invalidation signal. Re-read the desired API when a newer monotonic revision arrives. diff --git a/nodejs/src/factory.ts b/nodejs/src/factory.ts index 6212f462b4..174318394f 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 { @@ -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. * @@ -198,7 +223,7 @@ export interface FactoryContext { args: TArgs; /** * The session instance returned by `joinSession`. It refuses calls that - * start or resume a factory run. + * start, resume, or pause a factory run. */ session: CopilotSession; /** Cooperative cancellation signal for the current factory run. */ @@ -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. */ @@ -314,13 +339,14 @@ export interface SessionFactoryApi { * Run a registered factory and resolve with its run envelope. * * The envelope is returned for every outcome, including `error`, `halted`, - * and `cancelled` — inspect `status` and read `result` only when the run - * completed. SDK-initiated runs do not request permission, so they have no - * declined outcome. The model's `run_factory` tool requests permission - * before a durable row exists; declining it creates no run row. Failures - * that occur before a run exists (such as an unknown factory or attempting - * to start a run while the session is at its active top-level run limit) - * still reject. + * `paused`, and `cancelled` — inspect `status` and read `result` only when + * the run completed. `paused` settles the current attempt, but the same + * durable run can later resume under its existing run ID. SDK-initiated + * runs do not request permission, so they have no declined outcome. The + * model's `run_factory` tool requests permission before a durable row + * exists; declining it creates no run row. Failures that occur before a run + * exists (such as an unknown factory or attempting to start a run while the + * session is at its active top-level run limit) still reject. */ run(name: string, options?: RunOptions): Promise; run( @@ -338,12 +364,13 @@ export interface SessionFactoryApi { /** Read the latest durable envelope for a factory run. */ getRun(runId: string): Promise; /** - * Wait for a run to settle and resolve with its terminal envelope. + * Wait for the current attempt to settle and resolve with its envelope. * - * Resolves as soon as the run reaches `completed`, `error`, `halted`, or - * `cancelled`, and resolves immediately when it has already settled. A - * terminal envelope is final, so the resolved value never changes - * afterwards. + * Resolves as soon as the run reaches `completed`, `error`, `halted`, + * `paused`, or `cancelled`, and resolves immediately when the current + * attempt has already settled. A `paused` envelope is an attempt-level + * snapshot: resuming the same durable run can later change the envelope + * returned by {@link SessionFactoryApi.getRun}. * * This watches the run's `factory.run_updated` invalidation events and * periodically re-reads the durable envelope so a missed event cannot @@ -375,6 +402,8 @@ export interface SessionFactoryApi { runId: string, options?: Omit ): Promise; + /** Pause a running factory attempt and return its `paused` 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..896eded0bd 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,29 @@ 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, 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 +201,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 +233,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 +452,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 +533,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 +674,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 +1589,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 +1654,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/e2e/factory.e2e.test.ts b/nodejs/test/e2e/factory.e2e.test.ts index 98004406f9..13e991fa10 100644 --- a/nodejs/test/e2e/factory.e2e.test.ts +++ b/nodejs/test/e2e/factory.e2e.test.ts @@ -259,6 +259,80 @@ it("resumes a failed factory when its session denies every permission request", expect(denyPermissions).not.toHaveBeenCalled(); }); +it("pauses a running factory through the session API", async () => { + if (!factoryTestContext) { + throw new Error("Factory E2E requires the stdio transport"); + } + const { workDir } = factoryTestContext; + const extensionDir = join(workDir, ".github", "extensions", "factory-smoke"); + await using session = await setupFactoryExtension(workDir); + + const execution = session.factory.run("externally-paused", { + notifyOnComplete: false, + }); + await retry( + "wait for the externally paused factory to enter its body", + async () => { + expect(existsSync(join(extensionDir, "external-pause-entered"))).toBe(true); + }, + 100, + 100 + ); + + let runId: string | undefined; + await retry( + "find the running factory before pausing it", + async () => { + const running = (await session.factory.listRuns()).find( + (run) => run.factoryName === "externally-paused" && run.status === "running" + ); + expect(running).toBeDefined(); + runId = running?.runId; + }, + 100, + 100 + ); + if (!runId) { + throw new Error("Running factory did not expose a run ID"); + } + + await expect(session.factory.pause(runId)).resolves.toMatchObject({ + runId, + status: "paused", + }); + await expect(execution).resolves.toMatchObject({ + runId, + status: "paused", + }); +}); + +it("pauses once at a durable checkpoint and continues after resume", async () => { + if (!factoryTestContext) { + throw new Error("Factory E2E requires the stdio transport"); + } + const { workDir } = factoryTestContext; + const extensionDir = join(workDir, ".github", "extensions", "factory-smoke"); + await using session = await setupFactoryExtension(workDir); + + const paused = await session.factory.run("durable-pause-checkpoint", { + notifyOnComplete: false, + }); + expect(paused).toMatchObject({ status: "paused" }); + expect(readFileSync(join(extensionDir, "checkpoint-attempts"))).toHaveLength(1); + expect(readFileSync(join(extensionDir, "checkpoint-preparations"))).toHaveLength(1); + + const resumed = await session.factory.resume(paused.runId, { + notifyOnComplete: false, + }); + expect(resumed).toMatchObject({ + runId: paused.runId, + status: "completed", + result: { attempt: 2, prepared: 1 }, + }); + expect(readFileSync(join(extensionDir, "checkpoint-attempts"))).toHaveLength(2); + expect(readFileSync(join(extensionDir, "checkpoint-preparations"))).toHaveLength(1); +}); + it("refuses a factory started through the context session from a factory body", async () => { if (!factoryTestContext) { throw new Error("Factory E2E requires the stdio transport"); @@ -272,7 +346,7 @@ it("refuses a factory started through the context session from a factory body", expect(result).toMatchObject({ status: "completed", - result: expect.stringContaining("factory.run and factory.resume"), + result: expect.stringContaining("factory.run, factory.resume, and factory.pause"), }); expect((result as { result: string }).result).toContain("factory body"); }); @@ -290,7 +364,7 @@ it("refuses a factory started through the module session from a factory body", a expect(result).toMatchObject({ status: "completed", - result: expect.stringContaining("factory.run and factory.resume"), + result: expect.stringContaining("factory.run, factory.resume, and factory.pause"), }); expect((result as { result: string }).result).toContain("factory body"); }); diff --git a/nodejs/test/e2e/fixtures/factory-extension.mjs b/nodejs/test/e2e/fixtures/factory-extension.mjs index fb344b4863..b082599bc9 100644 --- a/nodejs/test/e2e/fixtures/factory-extension.mjs +++ b/nodejs/test/e2e/fixtures/factory-extension.mjs @@ -1,4 +1,4 @@ -import { existsSync, writeFileSync } from "node:fs"; +import { closeSync, existsSync, fstatSync, openSync, writeFileSync, writeSync } from "node:fs"; import { defineFactory, joinSession } from "@github/copilot-sdk/extension"; const marker = (name) => new URL(`./${name}`, import.meta.url); @@ -13,6 +13,16 @@ async function waitForMarker(name, timeoutMs) { } } +function incrementMarker(name) { + const descriptor = openSync(marker(name), "a+"); + try { + writeSync(descriptor, "1"); + return fstatSync(descriptor).size; + } finally { + closeSync(descriptor); + } +} + const argumentEcho = defineFactory({ meta: { name: "argument-echo", @@ -152,6 +162,40 @@ const failsOnce = defineFactory({ }, }); +const externallyPaused = defineFactory({ + meta: { + name: "externally-paused", + description: "Wait until the calling session pauses this run.", + phases: [], + }, + run: async ({ signal }) => { + writeFileSync(marker("external-pause-entered"), "entered"); + await new Promise((_, reject) => { + const abort = () => reject(signal.reason ?? new Error("Factory aborted")); + if (signal.aborted) { + abort(); + return; + } + signal.addEventListener("abort", abort, { once: true }); + }); + return "unexpectedly completed"; + }, +}); + +const durablePauseCheckpoint = defineFactory({ + meta: { + name: "durable-pause-checkpoint", + description: "Pause once after journaled preparation, then complete after resume.", + phases: [], + }, + run: async ({ pause, step }) => { + const attempt = incrementMarker("checkpoint-attempts"); + const prepared = await step("prepare", () => incrementMarker("checkpoint-preparations")); + await pause("review-ready"); + return { attempt, prepared }; + }, +}); + session = await joinSession({ factories: [ argumentEcho, @@ -162,6 +206,8 @@ session = await joinSession({ startsFromModuleSession, parked, failsOnce, + externallyPaused, + durablePauseCheckpoint, ], }); diff --git a/nodejs/test/factory.test.ts b/nodejs/test/factory.test.ts index c9b8f65074..182e2b8d3a 100644 --- a/nodejs/test/factory.test.ts +++ b/nodejs/test/factory.test.ts @@ -479,8 +479,13 @@ describe("factories", () => { "Options are exactly `label`, `schema`, `model`, `agent`, `reasoningEffort`, and `contextTier`" ); expect(normalizedGuide).toContain( - "session returned by `joinSession`. It refuses calls that start or resume a factory run" + "session returned by `joinSession`. It refuses calls that start, resume, or pause a factory run" ); + expect(normalizedGuide).toContain( + "A `paused` envelope means that the current attempt settled, not that the durable run is permanently finished" + ); + expect(normalizedGuide).toContain('await ctx.pause("review-ready")'); + expect(normalizedGuide).toContain("session.factory.pause(runId)"); expect(normalizedPublicApi).toContain("SDK-initiated runs do not request permission"); expect(normalizedPublicApi).toContain("declining it creates no run row"); @@ -490,7 +495,10 @@ describe("factories", () => { expect(normalizedPublicApi).toContain("SDK-initiated resumes do not request permission"); expect(normalizedPublicApi).toContain("with a documented resume code rejects with"); expect(normalizedPublicApi).toContain( - "session instance returned by `joinSession`. It refuses calls that start or resume a factory run" + "session instance returned by `joinSession`. It refuses calls that start, resume, or pause a factory run" + ); + expect(normalizedPublicApi).toContain( + "`paused` settles the current attempt, but the same durable run can later resume" ); }); @@ -1407,6 +1415,7 @@ describe("factories", () => { await session.clientSessionApis.factory!.abort({ sessionId: session.sessionId, runId, + executionToken: "execution-token", }); await step( "volatile", @@ -1571,6 +1580,199 @@ 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.run, factory.resume, and factory.pause are not allowed while a factory body is running on this call path." + ); + 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 +2183,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 +2223,69 @@ 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 +2326,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 +2474,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 +2588,7 @@ describe("factory run settlement", () => { ["completed", true], ["error", true], ["halted", true], + ["paused", true], ["cancelled", true], ["pending", false], ["running", false],