diff --git a/nodejs/docs/extensions.md b/nodejs/docs/extensions.md index 8b36de8a50..d33a733120 100644 --- a/nodejs/docs/extensions.md +++ b/nodejs/docs/extensions.md @@ -56,4 +56,5 @@ The `session` object provides methods for sending messages, logging to the timel ## Further Reading - `examples.md` — Practical code examples for tools, hooks, events, and complete extensions +- `factories.md`: Authoring, running, resuming, and observing Agent Factories - `agent-author.md` — Step-by-step workflow for agents authoring extensions programmatically diff --git a/nodejs/docs/factories.md b/nodejs/docs/factories.md new file mode 100644 index 0000000000..0c1f0f09a0 --- /dev/null +++ b/nodejs/docs/factories.md @@ -0,0 +1,240 @@ +# Agent Factories + +Agent Factories are extension-authored, session-scoped workflows that coordinate subagents and durable steps. The API is experimental. + +## Define and register a factory + +Use `defineFactory` and pass the returned handle to `joinSession`: + +```js +import { defineFactory, joinSession } from "@github/copilot-sdk/extension"; + +const reviewChanged = defineFactory({ + meta: { + name: "review-changed", + description: + "Review changed files and verify the findings. " + + "args: { files: string[] } — the paths to review.", + phases: [{ title: "Review" }, { title: "Verify" }], + limits: { + maxConcurrentSubagents: 3, + maxTotalSubagents: 10, + timeoutSeconds: 90.5, + maxAiCredits: 5, + }, + }, + run: async (ctx) => { + ctx.phase("Review"); + const reviews = await ctx.parallel( + ctx.args.files.map( + (file) => () => ctx.agent(`Review ${file}`, { label: `Review ${file}` }) + ) + ); + + ctx.phase("Verify"); + const report = await ctx.step("report", () => ({ reviews })); + ctx.log(`Completed factory run ${ctx.runId}`); + return report; + }, +}); + +const session = await joinSession({ factories: [reviewChanged] }); +``` + +Factory metadata contains a stable `name`, a human-readable `description`, declared `phases`, and optional `limits`. Phase entries contain a `title` and optional `detail`. + +There is no declared schema for `ctx.args`. The `run_factory` tool forwards `args` verbatim and its parameter is untyped, so **the `description` is the only thing telling an agent what arguments to supply** — state the expected shape there whenever a factory reads `ctx.args`, as the example above does. Arguments supplied by an extension calling `session.factory.run(...)` directly are typed through `defineFactory`, but that typing does not reach the model. A factory that reads `ctx.args` should validate it rather than assume a shape. + +`defineFactory` accepts a `run(context)` function returning `Promise`, where `TResult` is `JsonValue | void`. Objects, arrays, strings, numbers, booleans, and `null` are valid results. Returning `undefined` completes the factory with no result. Other non-JSON values are rejected. + +## Factory context + +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`, and `model`. 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 full session returned by `joinSession`. +- `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. + +### Subagent calls + +`ctx.agent(prompt, options?)` spawns one factory-scoped subagent and awaits it. Without a schema it resolves to the subagent's final text. With `options.schema` it resolves to the parsed JSON value. + +**Identical calls are memoized into one subagent.** Each call is journaled by its canonical prompt and options, including `label`. Two calls with the same prompt and the same options return one shared result — even when issued concurrently. To spawn N *independent* subagents, give each a unique `label` or vary the prompt: + +```js +// One subagent, awaited five times — almost certainly not what you want. +await ctx.parallel([1, 2, 3, 4, 5].map(() => () => ctx.agent("Find a bug"))); + +// Five independent subagents. +await ctx.parallel( + [1, 2, 3, 4, 5].map((i) => () => ctx.agent("Find a bug", { label: `finder:${i}` })) +); +``` + +**An ordinary failure resolves to `null` — it does not throw.** A subagent that errors, returns nothing, or (with a schema) produces output that still fails to parse or match after its one retry resolves `null`. Always guard the result before using it, including a bare `await ctx.agent(...)`: + +```js +const finding = await ctx.agent(prompt, { label: "inspector" }); +if (!finding) return { finding: null }; +``` + +Cancellation and hard runtime failures — a reached limit, a durable-state failure — reject instead, aborting the run. When filtering results, prefer `v => v !== null` over `Boolean`, which also discards a valid `false`, `0`, or `""`. + +**`schema` is a structural subset of JSON Schema, not a validator.** Honored: `type`, `required`, `enum`, `const`, recursive `properties`/`items`, and `anyOf`/`oneOf`/`allOf` — where `oneOf` is treated as `anyOf`, meaning at least one branch matches rather than exactly one. Ignored and *not* enforced: `additionalProperties`, `pattern`, `minLength`/`maxLength`, `format`, numeric ranges, and boolean schemas. Do not rely on an ignored keyword to constrain a result. A schema call retries once on a parse or match failure, so it may spawn twice, and both spawns count toward `maxTotalSubagents`. + +### Choosing between pipeline and parallel + +Prefer `pipeline` for multi-stage work. It has no barrier between stages, so each item advances as soon as its own prior stage finishes. + +Reach for a barrier — `parallel` between stages — only when a stage genuinely needs every prior result at once: deduplicating or merging across the full set, an early exit based on the total, or a prompt that compares one result against the others. Needing to map, filter, or flatten is not a reason to use a barrier; do that inside a pipeline stage. Barrier latency is real: if the slowest of N subagents takes three times the fastest, a barrier wastes the rest of the pool's time. + +See [factory-patterns.md](./factory-patterns.md) for composable orchestration patterns built on these primitives. + +## Resource limits + +Limits may be declared in `meta.limits` and overridden per invocation. All limits must be positive when present. + +- `maxConcurrentSubagents`: Positive integer concurrent-subagent cap. Additional subagents wait in a queue. Queueing applies backpressure and does not fail the run. +- `maxTotalSubagents`: Positive integer cumulative admission cap. An attempted subagent beyond the cap ends the attempt with failure kind `maxTotalSubagents`. +- `timeoutSeconds`: Positive finite number of seconds, including positive fractions, capped at `2_147_483.647`. It measures accumulated active-execution time across attempts, including the extension body, subprocess waits, queued-agent waits, and sleeps. Time between attempts is excluded. The timeout is soft because already-running work may take time to stop. Its failure kind is `timeoutSeconds`. +- `maxAiCredits`: Positive finite AI-credit budget for the whole run's factory subagent subtree, including descendants. AI credits are GitHub Copilot's universal usage metric. This is a soft, post-paid ceiling, so completed or parallel turns can settle above it before the run stops. Accounting is fail-closed: an accounting failure stops a budgeted run rather than allowing untracked use. Its failure kind is `maxAiCredits`. + +`maxTotalSubagents`, `timeoutSeconds`, and `maxAiCredits` use reject-and-retry semantics. A rejected attempt ends with run status `error` and `failure.type` set to `factory_limit_reached`. The failed run keeps its ID, arguments, journal, and accounting. Resume the run with a raised limit when additional work is approved. Previously consumed resources still count. + +## Run and resume + +Run by registered name or handle: + +```ts +const run = await session.factory.run("review-changed", { + args: { files: ["src/a.ts"] }, + limits: { maxAiCredits: 3 }, +}); + +if (run.status === "completed") { + console.log(run.result); +} else { + console.error(`run ${run.runId} ended as ${run.status}`, run.failure ?? run.error); +} +``` + +The name overload is: + +```ts +session.factory.run( + name: string, + options?: { args?: JsonValue; limits?: FactoryLimits }, +): Promise; +``` + +Resume by run ID without resending the name or arguments: + +```ts +const run = await session.factory.resume(runId, { + limits: { maxAiCredits: 6 }, +}); +``` + +The signature is: + +```ts +session.factory.resume( + runId: string, + options?: { limits?: FactoryLimits }, +): Promise; +``` + +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`. A declined fresh run is not a pre-execution failure: the run row already exists by the time the prompt is answered, so it resolves with a terminal `cancelled` envelope carrying the run ID. Only failures that occur *before* a run exists reject: an unknown factory name or an already-active session. Pre-execution resume failures, including a declined reapproval, throw `FactoryResumeError`, whose `code` is one of `not_found`, `non_resumable`, `already_active`, `reapproval_declined`, or `no_approval_provider`. + +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. + +The agent-facing `run_factory` tool has exactly two input branches: + +```ts +{ name: string; args?: JsonValue; limits?: FactoryLimits } +{ resumeFromRunId: string; limits?: FactoryLimits } +``` + +## Authoring a factory from inside a session + +The agent-facing `factories_manage` tool writes a factory into a session-scoped extension at runtime with `operation: "author"`. The rules above all apply, plus one constraint that does not affect an extension author. + +**The `run` body is self-contained.** It is emitted verbatim into a generated module as a single async function expression. It closes over nothing: not the conversation that authored it, and not any authoring-time binding. Only its own locals, its `ctx` parameter, and standard Node and JavaScript globals are in scope, so every schema, constant, and helper must be defined *inside* the function. The generated module imports the SDK itself; the expression cannot add static `import` statements or use `require`. Load anything else with a dynamic `await import("...")` in the body. + +```js +async ({ args, agent, phase }) => { + // Defined inside — there is no outer scope to close over. + const VERDICT = { type: "object", properties: { real: { type: "boolean" } }, required: ["real"] }; + + phase("Inspect"); + const finding = await agent(`Name one likely bug in ${args.file ?? "the code"}.`, { + label: "inspector", + }); + if (!finding) return { finding: null, real: false }; + + phase("Verify"); + const verdict = await agent(`Is this a real bug? Claim: ${finding}`, { + label: "verifier", + schema: VERDICT, + }); + return { finding, real: verdict?.real === true }; +}; +``` + +Authoring registers the factory but does not run it. Invoke it afterwards with `run_factory`. Use `factories_manage` with `operation: "list"` to see the factories already registered in the session and `operation: "inspect"` to read one factory's description, phases, and limits before running it. + +## Observe a run + +The calling session can inspect its own factory runs: + +```ts +const runs = await session.factory.listRuns(); +const detail = await session.factory.getRunDetail(runId); +const page = await session.factory.getRunProgress(runId, { + phaseId, + afterSeq, + beforeSeq, + limit, +}); +``` + +- `listRuns()` returns summaries in durable creation order. +- `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. + +`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: + +```ts +const settled = await session.factory.waitForRun(runId); +if (settled.status === "completed") { + console.log(settled.result); +} +``` + +It watches `factory.run_updated` and re-reads the durable envelope on each invalidation, collapsing a burst of events into a single in-flight read. A low-frequency periodic re-read runs alongside the subscription, so a dropped or missing invalidation degrades into a slightly late resolution rather than an unbounded wait. Pass a `signal` to stop waiting: + +```ts +const controller = new AbortController(); +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. + +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. + +Revisions cover durable lifecycle, accounting, phase, agent, and progress changes. Continuous read-time fields can change without a new revision. These include `observedAt`, active-time calculations, live counts, and a live agent's status or prompt-safe activity text. Factory prompts are never exposed by these APIs. A run is visible only through the session that owns it. diff --git a/nodejs/docs/factory-patterns.md b/nodejs/docs/factory-patterns.md new file mode 100644 index 0000000000..66c6d13b1f --- /dev/null +++ b/nodejs/docs/factory-patterns.md @@ -0,0 +1,194 @@ +# Agent Factory patterns + +Composable orchestration patterns built on the factory context. Read [factories.md](./factories.md) first for the API and its semantics. The API is experimental. + +Every snippet below assumes the surrounding `async (ctx) => { ... }` run body and destructures the hooks it uses. Three rules apply throughout, because breaking them fails silently: + +- **Give every independent subagent a unique `label`.** Identical prompt-and-options pairs memoize into a single shared subagent. +- **Guard every `agent()` result.** An ordinary failure resolves to `null` rather than throwing. +- **Filter with `v => v !== null`,** not `Boolean`, which also discards a valid `false`, `0`, or `""`. + +## Multi-stage review + +The default shape: fan out across dimensions, and let each dimension verify as soon as its own review lands. No barrier, so a slow dimension never holds up a fast one. + +```js +async ({ pipeline, parallel, agent, phase, log }) => { + const FINDINGS = { + type: "object", + properties: { + findings: { + type: "array", + items: { + type: "object", + properties: { title: { type: "string" } }, + required: ["title"], + }, + }, + }, + required: ["findings"], + }; + const VERDICT = { + type: "object", + properties: { isReal: { type: "boolean" } }, + required: ["isReal"], + }; + const DIMENSIONS = [ + { key: "bugs", prompt: "Review the diff for correctness bugs. Return JSON {findings:[{title}]}." }, + { key: "perf", prompt: "Review the diff for performance issues. Return JSON {findings:[{title}]}." }, + ]; + + phase("Review"); // Run-global: set it before the fan-out, never inside a stage. + const perDimension = await pipeline( + DIMENSIONS, + (d) => agent(d.prompt, { label: `review:${d.key}`, schema: FINDINGS }), + (review, d) => { + if (!review) { + log(`review:${d.key} produced nothing`); + return []; + } + return parallel( + (review.findings ?? []).map((f, i) => () => + agent(`Adversarially verify this finding is real: ${f.title}`, { + label: `verify:${d.key}:${i}`, + schema: VERDICT, + }).then((v) => (v && v.isReal ? f : null)) + ) + ); + } + ); + + return { confirmed: perDimension.flat().filter((v) => v !== null) }; +}; +``` + +## When a barrier is correct + +Deduplicating across every finding needs the whole set in hand, so the barrier earns its cost here. Dedup itself is plain JavaScript, done in the body between the two fan-outs. This excerpt reuses `FINDINGS`, `VERDICT`, and `DIMENSIONS` from the previous example — define them inside your own function. + +```js +const all = await parallel( + DIMENSIONS.map((d) => () => agent(d.prompt, { label: `find:${d.key}`, schema: FINDINGS })) +); +const findings = all.filter((v) => v !== null).flatMap((r) => r.findings ?? []); +const deduped = [...new Map(findings.map((f) => [f.title, f])).values()]; // Needs all of them. +const verified = await parallel( + deduped.map((f, i) => () => agent(`Verify: ${f.title}`, { label: `verify:${i}`, schema: VERDICT })) +); +``` + +## Loop until count + +Accumulate toward a target. Each iteration needs a unique identity — a unique label plus a prompt that excludes what has already been found — a bounded attempt count, and a null guard. + +```js +const BUG = { + type: "object", + properties: { title: { type: "string" } }, + required: ["title"], +}; + +const bugs = []; +let attempt = 0; +while (bugs.length < 10 && attempt < 30) { + const r = await agent( + `Find ONE distinct bug NOT already listed: ${JSON.stringify(bugs.map((b) => b.title))}. Return JSON {title}.`, + { label: `finder:${attempt}`, schema: BUG } + ); + attempt++; + if (r && r.title) bugs.push(r); + log(`${bugs.length}/10 found`); +} +``` + +## Loop until dry + +Keep spawning finders until some number of consecutive rounds surface nothing new. Deduplicate against everything *seen*, not just what was kept, or discarded findings resurface every round. + +```js +const BUGS = { + type: "object", + properties: { + bugs: { + type: "array", + items: { type: "object", properties: { title: { type: "string" } }, required: ["title"] }, + }, + }, + required: ["bugs"], +}; +const VERDICT = { + type: "object", + properties: { real: { type: "boolean" } }, + required: ["real"], +}; + +const seen = new Set(); +const confirmed = []; +const keyOf = (b) => b.title.toLowerCase(); +let dry = 0; +let round = 0; + +while (dry < 2 && round < 20) { + const found = ( + await parallel( + [0, 1, 2].map((i) => () => + agent(`Find bugs (finder ${i}, round ${round}). Return JSON {bugs:[{title}]}.`, { + label: `find:${round}:${i}`, + schema: BUGS, + }) + ) + ) + ) + .filter((v) => v !== null) + .flatMap((r) => r.bugs ?? []); + + const fresh = found.filter((b) => { + const k = keyOf(b); + if (seen.has(k)) return false; + seen.add(k); + return true; + }); + + if (!fresh.length) { + dry++; + round++; + continue; + } + dry = 0; + + const judged = await parallel( + fresh.map((b, i) => () => + parallel( + ["correctness", "security", "repro"].map((lens) => () => + agent(`Judge via ${lens}: is "${b.title}" real? Return JSON {real}.`, { + label: `judge:${round}:${i}:${lens}`, + schema: VERDICT, + }) + ) + ).then((vs) => ({ b, real: vs.filter((v) => v !== null).filter((v) => v.real).length >= 2 })) + ) + ); + + confirmed.push(...judged.filter((v) => v !== null && v.real).map((v) => v.b)); + round++; +} +``` + +## Quality patterns + +Compose these freely. + +- **Adversarial verify.** Spawn several independent skeptics per finding, each prompted to *refute* it and to default to refuted when uncertain. Keep only what a majority fails to refute. +- **Perspective-diverse verify.** Give each verifier a distinct lens — correctness, security, performance, does-it-reproduce — instead of several identical skeptics. The distinct prompts also stop them memoizing into one subagent. +- **Judge panel.** Generate several independent attempts from different angles, score them with parallel judges, then synthesize from the winner while grafting the best ideas from the runners-up. +- **Multi-modal sweep.** Run parallel searchers that each look a different way: by container, by content, by entity, by time. +- **Completeness critic.** End with an agent asking what is missing — an angle not run, a claim unverified, a source unread — and use its answer to seed the next round. +- **No silent caps.** When the factory bounds its own coverage with a top-N, a sampling step, or a no-retry rule, `log()` what was dropped. + +## Scaling + +Match the orchestration to what was asked. A quick check wants a couple of subagents and single-vote verification; a request to be thorough or comprehensive wants a larger finder pool, a three-to-five vote adversarial pass, and a synthesis stage. + +There is no in-script budget object. Scale with your own counters, as in the loop patterns above, and treat the declared limits as the safety ceiling rather than the control mechanism. Only `agent()` spawns are throttled, by `maxConcurrentSubagents` falling back to `maxTotalSubagents`; with neither declared there is no built-in concurrency cap, so declare one before fanning out widely. `parallel` itself is `Promise.all`, so non-agent work in a thunk runs fully concurrently regardless. + +These patterns are not exhaustive. Compose novel harnesses — tournament brackets, self-repair loops, staged escalation — when the task calls for it. diff --git a/nodejs/src/client.ts b/nodejs/src/client.ts index 61f4a99416..6d99ce49e3 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 72bde93bde..c3ae0fd874 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,42 @@ 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, + FactoryResumeError, + isFactoryRunTerminal, + type RunOptions, + type ResumeOptions, + type FactoryResumeErrorCode, + type SessionFactoryApi, + type FactoryAgentOptions, + type FactoryContext, + type FactoryDefinition, + type FactoryHandle, + type FactoryJsonSchema, + type JsonValue, + type FactoryPipelineStage, + type FactoryStepOptions, + type FactoryRunResult, + type FactoryRunStatus, + type FactoryRunSummary, + type FactoryRunDetail, + type FactoryProgressPage, + type FactoryProgressLine, + type FactoryPhaseObservation, + type FactoryPhaseStatus, + type FactoryAgentSummary, +} from "./factory.js"; /** * Joins the current foreground session. @@ -58,14 +91,22 @@ export async function joinSession(config: JoinSessionConfig = {}): Promise & { + /** Completed factory result. */ + result?: JsonValue; +}; + +export type { + FactoryAgentSummary, + FactoryPhaseStatus, + FactoryPhaseObservation, + FactoryProgressLine, + FactoryProgressPage, + FactoryRunDetail, + FactoryRunStatus, + FactoryRunSummary, +} from "./generated/rpc.js"; + +/** + * 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. + */ +const FACTORY_TERMINAL_STATUSES: ReadonlySet = new Set([ + "completed", + "halted", + "cancelled", + "error", +]); + +/** + * Whether a factory run status is terminal. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export function isFactoryRunTerminal(status: FactoryRunStatus): boolean { + return FACTORY_TERMINAL_STATUSES.has(status); +} + +declare const factoryHandleBrand: unique symbol; + +/** A value that can be represented losslessly on the SDK JSON wire. */ +export type JsonValue = + | null + | boolean + | number + | string + | JsonValue[] + | { [key: string]: JsonValue }; + +/** + * Conservative JSON shape language accepted for structured factory agent output. + * + * This is a best-effort structural guard used to decide whether a subagent's + * structured output should be accepted or retried — **not** a full JSON Schema + * validator. Only these keywords are honored: `type`, `required`, `enum`, + * `const`, recursive `properties`/`items`, and `anyOf`/`oneOf`/`allOf`. + * + * Everything else is **ignored, not enforced**. In particular, string + * constraints (`pattern`, `minLength`, `maxLength`, `format`), numeric ranges + * (`minimum`, `maximum`), `additionalProperties`, and boolean (`true`/`false`) + * schemas do not reject non-conforming output. `oneOf` is treated like `anyOf` + * (at least one branch must match) rather than strict exactly-one. Author + * schemas within this subset; do not rely on unsupported constraints for + * correctness. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export type FactoryJsonSchema = { [key: string]: JsonValue }; + +/** + * 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 { + /** Stable identifier for the current factory run. */ + readonly runId: string; + /** 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 | JsonValue, + options?: FactoryStepOptions + ): Promise; + /** + * Run thunks concurrently and await all of them. + * + * 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. + */ + parallel( + thunks: Array<() => Promise | TResult> + ): Promise>; + /** + * Run each item through every stage without barriers between stages. + * + * A stage that throws drops that item to `null` and skips its remaining + * stages. As with {@link FactoryContext.parallel}, cancellation and hard + * runtime failures propagate instead of being recorded per item. + */ + 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?: JsonValue): 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< + TArgs extends JsonValue = JsonValue, + TResult extends JsonValue | void = JsonValue | void, +> { + meta: FactoryMeta; + run(context: FactoryContext): Promise; +} + +/** + * A deeply immutable view of a value. + * + * `defineFactory` deep-freezes the metadata it stores, so the handle's view of + * it has to be readonly all the way down or `handle.meta.name = "..."` and + * `handle.meta.phases.push(...)` would compile and then throw at runtime. + */ +type DeepReadonly = T extends (infer U)[] + ? readonly DeepReadonly[] + : T extends object + ? { readonly [K in keyof T]: DeepReadonly } + : T; + +/** + * 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< + TArgs extends JsonValue = JsonValue, + TResult extends JsonValue | void = JsonValue | void, +> { + readonly meta: DeepReadonly; + 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; + /** Optional per-invocation resource ceiling overrides. */ + limits?: FactoryLimits; + /** + * Prior run whose persisted identity, arguments, journal, and accounting should be resumed. + * + * @deprecated Use {@link SessionFactoryApi.resume} instead. + */ + resumeFromRunId?: string; +} + +/** + * Options for resuming a factory run by ID. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export interface ResumeOptions { + /** Optional per-invocation resource ceiling overrides. */ + limits?: FactoryLimits; +} + +/** + * Machine-readable pre-execution factory resume failure. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export type FactoryResumeErrorCode = + | "not_found" + | "non_resumable" + | "already_active" + | "reapproval_declined" + | "no_approval_provider"; + +/** + * 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 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. A declined fresh run resolves with a terminal `cancelled` + * envelope. Failures that occur before a run exists (such as an unknown + * factory or an already-active session) still reject. + */ + run(name: string, options?: RunOptions): Promise; + run( + factory: FactoryHandle, + options?: RunOptions + ): Promise; + /** + * Resume a run from its persisted factory name, arguments, journal, and accounting. + * + * Resolves with the run envelope like {@link SessionFactoryApi.run}. A + * pre-execution failure, including declined reapproval, rejects with + * {@link FactoryResumeError}. + */ + resume(runId: string, options?: ResumeOptions): Promise; + /** 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. + * + * 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. + * + * This watches the run's `factory.run_updated` invalidation events and + * periodically re-reads the durable envelope so a missed event cannot + * leave the wait hanging. Pass a `signal` to stop waiting; aborting rejects + * and has no effect on the run itself, which keeps executing. Use + * {@link SessionFactoryApi.cancel} to actually stop it. + */ + waitForRun(runId: string, options?: { signal?: AbortSignal }): Promise; + /** List this session's durable factory runs in creation order. */ + listRuns(): Promise; + /** Read durable phases, direct agents, and the latest progress tail for a run. */ + getRunDetail(runId: string): Promise; + /** Page durable progress forward, backward, or from the latest tail. */ + getRunProgress( + runId: string, + options?: Omit + ): Promise; + /** Cancel a factory run and return its terminal envelope. */ + cancel(runId: string): Promise; +} + +/** + * Error thrown when a factory cannot be resumed before execution begins. + * + * @experimental Part of the experimental Agent Factories surface and may + * change or be removed in future SDK or CLI releases. + */ +export class FactoryResumeError extends Error { + constructor( + public readonly code: FactoryResumeErrorCode, + message: string + ) { + super(message); + this.name = "FactoryResumeError"; + } +} + +interface StoredFactory { + meta: FactoryMeta; + run(context: FactoryContext): Promise; +} + +const factoryHandles = new WeakMap(); + +/** Maximum accepted factory timeout in seconds, derived from Node's maximum timer delay. */ +const MAX_FACTORY_TIMEOUT_SECONDS = 2_147_483.647; +const NANO_AIU_PER_AIU = 1_000_000_000; + +function deepFreeze(value: T): T { + if (value !== null && typeof value === "object" && !Object.isFrozen(value)) { + Object.freeze(value); + for (const nested of Object.values(value)) { + deepFreeze(nested); + } + } + return value; +} + +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.timeoutSeconds !== undefined && + (!Number.isFinite(limits.timeoutSeconds) || limits.timeoutSeconds <= 0) + ) { + throw new Error( + 'Factory limit "timeoutSeconds" must be a positive, finite number of seconds' + ); + } + if ( + limits.timeoutSeconds !== undefined && + limits.timeoutSeconds > MAX_FACTORY_TIMEOUT_SECONDS + ) { + throw new Error( + `Factory limit "timeoutSeconds" must not exceed ${MAX_FACTORY_TIMEOUT_SECONDS} seconds` + ); + } + + if (limits.maxAiCredits !== undefined) { + const maxNanoAiu = Math.round(limits.maxAiCredits * NANO_AIU_PER_AIU); + if ( + !Number.isFinite(limits.maxAiCredits) || + limits.maxAiCredits <= 0 || + !Number.isSafeInteger(maxNanoAiu) || + maxNanoAiu < 1 + ) { + throw new Error( + 'Factory limit "maxAiCredits" must be a positive, finite number that rounds to a safe positive integer nano-AIU ceiling' + ); + } + } +} + +function validatePhases(meta: FactoryMeta): void { + const titles = new Set(); + for (const phase of meta.phases) { + if (phase.title.trim().length === 0) { + throw new Error("Factory phase titles must not be empty"); + } + if (titles.has(phase.title)) { + throw new Error(`Factory phase title "${phase.title}" is declared more than once`); + } + titles.add(phase.title); + } +} + +/** + * 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< + TArgs extends JsonValue = JsonValue, + TResult extends JsonValue | void = JsonValue | void, +>(definition: FactoryDefinition): FactoryHandle { + // Snapshot before validating so post-registration mutation of the caller's + // object cannot slip past the authoring-boundary checks. + const meta = deepFreeze(structuredClone(definition.meta)); + validateLimits(meta); + validatePhases(meta); + + const stored: StoredFactory = { + meta, + run: definition.run, + }; + const handle = Object.freeze({ meta }) as unknown 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/index.ts b/nodejs/src/index.ts index 683adb4d5b..99df267b5b 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, FactoryResumeError, isFactoryRunTerminal } from "./factory.js"; export { Canvas, CanvasError, @@ -93,6 +94,8 @@ export type { LargeToolOutputConfig, MemoryConfiguration, UiInputOptions, + FactoryLimits, + FactoryMeta, MCPStdioServerConfig, MCPHTTPServerConfig, MCPServerConfig, @@ -168,3 +171,26 @@ export type { TypedSessionLifecycleHandler, ZodSchema, } from "./types.js"; +export type { + RunOptions, + ResumeOptions, + FactoryResumeErrorCode, + SessionFactoryApi, + FactoryAgentOptions, + FactoryContext, + FactoryDefinition, + FactoryHandle, + FactoryJsonSchema, + JsonValue, + FactoryPipelineStage, + FactoryStepOptions, + FactoryRunResult, + FactoryRunStatus, + FactoryRunSummary, + FactoryRunDetail, + FactoryProgressPage, + FactoryProgressLine, + FactoryPhaseObservation, + FactoryPhaseStatus, + FactoryAgentSummary, +} from "./factory.js"; diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 89403ade7b..8b946c7860 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -15,6 +15,11 @@ import type { CanvasActionInvokeResult, CurrentToolMetadata, McpOauthPendingRequestResponse, + FactoryLogLine, + FactoryRunRequest, + FactoryExecuteResult, + FactoryJournalPutRequest, + FactoryRunResult as WireFactoryRunResult, } from "./generated/rpc.js"; import { type Canvas, CanvasError } from "./canvas.js"; import type { OpenCanvasInstance } from "./generated/rpc.js"; @@ -60,6 +65,29 @@ import type { UserInputRequest, UserInputResponse, } from "./types.js"; +import { + getFactoryDefinition, + FactoryResumeError, + isFactoryRunTerminal, + type FactoryResumeErrorCode, + type FactoryRunResult, + type RunOptions, + type SessionFactoryApi, + type FactoryContext, + type FactoryHandle, + type JsonValue, + type FactoryStepOptions, +} from "./factory.js"; + +function isFactoryResumeErrorCode(value: unknown): value is FactoryResumeErrorCode { + return ( + value === "not_found" || + value === "non_resumable" || + value === "already_active" || + value === "reapproval_declined" || + value === "no_approval_provider" + ); +} /** * Convert a raw hook input received over the wire into its public-facing shape. @@ -103,6 +131,242 @@ 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 and hard runtime failures must propagate out + // of the combinator rather than be mapped to a successful + // `null`; otherwise an aborted run, or one that hit a + // resource ceiling or durable-state failure, could be + // reported as completed. An ordinary subagent failure never + // rejects — it already resolves `null`. + if (isFactoryFatalError(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 and hard runtime failures instead + // of mapping them to `null`, so an aborted stage — or one + // that hit a resource ceiling or durable-state failure — + // does not let the run report success. + if (isFactoryFatalError(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; + this.clearFlushTimer(); + const lines = this.pending.splice(0); + await this.flushTail; + if (this.flushFailed) { + throw this.flushError; + } + if (lines.length > 0) { + try { + await this.send(lines); + } catch (error) { + console.warn( + "Failed to flush final factory progress after the factory body settled", + error + ); + } + } + } + + 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; + } + } +} + +/** + * Reconcile the generated envelope with the public one. + * + * The two are identical at runtime. They differ only in how `result` is typed: + * the runtime returns any JSON value, but the schema models the field as an + * opaque node, which the generator renders as an object. {@link FactoryRunResult} + * corrects that for the factory surface without changing `x-opaque-json` + * handling for any other consumer, so the boundary needs a cast rather than a + * conversion. + * + * Delete this along with the {@link FactoryRunResult} override once the schema + * distinguishes opaque JSON values from opaque in-process values — + * github/copilot-agent-runtime#14122. + */ +function toPublicFactoryRunResult(envelope: WireFactoryRunResult): FactoryRunResult { + return envelope as FactoryRunResult; +} + +async function awaitFactoryOperation( + operation: () => Promise, + signal: AbortSignal +): Promise { + // The operation is a thunk so an already-aborted run never dispatches the + // RPC at all, rather than sending it and rejecting locally afterwards. + let rejectAbort: ((reason?: unknown) => void) | undefined; + const abortPromise = new Promise((_resolve, reject) => { + rejectAbort = reject; + }); + const onAbort = () => + rejectAbort?.(signal.reason ?? new DOMException("Factory run was aborted", "AbortError")); + // Register before the abort check and before dispatching, so an abort can + // neither be missed by a not-yet-attached listener nor start work on an + // already-cancelled run. + signal.addEventListener("abort", onAbort, { once: true }); + try { + throwIfFactoryAborted(signal); + return await Promise.race([operation(), abortPromise]); + } 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" + ); +} + +/** + * Errors a factory combinator must never swallow into a `null` item. + * + * Cooperative cancellation aborts the run, and a rejected RPC is a hard + * runtime failure — a reached limit, a durable-state failure, or a dropped + * transport — that must terminate the run rather than be reported as a + * successfully-`null` item. An ordinary subagent failure does not reject; the + * runtime already resolves it as `null`. + */ +function isFactoryFatalError(error: unknown): boolean { + return ( + isFactoryAbortError(error) || + error instanceof ResponseError || + error instanceof ConnectionError + ); +} + /** Assistant message event - the final response from the assistant. */ export type AssistantMessageEvent = Extract; @@ -147,6 +411,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; @@ -164,6 +430,153 @@ 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; + if (options?.resumeFromRunId !== undefined) { + return this.factory.resume(options.resumeFromRunId, { + limits: options.limits, + }); + } + const envelope = await this.rpc.factory.run({ + name, + args: (options?.args === undefined + ? {} + : options.args) as FactoryRunRequest["args"], + options: { + limits: options?.limits, + }, + }); + + return toPublicFactoryRunResult(envelope); + }) as SessionFactoryApi["run"], + resume: (async (runId: string, options?: Parameters[1]) => { + let response; + try { + response = await this.rpc.factory.resume({ + runId, + limits: options?.limits, + }); + } catch (error) { + if ( + error instanceof ResponseError && + typeof error.data === "object" && + error.data !== null + ) { + const code = (error.data as { code?: unknown }).code; + if (isFactoryResumeErrorCode(code)) { + throw new FactoryResumeError(code, error.message); + } + } + throw error; + } + return toPublicFactoryRunResult(response.run); + }) as SessionFactoryApi["resume"], + getRun: async (runId) => toPublicFactoryRunResult(await this.rpc.factory.getRun({ runId })), + waitForRun: (runId, options) => this.waitForFactoryRun(runId, options?.signal), + listRuns: async () => (await this.rpc.factory.listRuns()).runs, + getRunDetail: (runId) => this.rpc.factory.getRunDetail({ runId }), + getRunProgress: (runId, options = {}) => + this.rpc.factory.getRunProgress({ runId, ...options }), + cancel: async (runId) => toPublicFactoryRunResult(await this.rpc.factory.cancel({ runId })), + }; + + /** + * Resolve when a factory run reaches a terminal status. + * + * The subscription is installed *before* the first read so a transition + * landing between the two cannot be missed, and re-reads are serialized so + * overlapping invalidation events cannot interleave — the run's revision + * advances once per operation, so a burst of events is common and must + * collapse into a single in-flight read. A bounded periodic re-read keeps a + * dropped invalidation from leaving the wait pending forever. + */ + private waitForFactoryRun(runId: string, signal?: AbortSignal): Promise { + const abortError = (): unknown => + signal?.reason ?? new DOMException("Factory run wait was aborted", "AbortError"); + if (signal?.aborted === true) { + return Promise.reject(abortError()); + } + + return new Promise((resolve, reject) => { + let settled = false; + let reading = false; + let rereadRequested = false; + let pollHandle: ReturnType | undefined; + let unsubscribe: (() => void) | undefined; + let onAbort: (() => void) | undefined; + + const finish = (complete: () => void): void => { + if (settled) { + return; + } + settled = true; + if (pollHandle !== undefined) { + clearInterval(pollHandle); + } + unsubscribe?.(); + if (onAbort !== undefined) { + signal?.removeEventListener("abort", onAbort); + } + complete(); + }; + + const read = async (): Promise => { + if (settled) { + return; + } + if (reading) { + rereadRequested = true; + return; + } + reading = true; + try { + do { + rereadRequested = false; + const envelope = await this.rpc.factory.getRun({ runId }); + if (isFactoryRunTerminal(envelope.status)) { + finish(() => resolve(toPublicFactoryRunResult(envelope))); + return; + } + } while (rereadRequested && !settled); + } catch (error) { + finish(() => reject(error)); + } finally { + reading = false; + } + }; + + if (signal !== undefined) { + onAbort = (): void => finish(() => reject(abortError())); + signal.addEventListener("abort", onAbort, { once: true }); + } + + unsubscribe = this.on("factory.run_updated", (event) => { + if (event.data.runId === runId) { + void read(); + } + }); + + pollHandle = setInterval(() => void read(), 5_000); + // The re-read is a safety net, not work the process owes anyone: an + // outstanding wait must never keep Node alive on its own. + pollHandle.unref?.(); + void read(); + }); + } + /** * Creates a new CopilotSession instance. * @@ -367,6 +780,13 @@ export class CopilotSession { this.autoModeSwitchHandler = undefined; this.commandHandlers.clear(); this.canvases.clear(); + this.factories.clear(); + for (const controllersForRun of this.factoryAbortControllers.values()) { + for (const controller of controllersForRun.values()) { + controller.abort(); + } + } + this.factoryAbortControllers.clear(); this.transformCallbacks?.clear(); } @@ -881,6 +1301,179 @@ 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(); + // Keyed by execution token as well as run ID so overlapping + // attempts for one run stay individually addressable. + let controllersForRun = self.factoryAbortControllers.get(params.runId); + if (controllersForRun === undefined) { + controllersForRun = new Map(); + self.factoryAbortControllers.set(params.runId, controllersForRun); + } + controllersForRun.set(params.executionToken, controller); + const progress = new FactoryProgressBuffer(async (lines) => { + await self.rpc.factory.log({ + runId: params.runId, + executionToken: params.executionToken, + lines, + }); + }); + try { + const context: FactoryContext = { + runId: params.runId, + args: params.args as JsonValue, + 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, + executionToken: params.executionToken, + prompt, + opts: { + label: options.label, + schema: options.schema, + model: options.model, + }, + }), + controller.signal + ); + return response.result ?? null; + }, + step: async ( + key: string, + producer: () => Promise | JsonValue, + options: FactoryStepOptions = {} + ): Promise => { + await progress.flush(); + if (options.volatile) { + // The flush above is an await point, so an abort can land + // between entering step() and running the producer. The + // journaled branch is covered by awaitFactoryOperation; + // this one has to check for itself, or a cancelled run + // would still start new extension work. + throwIfFactoryAborted(controller.signal); + return producer(); + } + const cached = await awaitFactoryOperation( + () => + self.rpc.factory.journal.get({ + runId: params.runId, + executionToken: params.executionToken, + key, + }), + controller.signal + ); + if (cached.hit) { + if (cached.resultJson === undefined) { + throw new Error( + `step("${key}") journal returned a hit without a result` + ); + } + assertFactoryStepResult(cached.resultJson, key); + return cached.resultJson; + } + + // Producers are best-effort at-least-once across crashes or + // concurrent callers, so authors must make side effects idempotent. + const result = await producer(); + assertFactoryStepResult(result, key); + await awaitFactoryOperation( + () => + self.rpc.factory.journal.put({ + runId: params.runId, + executionToken: params.executionToken, + key, + resultJson: + result as FactoryJournalPutRequest["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); + if (result === undefined) { + return {}; + } + assertFactoryResult(result); + return { result } as FactoryExecuteResult; + } finally { + try { + await progress.close(); + } finally { + const controllersForRun = self.factoryAbortControllers.get(params.runId); + if (controllersForRun?.get(params.executionToken) === controller) { + controllersForRun.delete(params.executionToken); + if (controllersForRun.size === 0) { + self.factoryAbortControllers.delete(params.runId); + } + } + } + } + }, + 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); + } + } + return {}; + }, + }; + } + /** * Registers per-provider {@link BearerTokenProvider} callbacks for BYOK providers * configured with managed-identity / on-demand bearer-token auth. @@ -1460,3 +2053,201 @@ function toCanvasRpcError(error: unknown): ResponseError { const message = error instanceof Error ? error.message : String(error); return new ResponseError(ErrorCodes.InternalError, message, { code, message }); } + +type FactoryResultValidationCategory = + | "unsupported_type" + | "non_finite_number" + | "negative_zero" + | "cyclic_value" + | "nested_undefined" + | "unsupported_object"; + +interface StrictJsonValidationContext { + code: "factory_result_not_json" | "factory_step_not_json"; + label: string; + allowTopLevelUndefined: boolean; +} + +function strictJsonValidationError( + context: StrictJsonValidationContext, + category: FactoryResultValidationCategory, + message: string, + path: string +): ResponseError<{ code: string; category: FactoryResultValidationCategory; path: string }> { + return new ResponseError(ErrorCodes.InternalError, message, { + code: context.code, + category, + path, + }); +} + +function assertStrictJson( + value: unknown, + context: StrictJsonValidationContext +): asserts value is JsonValue | undefined { + const ancestors = new Set(); + + const visit = (current: unknown, path: string, allowUndefined: boolean): void => { + if (current === undefined) { + if (allowUndefined) { + return; + } + throw strictJsonValidationError( + context, + "nested_undefined", + `${context.label} contains nested undefined at ${path}`, + path + ); + } + if (current === null || typeof current === "boolean" || typeof current === "string") { + return; + } + if (typeof current === "number") { + if (!Number.isFinite(current)) { + throw strictJsonValidationError( + context, + "non_finite_number", + `${context.label} contains a non-finite number at ${path}`, + path + ); + } + // JSON serializes -0 as "0", so a journaled -0 would come back as 0 + // after a resume and break the lossless replay guarantee. + if (Object.is(current, -0)) { + throw strictJsonValidationError( + context, + "negative_zero", + `${context.label} contains negative zero at ${path}; normalize it to 0`, + path + ); + } + return; + } + if ( + typeof current === "function" || + typeof current === "symbol" || + typeof current === "bigint" + ) { + throw strictJsonValidationError( + context, + "unsupported_type", + `${context.label} contains a function, symbol, or BigInt at ${path}`, + path + ); + } + if (typeof current !== "object") { + throw strictJsonValidationError( + context, + "unsupported_type", + `${context.label} contains a function, symbol, or BigInt at ${path}`, + path + ); + } + if (ancestors.has(current)) { + throw strictJsonValidationError( + context, + "cyclic_value", + `${context.label} contains a cyclic reference at ${path}`, + path + ); + } + + ancestors.add(current); + try { + if (Array.isArray(current)) { + const keys = Reflect.ownKeys(current); + if ( + keys.length !== current.length + 1 || + keys.some( + (key) => + key !== "length" && + (typeof key !== "string" || + !/^(0|[1-9]\d*)$/.test(key) || + Number(key) >= current.length) + ) + ) { + throw strictJsonValidationError( + context, + "unsupported_object", + `${context.label} contains a non-JSON array property at ${path}`, + path + ); + } + for (let index = 0; index < current.length; index++) { + const descriptor = Object.getOwnPropertyDescriptor(current, String(index)); + if ( + descriptor === undefined || + !descriptor.enumerable || + !("value" in descriptor) + ) { + throw strictJsonValidationError( + context, + "unsupported_object", + `${context.label} contains a non-JSON array property at ${path}[${index}]`, + `${path}[${index}]` + ); + } + visit(descriptor.value, `${path}[${index}]`, false); + } + return; + } + + const prototype = Object.getPrototypeOf(current); + if (prototype !== Object.prototype && prototype !== null) { + throw strictJsonValidationError( + context, + "unsupported_object", + `${context.label} contains a non-JSON object at ${path}`, + path + ); + } + for (const key of Reflect.ownKeys(current)) { + if (typeof key === "symbol") { + throw strictJsonValidationError( + context, + "unsupported_type", + `${context.label} contains a function, symbol, or BigInt at ${path}`, + path + ); + } + const propertyPath = /^[A-Za-z_$][\w$]*$/.test(key) + ? `${path}.${key}` + : `${path}[${JSON.stringify(key)}]`; + const descriptor = Object.getOwnPropertyDescriptor(current, key); + if ( + descriptor === undefined || + !descriptor.enumerable || + !("value" in descriptor) + ) { + throw strictJsonValidationError( + context, + "unsupported_object", + `${context.label} contains a non-JSON property at ${propertyPath}`, + propertyPath + ); + } + visit(descriptor.value, propertyPath, false); + } + } finally { + ancestors.delete(current); + } + }; + + visit(value, "$", context.allowTopLevelUndefined); +} + +function assertFactoryResult(value: unknown): asserts value is JsonValue | undefined { + assertStrictJson(value, { + code: "factory_result_not_json", + label: "Factory result", + allowTopLevelUndefined: true, + }); +} + +function assertFactoryStepResult(value: unknown, key: string): asserts value is JsonValue { + assertStrictJson(value, { + code: "factory_step_not_json", + label: `Factory step "${key}" result`, + allowTopLevelUndefined: false, + }); +} diff --git a/nodejs/src/types.ts b/nodejs/src/types.ts index 9084e7db1c..31a1afed88 100644 --- a/nodejs/src/types.ts +++ b/nodejs/src/types.ts @@ -1892,6 +1892,44 @@ 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; + /** Maximum AI credits consumed by factory subagents and descendants. This post-paid ceiling is soft. */ + maxAiCredits?: number; + /** + * Maximum accumulated active-execution time, in seconds. Active execution includes the entire extension body, + * subprocess waits, queued-agent waits, and sleeps. The limit is armed from the remaining headroom when a run + * resumes; time between attempts is not counted. Must be finite and positive when present. + */ + timeoutSeconds?: 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/e2e/factory.e2e.test.ts b/nodejs/test/e2e/factory.e2e.test.ts new file mode 100644 index 0000000000..547ecbbd5d --- /dev/null +++ b/nodejs/test/e2e/factory.e2e.test.ts @@ -0,0 +1,77 @@ +import { execFileSync } from "node:child_process"; +import { existsSync } from "node:fs"; +import { copyFile, mkdir } from "node:fs/promises"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, it } from "vitest"; +import { approveAll } from "../../src/index.js"; +import { + createSdkTestContext, + DEFAULT_GITHUB_TOKEN, + isInProcessTransport, +} from "./harness/sdkTestContext.js"; +import { retry } from "./harness/sdkTestHelper.js"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const factoryTestContext = isInProcessTransport + ? undefined + : await createSdkTestContext({ + copilotClientOptions: { + env: { + COPILOT_CLI_ENABLED_FEATURE_FLAGS: "EXTENSIONS,AGENT_FACTORIES", + }, + }, + }); + +it.skipIf(isInProcessTransport)( + "runs an extension-authored factory across the SDK process boundary", + async () => { + if (!factoryTestContext) { + throw new Error("Factory E2E requires the stdio transport"); + } + const { copilotClient, openAiEndpoint, workDir } = factoryTestContext; + + await openAiEndpoint.setCopilotUserByToken(DEFAULT_GITHUB_TOKEN, { + login: "factory-e2e-user", + copilot_plan: "individual_pro", + token_based_billing: true, + }); + + const extensionDir = join(workDir, ".github", "extensions", "factory-smoke"); + const readyFile = join(extensionDir, "ready"); + await mkdir(extensionDir, { recursive: true }); + await copyFile( + join(__dirname, "fixtures", "factory-extension.mjs"), + join(extensionDir, "extension.mjs") + ); + execFileSync("git", ["init", "--quiet"], { cwd: workDir }); + + await using session = await copilotClient.createSession({ + requestExtensions: true, + extensionSdkPath: resolve(__dirname, "..", "..", "dist"), + onPermissionRequest: approveAll, + onElicitationRequest: async () => ({ + action: "accept", + content: { action: "approve" }, + }), + }); + + await retry( + "wait for the factory extension to join the session", + async () => { + expect(existsSync(readyFile)).toBe(true); + }, + 300, + 100 + ); + + const result = await session.factory.run("argument-echo", { + args: { source: "sdk-e2e", count: 11 }, + }); + + expect(result).toMatchObject({ + status: "completed", + result: { source: "sdk-e2e", count: 11 }, + }); + } +); diff --git a/nodejs/test/e2e/fixtures/factory-extension.mjs b/nodejs/test/e2e/fixtures/factory-extension.mjs new file mode 100644 index 0000000000..fab95a90f1 --- /dev/null +++ b/nodejs/test/e2e/fixtures/factory-extension.mjs @@ -0,0 +1,14 @@ +import { writeFileSync } from "node:fs"; +import { defineFactory, joinSession } from "@github/copilot-sdk/extension"; + +const argumentEcho = defineFactory({ + meta: { + name: "argument-echo", + description: "Return the invocation arguments verbatim.", + phases: [], + }, + run: async ({ args }) => args, +}); + +await joinSession({ factories: [argumentEcho] }); +writeFileSync(new URL("./ready", import.meta.url), "ready"); diff --git a/nodejs/test/extension.test.ts b/nodejs/test/extension.test.ts index 1baa83a3ab..e94ad2204f 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 0000000000..0282cc4a58 --- /dev/null +++ b/nodejs/test/factory.test.ts @@ -0,0 +1,1954 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + *--------------------------------------------------------------------------------------------*/ + +import { readFileSync } from "node:fs"; +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, + FactoryResumeError, + isFactoryRunTerminal, + type FactoryAgentOptions, + type FactoryContext, + type FactoryDefinition, + type JsonValue, +} from "../src/factory.js"; + +/** Builds a `factory.run_updated` invalidation event for a run. */ +function runUpdatedEvent(runId: string, revision: number): Record { + return { + type: "factory.run_updated", + id: `event-${runId}-${revision}`, + parentId: null, + timestamp: new Date().toISOString(), + ephemeral: true, + data: { runId, revision }, + }; +} + +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).toEqual(meta); + expect(handle.meta).not.toBe(meta); + expect(Object.isFrozen(handle)).toBe(true); + expect(Object.isFrozen(handle.meta)).toBe(true); + + // The handle holds a snapshot, so mutating the caller's object after + // registration cannot desynchronize the advertised metadata. + meta.name = "mutated"; + (meta.phases as string[]).push("late"); + expect(handle.meta.name).toBe("no-limits"); + expect(handle.meta.phases).toEqual([]); + meta.name = "no-limits"; + meta.phases.length = 0; + + // The stored metadata is deep-frozen, so the handle's view of it must be + // readonly all the way down. Assert both halves: the mutation is a type + // error, and it also throws at runtime. + expect(() => { + // @ts-expect-error handle.meta is deeply readonly. + handle.meta.name = "mutated"; + }).toThrow(TypeError); + expect(() => { + // @ts-expect-error handle.meta.phases is a readonly array. + handle.meta.phases.push({ title: "late" }); + }).toThrow(TypeError); + + 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", + executionToken: "execution-token", + args: { value: 42 }, + }); + + expect(run).toHaveBeenCalledOnce(); + expect(result).toEqual({ result: { value: 42 } }); + }); + + it.each([ + [[{ title: "" }], "must not be empty"], + [[{ title: "Inspect" }, { title: "Inspect" }], "declared more than once"], + ])("rejects invalid declared phase titles", (phases, message) => { + expect(() => + defineFactory({ + meta: { + name: "invalid-phases", + description: "Invalid phase metadata", + phases, + }, + run: async () => {}, + }) + ).toThrow(message); + }); + + it("returns an absent execute result for a void factory", async () => { + const factory = defineFactory({ + meta: { + name: "void-result", + description: "Returns no result", + phases: [], + }, + run: async () => {}, + }); + const session = new CopilotSession("session-void-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "void-result", + runId: "run-void-result", + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({}); + }); + + it.each([42, "factory-result", [1, "two", false]])( + "returns non-object JSON factory result %j", + async (factoryResult) => { + const factory = defineFactory({ + meta: { + name: "json-result", + description: "Returns any JSON value", + phases: [], + }, + run: async () => factoryResult, + }); + const session = new CopilotSession("session-json-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "json-result", + runId: "run-json-result", + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({ result: factoryResult }); + } + ); + + it.each([ + ["function", { nested: () => undefined }, "$.nested"], + ["symbol", [Symbol("invalid")], "$[0]"], + ["BigInt", { nested: 1n }, "$.nested"], + ])("rejects a %s anywhere in a factory result", async (_label, factoryResult, expectedPath) => { + const factory = defineFactory({ + meta: { + name: "unsupported-result", + description: "Returns an unsupported value", + phases: [], + }, + run: async () => factoryResult as never, + }); + const session = new CopilotSession("session-unsupported-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "unsupported-result", + runId: "run-unsupported-result", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + message: `Factory result contains a function, symbol, or BigInt at ${expectedPath}`, + data: { + code: "factory_result_not_json", + category: "unsupported_type", + }, + }); + }); + + it.each([ + ["NaN", Number.NaN], + ["Infinity", Number.POSITIVE_INFINITY], + ])("rejects the non-finite number %s in a factory result", async (_label, value) => { + const factory = defineFactory({ + meta: { + name: "non-finite-result", + description: "Returns a non-finite number", + phases: [], + }, + run: async () => ({ value }) as never, + }); + const session = new CopilotSession("session-non-finite-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "non-finite-result", + runId: "run-non-finite-result", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + message: "Factory result contains a non-finite number at $.value", + data: { + code: "factory_result_not_json", + category: "non_finite_number", + }, + }); + }); + + it("rejects a cyclic factory result", async () => { + const factoryResult: Record = {}; + factoryResult.self = factoryResult; + const factory = defineFactory({ + meta: { + name: "cyclic-result", + description: "Returns a cycle", + phases: [], + }, + run: async () => factoryResult as never, + }); + const session = new CopilotSession("session-cyclic-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "cyclic-result", + runId: "run-cyclic-result", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + message: "Factory result contains a cyclic reference at $.self", + data: { + code: "factory_result_not_json", + category: "cyclic_value", + }, + }); + }); + + it.each([ + ["object", { nested: undefined }, "$.nested"], + ["array", [undefined], "$[0]"], + ])( + "rejects nested undefined in a factory result %s", + async (_label, factoryResult, expectedPath) => { + const factory = defineFactory({ + meta: { + name: "nested-undefined-result", + description: "Returns nested undefined", + phases: [], + }, + run: async () => factoryResult as never, + }); + const session = new CopilotSession("session-nested-undefined-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "nested-undefined-result", + runId: "run-nested-undefined-result", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + message: `Factory result contains nested undefined at ${expectedPath}`, + data: { + code: "factory_result_not_json", + category: "nested_undefined", + }, + }); + } + ); + + 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], + ["timeoutSeconds", 0], + ["timeoutSeconds", Number.NaN], + ["timeoutSeconds", Number.POSITIVE_INFINITY], + ["maxAiCredits", 0], + ["maxAiCredits", Number.NaN], + ["maxAiCredits", Number.POSITIVE_INFINITY], + ["maxAiCredits", 0.000_000_000_4], + ["maxAiCredits", (Number.MAX_SAFE_INTEGER + 2) / 1_000_000_000], + ] 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("accepts positive fractional timeoutSeconds through the Node timer ceiling", () => { + for (const timeoutSeconds of [0.001, 1.5, 2_147_483.647]) { + expect(() => + defineFactory({ + meta: { + name: `accepted-timeout-${timeoutSeconds}`, + description: "Factory with an accepted active-execution timeout", + phases: [], + limits: { timeoutSeconds }, + }, + run: async () => null, + }) + ).not.toThrow(); + } + }); + + it("accepts AI-credit ceilings that round to a positive safe nano-AIU integer", () => { + for (const maxAiCredits of [ + 0.000_000_000_5, + 1.25, + Number.MAX_SAFE_INTEGER / 1_000_000_000, + ]) { + expect(() => + defineFactory({ + meta: { + name: `accepted-credits-${maxAiCredits}`, + description: "Factory with an accepted AI-credit ceiling", + phases: [], + limits: { maxAiCredits }, + }, + run: async () => null, + }) + ).not.toThrow(); + } + }); + + it("rejects timeoutSeconds above the Node setTimeout ceiling", () => { + const definition = { + meta: { + name: "oversized-timeout", + description: "Factory with an out-of-range timeout", + phases: [], + limits: { timeoutSeconds: 2_147_483.648 }, + }, + run: async () => null, + } as FactoryDefinition; + + expect(() => defineFactory(definition)).toThrow( + 'Factory limit "timeoutSeconds" must not exceed 2147483.647 seconds' + ); + }); + + it("documents timeoutSeconds as accumulated active-execution time in public and generated types", () => { + const publicTypes = readFileSync(new URL("../src/types.ts", import.meta.url), "utf8"); + const generatedRpc = readFileSync( + new URL("../src/generated/rpc.ts", import.meta.url), + "utf8" + ); + + expect(publicTypes).toContain("Maximum accumulated active-execution time, in seconds."); + expect(publicTypes).toContain("subprocess waits, queued-agent waits, and sleeps"); + expect(publicTypes).toContain("timeoutSeconds?: number;"); + expect(generatedRpc).toContain("Maximum accumulated active-execution time in seconds."); + expect(generatedRpc).toContain("subprocess waits, queued-agent waits, and sleeps"); + expect(generatedRpc).toContain("timeoutSeconds?: number;"); + }); + + 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<{ + runId: string; + 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", + executionToken: "execution-token", + args: { value: 42 }, + }); + const context = await contextSeen.promise; + + expect(context.runId).toBe("run-context"); + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + args: {}, + }); + await vi.waitFor(() => { + expect(sendRequest).toHaveBeenCalledWith("session.factory.log", { + sessionId: session.sessionId, + runId: "run-live-progress", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({ result: "pong" }); + expect(sendRequest).toHaveBeenCalledWith("session.factory.agent", { + sessionId: session.sessionId, + factoryRunId: "run-agent", + executionToken: "execution-token", + prompt: "Reply with pong", + opts: { + label: "Pong helper", + model: "gpt-test", + schema: { type: "string" }, + }, + }); + }); + + it("keeps each execution token on callbacks from overlapping contexts with the same run id", async () => { + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.agent") { + return { result: "agent result" }; + } + if (method === "session.factory.journal.get") { + return { hit: false }; + } + return {}; + }); + const session = new CopilotSession("session-overlapping-attempts", { + sendRequest, + } as never); + const contexts: FactoryContext[] = []; + const bodies = [Promise.withResolvers(), Promise.withResolvers()]; + const contextsReady = Promise.withResolvers(); + const factory = defineFactory({ + meta: { + name: "overlapping-attempts", + description: "Execution token capture test", + phases: [], + }, + run: async (context) => { + const invocation = contexts.length; + contexts.push(context); + if (contexts.length === 2) { + contextsReady.resolve(); + } + await bodies[invocation].promise; + return `attempt ${invocation + 1}`; + }, + }); + session.registerFactories([factory]); + const first = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "overlapping-attempts", + runId: "shared-run", + executionToken: "old-token", + args: {}, + }); + const second = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "overlapping-attempts", + runId: "shared-run", + executionToken: "current-token", + args: {}, + }); + await contextsReady.promise; + + contexts[0].log("stale log"); + await contexts[0].agent("stale agent"); + await contexts[0].step("stale journal", () => "stale result"); + await contexts[1].agent("current agent"); + + expect(sendRequest).toHaveBeenCalledWith( + "session.factory.log", + expect.objectContaining({ executionToken: "old-token" }) + ); + expect(sendRequest).toHaveBeenCalledWith( + "session.factory.agent", + expect.objectContaining({ executionToken: "old-token", prompt: "stale agent" }) + ); + expect(sendRequest).toHaveBeenCalledWith( + "session.factory.journal.get", + expect.objectContaining({ executionToken: "old-token", key: "stale journal" }) + ); + expect(sendRequest).toHaveBeenCalledWith( + "session.factory.journal.put", + expect.objectContaining({ executionToken: "old-token", key: "stale journal" }) + ); + expect(sendRequest).toHaveBeenCalledWith( + "session.factory.agent", + expect.objectContaining({ executionToken: "current-token", prompt: "current agent" }) + ); + + bodies[0].resolve(); + bodies[1].resolve(); + await expect(first).resolves.toEqual({ result: "attempt 1" }); + await expect(second).resolves.toEqual({ result: "attempt 2" }); + }); + + 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", + executionToken: "execution-token", + 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.each([ + ["undefined", () => undefined], + ["NaN", () => Number.NaN], + ["Infinity", () => Number.POSITIVE_INFINITY], + ["function", () => () => undefined], + ["symbol", () => Symbol("invalid")], + ["BigInt", () => 1n], + [ + "cycle", + () => { + const value: Record = {}; + value.self = value; + return value; + }, + ], + ["non-plain object", () => new Date()], + [ + "accessor property", + () => Object.defineProperty({}, "value", { enumerable: true, get: () => "hidden" }), + ], + [ + "non-enumerable property", + () => Object.defineProperty({}, "value", { enumerable: false, value: "hidden" }), + ], + ["array hole", () => new Array(1)], + [ + "array accessor", + () => Object.defineProperty([], "0", { enumerable: true, get: () => "hidden" }), + ], + ["array extra key", () => Object.assign([1], { extra: "dropped" })], + ])("rejects a journaled step %s result", async (_label, makeValue) => { + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.journal.get") { + return { hit: false }; + } + throw new Error(`Unexpected method: ${method}`); + }); + const session = new CopilotSession("session-invalid-step", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "invalid-step", + description: "Rejects lossy step values", + phases: [], + }, + run: async ({ step }) => { + await step("invalid", async () => makeValue() as never); + return "must-not-complete"; + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "invalid-step", + runId: "run-invalid-step", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + data: { + code: "factory_step_not_json", + }, + }); + expect( + sendRequest.mock.calls.filter(([method]) => method === "session.factory.journal.put") + ).toHaveLength(0); + }); + + it("validates a journaled step cache hit before replay", async () => { + const cached = Object.assign([1], { extra: "dropped" }); + const producer = vi.fn(async () => "must-not-run"); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.journal.get") { + return { hit: true, resultJson: cached }; + } + throw new Error(`Unexpected method: ${method}`); + }); + const session = new CopilotSession("session-invalid-step-cache", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "invalid-step-cache", + description: "Rejects invalid cached values", + phases: [], + }, + run: async ({ step }) => step("cached", producer), + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "invalid-step-cache", + runId: "run-invalid-step-cache", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + data: { + code: "factory_step_not_json", + category: "unsupported_object", + }, + }); + expect(producer).not.toHaveBeenCalled(); + }); + + it("replays a journaled step value identically on resume", 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 producer = vi.fn(async () => ({ nested: [1, null, "same"] })); + const session = new CopilotSession("session-step-replay", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "step-replay", + description: "Replays strict JSON", + phases: [], + }, + run: async ({ step }) => step("same", producer), + }); + session.registerFactories([factory]); + + const first = await session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "step-replay", + runId: "run-step-replay", + executionToken: "execution-token", + args: {}, + }); + const replay = await session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "step-replay", + runId: "run-step-replay", + executionToken: "execution-token", + args: {}, + }); + + expect(replay).toEqual(first); + expect(producer).toHaveBeenCalledOnce(); + }); + + it("bypasses validation and journaling for a volatile step", async () => { + const sendRequest = vi.fn(); + const session = new CopilotSession("session-volatile-step", { sendRequest } as never); + const factory = defineFactory({ + meta: { + name: "volatile-step", + description: "Allows author-opted-out volatile values", + phases: [], + }, + run: async ({ step }) => { + const value = await step("volatile", async () => (() => "not JSON") as never, { + volatile: true, + }); + expect(typeof value).toBe("function"); + return "completed"; + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "volatile-step", + runId: "run-volatile-step", + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({ result: "completed" }); + expect(sendRequest).not.toHaveBeenCalled(); + }); + + it("does not start a volatile step producer after the run is aborted", async () => { + const sendRequest = vi.fn(); + const session = new CopilotSession("session-volatile-abort", { sendRequest } as never); + let producerRan = false; + const factory = defineFactory({ + meta: { + name: "volatile-abort", + description: "Volatile steps honour cancellation", + phases: [], + }, + run: async ({ step, runId }) => { + // Abort mid-run, then attempt a volatile step. The producer must + // not run: cancellation has to stop new extension work starting, + // exactly as it does on the journaled path. + await session.clientSessionApis.factory!.abort({ + sessionId: session.sessionId, + runId, + }); + await step( + "volatile", + () => { + producerRan = true; + return "should not happen"; + }, + { volatile: true } + ); + return "completed"; + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "volatile-abort", + runId: "run-volatile-abort", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toThrow(); + expect(producerRan).toBe(false); + }); + + it("rejects a factory result array with an extra own key", async () => { + const factory = defineFactory({ + meta: { + name: "array-extra-result", + description: "Rejects lossy array keys", + phases: [], + }, + run: async () => Object.assign([1], { extra: 1n }) as never, + }); + const session = new CopilotSession("session-array-extra-result", {} as never); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "array-extra-result", + runId: "run-array-extra-result", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toMatchObject({ + data: { + code: "factory_result_not_json", + category: "unsupported_object", + }, + }); + }); + + 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 observability methods and forwards paging options", async () => { + const summary = { + runId: "run-observe", + factoryName: "observe", + description: "Observe", + status: "running" as const, + revision: 4, + createdAt: 1, + startedAt: 2, + updatedAt: 3, + completedAt: null, + currentPhase: { id: "p0", ordinal: 0 }, + declaredPhaseCount: 1, + liveAgentCount: 1, + totalSpawnedAgentCount: 1, + consumed: { activeMs: 10, subagents: 1, nanoAiu: 5 }, + declaredLimits: {}, + approved: {}, + observedAt: 4, + activeSegmentStartedAt: 2, + terminal: null, + }; + const progress = { + records: [], + oldestSeq: null, + newestSeq: null, + hasMoreOlder: false, + hasMoreNewer: false, + revision: 4, + }; + const detail = { ...summary, phases: [], agents: [], progress }; + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.listRuns") return { runs: [summary] }; + if (method === "session.factory.getRunDetail") return detail; + return progress; + }); + const session = new CopilotSession("session-observe", { sendRequest } as never); + + await expect(session.factory.listRuns()).resolves.toEqual([summary]); + await expect(session.factory.getRunDetail("run-observe")).resolves.toEqual(detail); + await expect( + session.factory.getRunProgress("run-observe", { + phaseId: "p0", + afterSeq: 10, + limit: 50, + }) + ).resolves.toEqual(progress); + expect(sendRequest).toHaveBeenNthCalledWith(1, "session.factory.listRuns", { + sessionId: session.sessionId, + }); + expect(sendRequest).toHaveBeenNthCalledWith(2, "session.factory.getRunDetail", { + sessionId: session.sessionId, + runId: "run-observe", + }); + expect(sendRequest).toHaveBeenNthCalledWith(3, "session.factory.getRunProgress", { + sessionId: session.sessionId, + runId: "run-observe", + phaseId: "p0", + afterSeq: 10, + limit: 50, + }); + }); + + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toThrow("body failed"); + expect(sendRequest).toHaveBeenCalledWith("session.factory.log", { + sessionId: session.sessionId, + runId: "run-throw-progress", + executionToken: "execution-token", + lines: [{ seq: 0, kind: "log", text: "before throw" }], + }); + }); + + it("keeps a completed execution successful when only the final progress flush fails", async () => { + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.log") { + throw new Error("final transport failure"); + } + return {}; + }); + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + const session = new CopilotSession("session-final-flush-failure", { + sendRequest, + } as never); + const factory = defineFactory({ + meta: { + name: "final-flush-failure", + description: "Final flush failure regression test", + phases: [], + }, + run: async ({ log }) => { + log("final line"); + return "done"; + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "final-flush-failure", + runId: "run-final-flush-failure", + executionToken: "execution-token", + args: {}, + }) + ).resolves.toEqual({ result: "done" }); + expect(warning).toHaveBeenCalledWith( + "Failed to flush final factory progress after the factory body settled", + expect.objectContaining({ message: "final transport failure" }) + ); + }); + + it("keeps a mid-run progress flush failure fatal", async () => { + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.factory.log") { + throw new Error("mid-run transport failure"); + } + if (method === "session.factory.agent") { + return { result: "must not complete" }; + } + return {}; + }); + const session = new CopilotSession("session-mid-run-flush-failure", { + sendRequest, + } as never); + const factory = defineFactory({ + meta: { + name: "mid-run-flush-failure", + description: "Mid-run flush failure regression test", + phases: [], + }, + run: async ({ agent, log }) => { + log("before agent"); + return agent("trigger a flush"); + }, + }); + session.registerFactories([factory]); + + await expect( + session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: "mid-run-flush-failure", + runId: "run-mid-run-flush-failure", + executionToken: "execution-token", + args: {}, + }) + ).rejects.toThrow("mid-run transport failure"); + expect(sendRequest).not.toHaveBeenCalledWith("session.factory.agent", expect.anything()); + }); + + 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", + executionToken: "execution-token", + 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", + executionToken: "execution-token", + 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.each(["parallel", "pipeline"] as const)( + "propagates cancellation out of %s instead of mapping it to null", + async (combinator) => { + 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-${combinator}`, + 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, pipeline }) => + combinator === "parallel" + ? parallel([() => agent("wait forever")]) + : pipeline(["wait forever"], (_previous, item) => agent(item as string)), + }); + session.registerFactories([factory]); + + const execution = session.clientSessionApis.factory!.execute({ + sessionId: session.sessionId, + name: `abort-${combinator}`, + runId: `run-abort-${combinator}`, + executionToken: "execution-token", + args: {}, + }); + await vi.waitFor(() => + expect(sendRequest).toHaveBeenCalledWith("session.factory.agent", expect.anything()) + ); + + await session.clientSessionApis.factory!.abort({ + sessionId: session.sessionId, + runId: `run-abort-${combinator}`, + }); + + 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", + executionToken: "execution-token", + 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 fresh factories and routes direct and legacy resumes by ID without args", 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 }) => + method === "session.factory.resume" + ? { + factoryName: "stored-name", + run: { + runId: "run-prior", + status: "completed", + result: { name: "stored-name", persistedArgs: true }, + }, + } + : { + runId: "run-foreground", + status: "completed", + result: { name: params.name }, + } + ); + const session = new CopilotSession("session-run", { sendRequest } as never); + + await expect( + session.factory.resume("run-prior", { + limits: { maxTotalSubagents: 7 }, + }) + ).resolves.toMatchObject({ + status: "completed", + result: { name: "stored-name", persistedArgs: true }, + }); + await expect( + session.factory.run("by-name", { + args: { value: 1 }, + limits: { maxTotalSubagents: 7 }, + resumeFromRunId: "run-prior", + }) + ).resolves.toMatchObject({ + status: "completed", + result: { name: "stored-name", persistedArgs: true }, + }); + await expect(session.factory.run(factory)).resolves.toMatchObject({ + status: "completed", + result: { name: "friendly-run" }, + }); + expect(sendRequest).toHaveBeenNthCalledWith(1, "session.factory.resume", { + sessionId: session.sessionId, + runId: "run-prior", + limits: { maxTotalSubagents: 7 }, + }); + expect(sendRequest).toHaveBeenNthCalledWith(2, "session.factory.resume", { + sessionId: session.sessionId, + runId: "run-prior", + limits: { maxTotalSubagents: 7 }, + }); + expect(sendRequest).toHaveBeenNthCalledWith(3, "session.factory.run", { + sessionId: session.sessionId, + name: "friendly-run", + args: {}, + options: { limits: undefined }, + }); + }); + + it("returns the full envelope for a failed foreground run", 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); + + // A run that exists resolves with its envelope; only pre-execution + // failures (no run id) reject. + await expect(session.factory.run("failing")).resolves.toEqual(envelope); + }); + + it.each([ + "not_found", + "non_resumable", + "already_active", + "reapproval_declined", + "no_approval_provider", + ] as const)( + "throws FactoryResumeError with code %s for pre-execution failures", + async (code) => { + const session = new CopilotSession("session-resume-error", { + sendRequest: vi.fn(async () => { + throw new ResponseError(-32602, `resume failed: ${code}`, { code }); + }), + } as never); + + const error = await session.factory + .resume("run-error") + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(FactoryResumeError); + expect((error as FactoryResumeError).code).toBe(code); + } + ); + + it("returns resumed execution failures as envelopes", async () => { + const envelope = { + runId: "run-execution-error", + status: "error" as const, + error: "resumed body failed", + }; + const session = new CopilotSession("session-resumed-run-error", { + sendRequest: vi.fn(async () => ({ factoryName: "stored-name", run: envelope })), + } as never); + + await expect(session.factory.resume("run-execution-error")).resolves.toEqual(envelope); + }); +}); + +describe("factory run settlement", () => { + it.each([ + ["completed", true], + ["error", true], + ["halted", true], + ["cancelled", true], + ["pending", false], + ["running", false], + ] as const)("classifies %s as terminal=%s", (status, expected) => { + expect(isFactoryRunTerminal(status)).toBe(expected); + }); + + it("resolves immediately when the run has already settled", async () => { + const envelope = { runId: "run-settled", status: "completed" as const, result: 42 }; + const sendRequest = vi.fn(async () => envelope); + const session = new CopilotSession("session-wait-settled", { sendRequest } as never); + + await expect(session.factory.waitForRun("run-settled")).resolves.toEqual(envelope); + expect(sendRequest).toHaveBeenCalledTimes(1); + expect(sendRequest).toHaveBeenCalledWith("session.factory.getRun", { + sessionId: session.sessionId, + runId: "run-settled", + }); + }); + + it("waits for a running run to reach a terminal status", async () => { + const running = { runId: "run-wait", status: "running" as const }; + const terminal = { runId: "run-wait", status: "completed" as const, result: "done" }; + let current: unknown = running; + const sendRequest = vi.fn(async () => current); + const session = new CopilotSession("session-wait-running", { sendRequest } as never); + + const settled = session.factory.waitForRun("run-wait"); + // The first read observed a running envelope, so the wait is still pending. + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(1)); + + // An invalidation event for an unrelated run must not trigger a re-read. + (session as never as { _dispatchEvent(event: unknown): void })._dispatchEvent( + runUpdatedEvent("some-other-run", 2) + ); + expect(sendRequest).toHaveBeenCalledTimes(1); + + current = terminal; + (session as never as { _dispatchEvent(event: unknown): void })._dispatchEvent( + runUpdatedEvent("run-wait", 3) + ); + + await expect(settled).resolves.toEqual(terminal); + }); + + it("periodically re-reads when a terminal invalidation is missed", async () => { + vi.useFakeTimers(); + const running = { runId: "run-poll", status: "running" as const }; + const terminal = { runId: "run-poll", status: "completed" as const, result: "polled" }; + let current: unknown = running; + const sendRequest = vi.fn(async () => current); + const session = new CopilotSession("session-wait-poll", { sendRequest } as never); + + try { + const settled = session.factory.waitForRun("run-poll"); + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(1)); + + current = terminal; + await vi.advanceTimersByTimeAsync(5_000); + + await expect(settled).resolves.toEqual(terminal); + expect(sendRequest).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it("stops watching once the run settles", async () => { + const running = { runId: "run-unsub", status: "running" as const }; + const terminal = { runId: "run-unsub", status: "error" as const, error: "body failed" }; + let current: unknown = running; + const sendRequest = vi.fn(async () => current); + const session = new CopilotSession("session-wait-unsub", { sendRequest } as never); + const handlersFor = (): Set | undefined => + ( + session as never as { + typedEventHandlers: Map>; + } + ).typedEventHandlers.get("factory.run_updated"); + + const settled = session.factory.waitForRun("run-unsub"); + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(1)); + expect(handlersFor()?.size ?? 0).toBe(1); + + current = terminal; + (session as never as { _dispatchEvent(event: unknown): void })._dispatchEvent( + runUpdatedEvent("run-unsub", 2) + ); + await expect(settled).resolves.toEqual(terminal); + + // The subscription must be released, or every completed wait leaks a + // listener for the lifetime of the session. + expect(handlersFor()?.size ?? 0).toBe(0); + + const callsAtSettlement = sendRequest.mock.calls.length; + // A late event for a settled run must not provoke another read. + (session as never as { _dispatchEvent(event: unknown): void })._dispatchEvent( + runUpdatedEvent("run-unsub", 3) + ); + expect(sendRequest).toHaveBeenCalledTimes(callsAtSettlement); + }); + + it("rejects when the signal is already aborted and never reads", async () => { + const sendRequest = vi.fn(async () => ({ runId: "run-pre", status: "running" })); + const session = new CopilotSession("session-wait-pre-abort", { sendRequest } as never); + + await expect( + session.factory.waitForRun("run-pre", { signal: AbortSignal.abort() }) + ).rejects.toThrow(); + expect(sendRequest).not.toHaveBeenCalled(); + }); + + it("rejects when aborted while waiting, leaving the run untouched", async () => { + const sendRequest = vi.fn(async () => ({ runId: "run-abort", status: "running" })); + const session = new CopilotSession("session-wait-abort", { sendRequest } as never); + const controller = new AbortController(); + + const settled = session.factory.waitForRun("run-abort", { signal: controller.signal }); + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(1)); + + controller.abort(); + await expect(settled).rejects.toThrow(); + // Aborting the wait must not cancel the run. + expect(sendRequest).not.toHaveBeenCalledWith("session.factory.cancel", expect.anything()); + }); + + it("propagates a read failure", async () => { + const sendRequest = vi.fn(async () => { + throw new Error("factory_storage_unavailable"); + }); + const session = new CopilotSession("session-wait-error", { sendRequest } as never); + + await expect(session.factory.waitForRun("run-broken")).rejects.toThrow( + "factory_storage_unavailable" + ); + }); + + it("collapses a burst of invalidation events into one in-flight read", async () => { + const running = { runId: "run-burst", status: "running" as const }; + const terminal = { runId: "run-burst", status: "completed" as const }; + let release: (() => void) | undefined; + const gate = new Promise((resolve) => (release = resolve)); + let readCount = 0; + const sendRequest = vi.fn(async () => { + readCount += 1; + if (readCount === 2) { + await gate; + } + // Reads 1 and 2 observe a running run; only the coalesced third + // read observes the terminal one. + return readCount >= 3 ? terminal : running; + }); + const session = new CopilotSession("session-wait-burst", { sendRequest } as never); + + const settled = session.factory.waitForRun("run-burst"); + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(1)); + + const dispatch = (revision: number): void => + (session as never as { _dispatchEvent(event: unknown): void })._dispatchEvent( + runUpdatedEvent("run-burst", revision) + ); + + // Second read is held open while three more events arrive; they must + // collapse into a single follow-up read rather than three. + dispatch(2); + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(2)); + dispatch(3); + dispatch(4); + dispatch(5); + expect(sendRequest).toHaveBeenCalledTimes(2); + + release?.(); + await expect(settled).resolves.toEqual(terminal); + // One initial read, the held read, and exactly one coalesced re-read + // standing in for all three queued events. + expect(sendRequest).toHaveBeenCalledTimes(3); + }); +}); diff --git a/nodejs/test/session-event-types.test.ts b/nodejs/test/session-event-types.test.ts index 37b49f8a48..f20c2db338 100644 --- a/nodejs/test/session-event-types.test.ts +++ b/nodejs/test/session-event-types.test.ts @@ -50,6 +50,9 @@ import type { UserMessageAgentMode, Attachment, WorkingDirectoryContextHostType, + FactoryContext, + FactoryDefinition, + JsonValue, } from "../src/index.js"; /** @@ -80,6 +83,17 @@ type _AssistantMessageEventStaysAlignedWithSessionEventUnion = _AssertEqual< Extract >; const _assistantMessageEventAlignmentCheck: _AssistantMessageEventStaysAlignedWithSessionEventUnion = true; +type _DefaultFactoryArgsAreJsonValue = _AssertEqual; +const _defaultFactoryArgsCheck: _DefaultFactoryArgsAreJsonValue = true; +type _DefaultFactoryResultIsJsonValueOrVoid = _AssertEqual< + Awaited>, + JsonValue | void +>; +const _defaultFactoryResultCheck: _DefaultFactoryResultIsJsonValueOrVoid = true; +// @ts-expect-error Factory arguments must be representable on the JSON wire. +type _FactoryArgsRejectUndefined = FactoryContext; +// @ts-expect-error Factory results must be JSON values or top-level void. +type _FactoryResultRejectsFunction = FactoryDefinition void>; describe("Session event type exports (#1156)", () => { it("exposes the headline ToolExecutionStartData type with a usable shape", () => { @@ -97,7 +111,7 @@ describe("Session event type exports (#1156)", () => { expect(data.toolName).toBe("shell"); expect(data.toolCallId).toBe("call-1"); - expect(data.arguments?.command).toBe("ls"); + expect(data.arguments).toEqual({ command: "ls" }); expect(data.mcpServerName).toBe("filesystem"); expect(data.mcpToolName).toBe("list_dir"); expect(data.turnId).toBe("turn-1"); diff --git a/test/harness/replayingCapiProxy.ts b/test/harness/replayingCapiProxy.ts index 5e07449f73..9b83cfe794 100644 --- a/test/harness/replayingCapiProxy.ts +++ b/test/harness/replayingCapiProxy.ts @@ -1747,6 +1747,7 @@ export type ToolResultNormalizer = { export type CopilotUserResponse = { login: string; copilot_plan?: string; + token_based_billing?: boolean; is_mcp_enabled?: boolean; endpoints?: { api?: string;