diff --git a/.changeset/late-review-thread-fixes.md b/.changeset/late-review-thread-fixes.md new file mode 100644 index 000000000..1f8144556 --- /dev/null +++ b/.changeset/late-review-thread-fixes.md @@ -0,0 +1,17 @@ +--- +"agent-bundle": patch +--- + +`agent-bundle dev` now leases the adopted epoch until another epoch replaces it +or the server closes, so store retention cannot delete the advertised last-good +build during a run of failing rebuilds, and an epoch that cannot be leased is +reported as `AB7211` instead of adopted; the `dev.contracts` matrix opens the +configured server on a target whose manifest carries it, applies the session +timeout per request, forwards each request's `_meta.progressToken` so generated +routes emit progress, and observes lifecycle progress through the session trace +(`ContractMatrixClient` from `agent-bundle/test` gains an optional +`observeProgress` seam, `ContractMatrixProgressSource`; `McpSession.callTool` +accepts `_meta`). Native Playground catalog readers wait for a hard-link +publisher to release its staging link before adopting the sidecar, return to +discovery when that publication is rolled back, and recover a staging link +abandoned by an exited publisher instead of rejecting the epoch forever. (#408) diff --git a/docs/effect-conventions.md b/docs/effect-conventions.md index abdcafb35..5ad5efd4b 100644 --- a/docs/effect-conventions.md +++ b/docs/effect-conventions.md @@ -345,12 +345,14 @@ must not regress it: `pnpm bench:hook-cold-start -- --check`. ## Parked toolchain follow-ups -Toolchain pins that are deliberately held back ride the same named chore as -the Effect RC re-pin (see `AGENTS.md`). Each row records the pin, the exact -registry state observed when the row was written (`npm view dist-tags` -/ `versions`), and the trigger that turns the row into a chore. Re-verify -every row during a re-pin; when a trigger has fired, do the upgrade in its -own chore PR and retire the row. +Toolchain pins that are deliberately held back are tracked here. Only the +`repos/effect` subtree update is coupled to the Effect RC re-pin (see +`AGENTS.md`); every other row has its own independent trigger. Each row +records the pin, the exact registry state observed when the row was written +(`npm view dist-tags` / `versions`), and the trigger that turns the row +into a chore. Re-verify every row during a re-pin, but never delay or bundle a +row whose trigger has already fired: do that upgrade in its own chore PR as +soon as the trigger fires and retire the row. | Recorded | Pin (where) | Observed registry state | Trigger / action | | --- | --- | --- | --- | diff --git a/packages/agent-bundle/README.md b/packages/agent-bundle/README.md index 265125564..17de0515d 100644 --- a/packages/agent-bundle/README.md +++ b/packages/agent-bundle/README.md @@ -35,7 +35,8 @@ assets against this anchor rather than the process working directory: Claude Cod launches stdio servers from the host's own working directory and ignores stdio `cwd` at runtime, and its placeholder-substitution table excludes `cwd`, so the Claude adapter emits no `cwd` for a plugin-root working directory (the absolute `${CLAUDE_PLUGIN_ROOT}/mcp/...` entry path plus this -env anchor carry the guarantee) and rejects token-bearing `cwd` values outright. A server's own +env anchor carry the guarantee). That canonical plugin-root `cwd` is the one accepted token-bearing +value on Claude; any other `cwd` that carries a path token is rejected. A server's own `env` entries win over the injected value, so declaring `env: { AGENT_BUNDLE_PLUGIN_ROOT: ... }` replaces the anchor. The `pluginRootEnvAnchor` export names the variable for consumer code. diff --git a/packages/agent-bundle/src/dev/dev-contract-runner.ts b/packages/agent-bundle/src/dev/dev-contract-runner.ts index aedffc9ce..b9bfae455 100644 --- a/packages/agent-bundle/src/dev/dev-contract-runner.ts +++ b/packages/agent-bundle/src/dev/dev-contract-runner.ts @@ -6,6 +6,7 @@ import { ContractMatrixViolationError, runDevEpochContractMatrix, type ContractMatrixClient, + type ContractProgressNotification, } from '../test/contract.ts'; import type { Diagnostic } from '../core/diagnostics.ts'; import { emptyCompiledRouteGraph } from '../routes/graph.ts'; @@ -45,10 +46,29 @@ const failed = ( summary, }); -const targetFor = (prepared: PreparedProject): string => { - const targets = prepared.model?.targets.map((target) => target.name) ?? []; - const target = targets.includes('portable') ? 'portable' : targets[0]; - if (target === undefined) throw new Error('Development contract matrix requires at least one generated target.'); +/** + * The generated target whose manifest carries the selected server. A server + * restricted to `targets: ['claude']` is absent from the portable manifest, so + * the choice is made over the server's own target list intersected with the + * project's; `portable` still wins whenever it is eligible. + */ +export const devContractTarget = ( + prepared: Pick, + serverName: string, +): string => { + const projectTargets = prepared.model?.targets.map((target) => target.name) ?? []; + const server = prepared.model?.mcpServers.find((candidate) => candidate.name === serverName); + const eligible = server === undefined + ? projectTargets + : projectTargets.filter((target) => server.targets.includes(target)); + const target = eligible.includes('portable') ? 'portable' : eligible[0]; + if (target === undefined) { + throw new Error( + projectTargets.length === 0 + ? 'Development contract matrix requires at least one generated target.' + : `Development contract matrix server ${JSON.stringify(serverName)} is emitted for none of the project's targets.`, + ); + } return target; }; @@ -61,40 +81,64 @@ const serverFor = (prepared: PreparedProject, requested: string | undefined): st return names[0]; }; -const matrixClient = (session: McpSession, signal: AbortSignal): ContractMatrixClient => ({ +/** The session's per-request timeout, applied afresh to each matrix request rather than once for the whole matrix. */ +const requestSignal = (session: Pick, options: { readonly signal?: AbortSignal } | undefined): AbortSignal => { + const timeout = AbortSignal.timeout(session.timeoutMs); + return options?.signal === undefined ? timeout : AbortSignal.any([timeout, options.signal]); +}; + +type MatrixSession = Pick< + McpSession, + 'callTool' | 'getPrompt' | 'listPrompts' | 'listResources' | 'listTools' | 'readResource' | 'subscribeTrace' | 'timeoutMs' | 'trace' +>; + +export const matrixClient = (session: MatrixSession): ContractMatrixClient => ({ callTool: async (params, options) => session.callTool({ + // Lifecycle fixtures pass their progress token here; generated routes only + // send progress when the request's `_meta.progressToken` is present. + ...(params._meta === undefined ? {} : { _meta: params._meta }), arguments: params.arguments ?? {}, name: params.name, - signal: options?.signal === undefined ? signal : AbortSignal.any([signal, options.signal]), + signal: requestSignal(session, options), ...(options?.timeout === undefined ? {} : { timeoutMs: options.timeout }), }), getPrompt: async (params, options) => session.getPrompt({ ...(params.arguments === undefined ? {} : { arguments: params.arguments }), name: params.name, - signal: options?.signal === undefined ? signal : AbortSignal.any([signal, options.signal]), + signal: requestSignal(session, options), ...(options?.timeout === undefined ? {} : { timeoutMs: options.timeout }), }), listPrompts: async (_params, options) => ({ prompts: [...await session.listPrompts({ - signal: options?.signal === undefined ? signal : AbortSignal.any([signal, options.signal]), + signal: requestSignal(session, options), ...(options?.timeout === undefined ? {} : { timeoutMs: options.timeout }), })], }), listResources: async (_params, options) => ({ resources: [...await session.listResources({ - signal: options?.signal === undefined ? signal : AbortSignal.any([signal, options.signal]), + signal: requestSignal(session, options), ...(options?.timeout === undefined ? {} : { timeoutMs: options.timeout }), })], }), listTools: async (_params, options) => ({ tools: [...await session.listTools({ - signal: options?.signal === undefined ? signal : AbortSignal.any([signal, options.signal]), + signal: requestSignal(session, options), ...(options?.timeout === undefined ? {} : { timeoutMs: options.timeout }), })], }), + // Lifecycle fixtures count live progress; the session records every server + // notification in its trace synchronously on receipt, so a live trace + // subscription is the supported notification path for this non-SDK client. + observeProgress: (listener) => { + const latest = session.trace().entries.at(-1)?.sequence ?? 0; + const subscription = session.subscribeTrace({ afterSequence: latest }, (entry) => { + if ('kind' in entry && entry.kind === 'progress') listener({ params: entry.payload as ContractProgressNotification['params'] }); + }); + return () => subscription.unsubscribe(); + }, readResource: async (params, options) => ({ contents: [...(await session.readResource({ - signal: options?.signal === undefined ? signal : AbortSignal.any([signal, options.signal]), + signal: requestSignal(session, options), ...(options?.timeout === undefined ? {} : { timeoutMs: options.timeout }), uri: params.uri, })).contents] as never[], @@ -113,8 +157,8 @@ export const runDevEpochContracts = async ( contracts.diagnostics, ); } - const target = targetFor(prepared); const serverName = serverFor(prepared, contracts.server); + const target = devContractTarget(prepared, serverName); const manifest = testManifestFromRouteGraph({ apps: prepared.model?.mcpApps ?? [], configPath: prepared.configPath, @@ -145,13 +189,12 @@ export const runDevEpochContracts = async ( serverName, target, }); - const signal = AbortSignal.timeout(session.timeoutMs); await runDevEpochContractMatrix({ fixtures: contracts.fixtures, manifest, ...(contracts.server === undefined ? {} : { server: contracts.server }), session: { - client: matrixClient(session, signal), + client: matrixClient(session), provenance: { epochId, proofLevel: DEV_EPOCH_PROOF_LEVEL, diff --git a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts index c814ea028..b9c96ec71 100644 --- a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts +++ b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts @@ -30,9 +30,21 @@ export const subscribeToEpochAdoption = ( }, ); +/** A held epoch-store reference; releasing it lets retention reclaim the epoch. */ +export interface EpochAdoptionLease { + close(): Promise; +} + export interface EpochAdoptionPolicyOptions { readonly contracts: () => PreparedDevContractMatrix | undefined; readonly eventHub: ProjectEventHub; + /** + * Leases the adopted epoch for as long as hosts are told to serve it. Store + * retention keeps only the active epoch, referenced epochs, and a handful of + * recent unreferenced ones, so without this lease a run of failing rebuilds + * would delete the last passing epoch while it is still advertised. + */ + readonly lease?: (epochId: string) => Promise; readonly run: ( epochId: string, contracts: PreparedDevContractMatrix, @@ -40,7 +52,7 @@ export interface EpochAdoptionPolicyOptions { } interface PendingEpoch { - readonly contracts: PreparedDevContractMatrix; + readonly contracts: PreparedDevContractMatrix | undefined; readonly epochId: string; readonly sequence: number; } @@ -62,6 +74,21 @@ const failedEvaluation = (epochId: string, error: unknown): EpochContractEvaluat summary: 'Development contract matrix could not complete.', }); +const leaseFailedEvaluation = (epochId: string, error: unknown): EpochContractEvaluation => Object.freeze({ + diagnostics: Object.freeze([Object.freeze({ + code: 'AB7211', + message: `Epoch ${epochId} could not be leased for host adoption: ${ + error instanceof Error ? error.message : String(error) + }`, + recovery: 'Rebuild so a new epoch is published; the last leased host epoch remains active.', + severity: 'error', + } satisfies Diagnostic)]), + epochId, + failures: Object.freeze([]), + state: 'failed', + summary: 'Adopted epoch could not be leased.', +}); + /** * One host-facing epoch gate. Workbench playground surfaces continue to follow * artifact.available directly; only subscribers here wait for contract proof. @@ -69,11 +96,13 @@ const failedEvaluation = (epochId: string, error: unknown): EpochContractEvaluat export class EpochAdoptionPolicy implements EpochAdoptionSource { readonly #contracts: () => PreparedDevContractMatrix | undefined; readonly #eventHub: ProjectEventHub; + readonly #lease: EpochAdoptionPolicyOptions['lease']; readonly #listeners = new Set(); readonly #run: EpochAdoptionPolicyOptions['run']; readonly #subscription: ProjectEventSubscription; #closed = false; #currentEpochId: string | undefined; + #currentLease: EpochAdoptionLease | undefined; #latestEvaluation: EpochContractEvaluation | undefined; #observed = false; #pending: PendingEpoch | undefined; @@ -83,6 +112,7 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { constructor(options: EpochAdoptionPolicyOptions) { this.#contracts = options.contracts; this.#eventHub = options.eventHub; + this.#lease = options.lease; this.#run = options.run; this.#subscription = options.eventHub.subscribe( { afterSequence: options.eventHub.latestSequence }, @@ -119,20 +149,25 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { #consider(epochId: string): void { if (this.#closed) return; this.#observed = true; - const contracts = this.#contracts(); - if (contracts === undefined) { - this.#latestEvaluation = undefined; - this.#adopt(epochId); - return; - } this.#sequence += 1; this.#pending = Object.freeze({ - contracts, + contracts: this.#contracts(), epochId, sequence: this.#sequence, }); + this.#schedule(); + } + + /** + * Starts a drain unless one is running. A candidate that arrives after the + * running drain saw an empty queue but before its completion handler clears + * `#processing` would otherwise sit in `#pending` until the next rebuild, so + * the handler restarts the drain when it finds work left behind. + */ + #schedule(): void { this.#processing ??= this.#drain().finally(() => { this.#processing = undefined; + if (!this.#closed && this.#pending !== undefined) this.#schedule(); }); } @@ -144,8 +179,9 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { }); } - settled(): Promise { - return this.#processing ?? Promise.resolve(); + /** Resolves once no drain is running, including any drain restarted during a handoff. */ + async settled(): Promise { + while (this.#processing !== undefined) await this.#processing; } async close(): Promise { @@ -155,30 +191,81 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { this.#pending = undefined; await this.#processing; this.#listeners.clear(); + const lease = this.#currentLease; + this.#currentLease = undefined; + await lease?.close(); } + /** + * Evaluates, then leases, then publishes, then announces. Leasing before the + * status publish keeps "passed" synonymous with "adopted" for status readers, + * and announcing synchronously with the publish preserves the ordering hosts + * observed before leases existed. + */ async #drain(): Promise { while (!this.#closed && this.#pending !== undefined) { const candidate = this.#pending; this.#pending = undefined; - let evaluation: EpochContractEvaluation; - try { - evaluation = await this.#run(candidate.epochId, candidate.contracts); - } catch (error) { - evaluation = failedEvaluation(candidate.epochId, error); + let evaluation: EpochContractEvaluation | undefined; + if (candidate.contracts !== undefined) { + try { + evaluation = await this.#run(candidate.epochId, candidate.contracts); + } catch (error) { + evaluation = failedEvaluation(candidate.epochId, error); + } + if (this.#closed || candidate.sequence !== this.#sequence) continue; + } + let lease: EpochAdoptionLease | undefined; + if (evaluation === undefined || evaluation.state === 'passed') { + const leased = await this.#acquireLease(candidate); + if (leased === 'superseded') continue; + if (leased instanceof Error) evaluation = leaseFailedEvaluation(candidate.epochId, leased); + else lease = leased.lease; + // A newer epoch may have been queued between #acquireLease's own check + // and this resumption; an obsolete candidate is never published or adopted. + if (this.#closed || candidate.sequence !== this.#sequence) { + await lease?.close().catch(() => undefined); + continue; + } } - if (this.#closed || candidate.sequence !== this.#sequence) continue; this.#latestEvaluation = evaluation; - this.#eventHub.publish({ - epochId: candidate.epochId, - payload: evaluation, - type: 'dev.contract.status', - }); - if (evaluation.state === 'passed') this.#adopt(candidate.epochId); + if (evaluation !== undefined) { + this.#eventHub.publish({ + epochId: candidate.epochId, + payload: evaluation, + type: 'dev.contract.status', + }); + } + if (evaluation === undefined || evaluation.state === 'passed') await this.#adopt(candidate.epochId, lease); } } - #adopt(epochId: string): void { + async #acquireLease( + candidate: PendingEpoch, + ): Promise { + if (this.#lease === undefined) return { lease: undefined }; + let lease: EpochAdoptionLease; + try { + lease = await this.#lease(candidate.epochId); + } catch (error) { + if (this.#closed || candidate.sequence !== this.#sequence) return 'superseded'; + return error instanceof Error ? error : new Error(String(error)); + } + if (this.#closed || candidate.sequence !== this.#sequence) { + await lease.close().catch(() => undefined); + return 'superseded'; + } + return { lease }; + } + + /** + * Pins the candidate before announcing it and releases the previous pin only + * afterwards, so there is never a moment where the advertised epoch is + * unleased. + */ + async #adopt(epochId: string, lease: EpochAdoptionLease | undefined): Promise { + const previous = this.#currentLease; + this.#currentLease = lease; this.#currentEpochId = epochId; for (const listener of this.#listeners) { try { @@ -187,6 +274,7 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { // Adoption consumers own their async failure reporting; one cannot starve its peers. } } + await previous?.close().catch(() => undefined); } } diff --git a/packages/agent-bundle/src/dev/mcp-session/mcp-session-types.ts b/packages/agent-bundle/src/dev/mcp-session/mcp-session-types.ts index fb8901267..8cefa778d 100644 --- a/packages/agent-bundle/src/dev/mcp-session/mcp-session-types.ts +++ b/packages/agent-bundle/src/dev/mcp-session/mcp-session-types.ts @@ -27,9 +27,16 @@ export interface McpRequestOptions { readonly timeout: number; } +/** + * Request metadata carried on the wire as `params._meta`. `progressToken` is + * the MCP-defined key generated routes consult before sending progress; any + * other key travels untouched. + */ +export type McpRequestMeta = Readonly<{ readonly progressToken?: string | number } & Record>; + export interface McpClient { callTool( - params: { readonly arguments: Record; readonly name: string }, + params: { readonly _meta?: McpRequestMeta; readonly arguments: Record; readonly name: string }, options?: McpRequestOptions, ): Promise; close(): Promise; @@ -80,6 +87,8 @@ export interface McpSessionRequestOptions { } export interface McpSessionToolCallOptions extends McpSessionRequestOptions { + /** Forwarded verbatim as the request's `params._meta` (for example a `progressToken`). */ + readonly _meta?: McpRequestMeta; readonly arguments: Record; readonly name: string; /** A caller-chosen identifier used to cancel an in-flight tool call. */ diff --git a/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts b/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts index 53dcac693..5c3719f43 100644 --- a/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts +++ b/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts @@ -331,7 +331,11 @@ export class McpSession { this.#requests.set(requestId, controller); return Effect.gen({ self: this }, function* (this: McpSession) { const client = yield* liftTry(() => this.#clientFor()); - const result = yield* liftPromise(() => client.callTool({ arguments: options.arguments, name: options.name }, { + const result = yield* liftPromise(() => client.callTool({ + ...(options._meta === undefined ? {} : { _meta: options._meta }), + arguments: options.arguments, + name: options.name, + }, { signal: controller.signal, timeout: requestOptions(options, this.#timeoutMs).timeout, })); diff --git a/packages/agent-bundle/src/dev/playground/native-playground-service.ts b/packages/agent-bundle/src/dev/playground/native-playground-service.ts index ff48b256c..a1c01f424 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -97,6 +97,8 @@ export interface NativePlaygroundServiceOptions { readonly catalogDirectory?: string; /** @internal Fault-injection seam for durable epoch-sidecar publication. */ readonly catalogStorage?: NativePlaygroundCatalogStorage; + /** @internal How long a reader waits for a concurrent hard-link publisher before recovering an abandoned staging link. */ + readonly catalogStagingSettleDeadlineMs?: number; /** Test seams preserve the same production discovery and harness contracts. */ readonly discover?: (projectRoot: string) => Promise; readonly environment?: Readonly; @@ -204,6 +206,25 @@ const nativeHosts = new Set(NATIVE_HOSTS); const catalogDurabilityPlatformKey = Symbol.for('agent-bundle.native-playground-service.catalog-durability-platform'); const maximumCatalogSelections = 256; const maximumCatalogSnapshotBytes = 8 * 1_024 * 1_024; +/** How long a reader waits for a hard-link publisher to release its staging link before treating it as abandoned. */ +const stagingPublicationSettleDeadlineMs = 5_000; +const stagingPublicationPollMs = 10; + +/** + * Whether the publisher named by a `..stage--` entry is + * gone. The current process and any pid that still answers signal 0 (or that + * this user may not signal) count as alive; only a missing process is exited. + */ +const stagingPublisherExited = (stagingEntry: string): boolean => { + const pid = Number(/\.stage-(\d+)-/u.exec(stagingEntry)?.[1]); + if (!Number.isSafeInteger(pid) || pid <= 0 || pid === process.pid) return false; + try { + process.kill(pid, 0); + return false; + } catch (error) { + return isErrno(error, 'ESRCH'); + } +}; const maximumCatalogSnapshotNodes = 65_536; const maximumFixtureEntries = 4_096; const maximumSnapshotDepth = 16; @@ -598,6 +619,7 @@ export class NativePlaygroundService { readonly #planFixture: NonNullable; readonly #projectRoot: string; readonly #removeWorkspace: NonNullable; + readonly #stagingSettleDeadlineMs: number; #abortDispatchDepth = 0; #closePromise: Promise | undefined; #closed = false; @@ -605,6 +627,7 @@ export class NativePlaygroundService { constructor(options: NativePlaygroundServiceOptions) { this.#catalogDirectory = options.catalogDirectory; this.#catalogStorage = options.catalogStorage ?? Object.freeze({ link, mkdir, open, remove: rm }); + this.#stagingSettleDeadlineMs = options.catalogStagingSettleDeadlineMs ?? stagingPublicationSettleDeadlineMs; this.#catalogMove = options.catalogStorage?.move ?? rename; this.#projectRoot = options.projectRoot; this.#removeWorkspace = options.removeWorkspace ?? (async (root) => rm(root, { force: true, recursive: true })); @@ -1020,8 +1043,9 @@ export class NativePlaygroundService { if (!metadata.isFile() || metadata.nlink < 1 || metadata.size > maximumCatalogSnapshotBytes) { throw new Error('Native Playground catalog snapshot is invalid.'); } - if (!allowMultipleLinks && metadata.nlink !== 1 && !(await this.#stagingLinkAccountsFor(file, path, metadata))) { - throw new Error('Native Playground catalog snapshot is invalid.'); + if (!allowMultipleLinks && metadata.nlink !== 1) { + const settled = await this.#awaitStagedPublication(file, path, metadata); + if (settled === 'withdrawn') return undefined; } const buffer = Buffer.allocUnsafe(maximumCatalogSnapshotBytes + 1); let offset = 0; @@ -1048,27 +1072,133 @@ export class NativePlaygroundService { /** * Hard-link publication leaves a freshly linked sidecar doubly linked until - * the winner releases its staging file. A concurrent reader must adopt that - * winner rather than reject it, so the extra link is accounted for by - * identity: exactly one staging sibling of this epoch shares the sidecar's - * dev/ino, or the staging link was released while the directory was being - * listed and the still-open handle now reports a single link. Any other - * extra link is hostile aliasing and stays rejected. + * the winner releases its staging file, and that winner may still roll the + * sidecar back if its directory fsync or staging cleanup fails. A concurrent + * reader therefore treats a matching staging link as a publication in + * progress: it waits until the still-open handle reports a single link with + * the sidecar path still naming this inode (`settled`), or until the sidecar + * was withdrawn or replaced (`withdrawn`). The extra link is accounted for by + * identity — exactly one staging sibling of this epoch shares the sidecar's + * dev/ino; any other extra link is hostile aliasing and stays rejected. + * + * A publisher that dies between `link()` and its staging cleanup leaves the + * sidecar doubly linked forever. Once the deadline passes, a matching staging + * link whose publisher pid (embedded in its name) is no longer running is an + * abandoned publication: the sidecar was fsynced before it was linked, so the + * reader withdraws the orphaned staging link and adopts it. A staging link + * whose publisher is still alive keeps being rejected rather than yanked. */ - async #stagingLinkAccountsFor(file: FileHandle, path: string, metadata: Stats): Promise { - if (metadata.nlink !== 2) return false; + async #awaitStagedPublication(file: FileHandle, path: string, metadata: Stats): Promise<'settled' | 'withdrawn'> { + const invalid = () => new Error('Native Playground catalog snapshot is invalid.'); + const settledOrWithdrawn = async () => ((await this.#sidecarStillLinked(path, metadata)) ? 'settled' as const : 'withdrawn' as const); + if (metadata.nlink !== 2 || (await this.#stagingLinksFor(path, metadata)).length === 0) { + if ((await file.stat()).nlink === 1) return settledOrWithdrawn(); + throw invalid(); + } + const deadline = Date.now() + this.#stagingSettleDeadlineMs; + for (;;) { + await new Promise((resolvePromise) => { setTimeout(resolvePromise, stagingPublicationPollMs); }); + const current = await file.stat(); + if (current.nlink < 1) return 'withdrawn'; + if (current.nlink === 1) return settledOrWithdrawn(); + if (current.nlink !== 2) throw invalid(); + const staging = await this.#stagingLinksFor(path, metadata); + if (staging.length === 0) { + if ((await file.stat()).nlink === 1) return settledOrWithdrawn(); + throw invalid(); + } + if (!(await this.#sidecarStillLinked(path, metadata))) return 'withdrawn'; + if (Date.now() < deadline) continue; + if (!staging.every((entry) => stagingPublisherExited(entry))) throw invalid(); + for (const entry of staging) await this.#catalogStorage.remove(join(dirname(path), entry), { force: true }); + // The exited publisher may never have fsynced the directory after link(): + // flush it here so a crash cannot keep the orphan and lose the sidecar. + try { + await this.#syncCatalogDirectory(dirname(path)); + } catch (error) { + await this.#restoreStagingGuard(path, staging); + throw new Error('Native Playground catalog snapshot is invalid.', { cause: error }); + } + if ((await file.stat()).nlink === 1) return settledOrWithdrawn(); + throw invalid(); + } + } + + /** + * A recovery whose directory fsync failed has not made the orphan's removal + * crash-durable, so the sidecar must not read as settled: re-create the + * staging links it removed, and if even that fails, withdraw the sidecar. + */ + async #restoreStagingGuard(path: string, staging: readonly string[]): Promise { + const directory = dirname(path); + const failures: unknown[] = []; + // Each compensating step is only trusted once the directory is fsynced; + // an unsynced guard could vanish in a crash while the earlier unlink persists. + const durable = async (step: () => Promise): Promise => { + try { + await step(); + await this.#syncCatalogDirectory(directory); + return true; + } catch (error) { + failures.push(error); + return false; + } + }; + if (await durable(async () => { + const restored = await Promise.allSettled(staging.map((entry) => this.#restoreStagingLink(path, entry))); + const rejected = restored.find((outcome) => outcome.status === 'rejected'); + if (rejected !== undefined) throw rejected.reason; + })) return; + if (await durable(() => this.#catalogStorage.remove(path, { force: true }))) return; + // Last resort: a fresh guard under this process's pid keeps the sidecar + // doubly linked (and recoverable once this process exits) rather than + // leaving a singly linked file that the next reader would adopt. + const guard = join(directory, `.${basename(path, '.json')}.stage-${process.pid}-guard-${Math.random().toString(16).slice(2)}`); + if (await durable(() => this.#catalogStorage.link(path, guard))) return; + throw new AggregateError( + failures, + 'Native Playground catalog recovery could not keep the sidecar guarded.', + { cause: failures.at(-1) }, + ); + } + + /** Re-links one staging entry; a concurrent recoverer that already restored the same alias counts as success. */ + async #restoreStagingLink(path: string, entry: string): Promise { + const stagingPath = join(dirname(path), entry); + try { + await this.#catalogStorage.link(path, stagingPath); + } catch (error) { + if (!isErrno(error, 'EEXIST')) throw error; + const [existing, sidecar] = await Promise.all([lstat(stagingPath), lstat(path)]); + if (!existing.isFile() || !sameFile(existing, sidecar)) throw error; + } + } + + /** The epoch's own staging entries that alias this sidecar's inode. */ + async #stagingLinksFor(path: string, metadata: Stats): Promise { const directory = dirname(path); const stagingPrefix = `.${basename(path, '.json')}.stage-`; + const matches: string[] = []; for (const entry of await readdir(directory)) { if (!entry.startsWith(stagingPrefix)) continue; try { const staged = await lstat(join(directory, entry)); - if (staged.isFile() && sameFile(staged, metadata)) return true; + if (staged.isFile() && sameFile(staged, metadata)) matches.push(entry); } catch (error) { if (!isErrno(error, 'ENOENT')) throw error; } } - return (await file.stat()).nlink === 1; + return matches; + } + + async #sidecarStillLinked(path: string, metadata: Stats): Promise { + try { + const current = await lstat(path); + return current.isFile() && sameFile(current, metadata); + } catch (error) { + if (isErrno(error, 'ENOENT')) return false; + throw error; + } } async #persistSnapshot( @@ -1087,13 +1217,14 @@ export class NativePlaygroundService { let handle: Awaited> | undefined; let created = false; let publicationIdentity: string | undefined; + let staged: Stats | undefined; let primary: unknown; const cleanupFailures: unknown[] = []; try { handle = await this.#catalogStorage.open(temporary, 'wx', 0o600); await handle.writeFile(contents, 'utf8'); await handle.sync(); - const staged = await handle.stat(); + staged = await handle.stat(); if (!staged.isFile() || staged.nlink !== 1) { throw new Error('Native Playground catalog staging file is invalid.'); } @@ -1115,10 +1246,24 @@ export class NativePlaygroundService { try { await handle.close(); } catch (error) { cleanupFailures.push(error); } } - try { await this.#catalogStorage.remove(temporary, { force: true }); } - catch (error) { cleanupFailures.push(error); } + // A failed publication withdraws its sidecar while the staging link still + // exists: concurrent readers keep seeing an in-progress (doubly linked) + // publication until the path is gone, never a settled singly linked file + // that is about to be rolled back. If the rollback could not withdraw the + // sidecar, the staging link stays in place for the same reason. + let releaseStaging = true; + if (primary !== undefined && created && publicationIdentity !== undefined) { + try { await this.#publicationReceipt(path, publicationIdentity, true, true).rollback(); } + catch (error) { cleanupFailures.push(error); } + created = false; + releaseStaging = staged === undefined || !(await this.#sidecarStillLinked(path, staged)); + } + if (releaseStaging) { + try { await this.#catalogStorage.remove(temporary, { force: true }); } + catch (error) { cleanupFailures.push(error); } + } } - if ((primary !== undefined || cleanupFailures.length > 0) && created && publicationIdentity !== undefined) { + if (primary === undefined && cleanupFailures.length > 0 && created && publicationIdentity !== undefined) { try { await this.#publicationReceipt(path, publicationIdentity, true, true).rollback(); } catch (error) { cleanupFailures.push(error); } } diff --git a/packages/agent-bundle/src/dev/workbench-server.ts b/packages/agent-bundle/src/dev/workbench-server.ts index 75a3e731e..120ecf64a 100644 --- a/packages/agent-bundle/src/dev/workbench-server.ts +++ b/packages/agent-bundle/src/dev/workbench-server.ts @@ -734,6 +734,7 @@ export const startDevServer = async (options: StartDevServerOptions): Promise latestValidPreparedProject?.devContracts, eventHub, + lease: (epochId) => epochStore.acquireEpochReference(epochId), run: (epochId, contracts) => { const prepared = latestValidPreparedProject; if (prepared === undefined || prepared.devContracts !== contracts) { diff --git a/packages/agent-bundle/src/test/contract.ts b/packages/agent-bundle/src/test/contract.ts index 01803dd13..272901454 100644 --- a/packages/agent-bundle/src/test/contract.ts +++ b/packages/agent-bundle/src/test/contract.ts @@ -186,10 +186,24 @@ export interface ContractRouteFixture { */ export type ContractAppCoverage = 'auto' | 'explicit'; +/** The shape of one `notifications/progress` delivery a lifecycle fixture counts. */ +export interface ContractProgressNotification { + readonly params?: { readonly progressToken?: string | number }; +} + +/** + * The explicit progress path a non-SDK client exposes to lifecycle fixtures. + * Returns the unsubscribe; an SDK `Client` needs none because its handler map + * is observed directly. + */ +export interface ContractMatrixProgressSource { + readonly observeProgress: (listener: (notification: ContractProgressNotification) => void) => () => void; +} + export type ContractMatrixClient = Pick< Client, 'callTool' | 'getPrompt' | 'listPrompts' | 'listResources' | 'listTools' | 'readResource' ->; +> & Partial; export interface ContractMatrixRestartSession { readonly client: Client; @@ -1183,6 +1197,43 @@ type ClientNotificationHandler = ( ...arguments_: readonly unknown[] ) => void | Promise; +const progressMethod = 'notifications/progress'; + +const sdkProgressObserver = ( + notificationHandlers: Map, +): ContractMatrixProgressSource['observeProgress'] => (listener) => { + const callerHandler = notificationHandlers.get(progressMethod); + notificationHandlers.set(progressMethod, async (...arguments_) => { + listener(arguments_[0] as ContractProgressNotification); + await callerHandler?.(...arguments_); + }); + return () => { + if (callerHandler === undefined) { + notificationHandlers.delete(progressMethod); + } else { + notificationHandlers.set(progressMethod, callerHandler); + } + }; +}; + +/** + * Resolves how lifecycle fixtures observe live progress: an explicit + * `observeProgress` seam wins, an SDK `Client` exposes its handler map, and + * anything else cannot gate a lifecycle fixture and says so instead of + * failing on a missing private field. + */ +export const contractProgressObserver = ( + client: ContractMatrixClient, +): ContractMatrixProgressSource['observeProgress'] => { + const { observeProgress } = client; + if (typeof observeProgress === 'function') return (listener) => observeProgress.call(client, listener); + const handlers = (client as { readonly _notificationHandlers?: unknown })._notificationHandlers; + if (handlers instanceof Map) return sdkProgressObserver(handlers as Map); + throw new Error( + 'Contract matrix client exposes no progress notification path; lifecycle fixtures need an SDK Client or observeProgress.', + ); +}; + const executeLifecycleTransitions = async ( client: ContractMatrixClient, descriptor: TestableRouteDescriptor, @@ -1210,27 +1261,25 @@ const executeLifecycleTransitions = async ( } const byPhase = new Map(); - const progressMethod = 'notifications/progress'; - const notificationHandlers = ( - client as unknown as { - readonly _notificationHandlers: Map; - } - )._notificationHandlers; - const callerHandler = notificationHandlers.get(progressMethod); + let observeProgress: ContractMatrixProgressSource['observeProgress']; + try { + observeProgress = contractProgressObserver(client); + } catch (error) { + return { + byPhase: new Map(), + orderFailure: error instanceof Error ? error.message : captured(error), + }; + } let activeProgressToken: string | undefined; let settled = true; let liveProgress = 0; - notificationHandlers.set(progressMethod, async (...arguments_) => { - const notification = arguments_[0] as { - readonly params?: { readonly progressToken?: string | number }; - }; + const stopObserving = observeProgress((notification) => { if ( notification.params?.progressToken === activeProgressToken && !settled ) { liveProgress += 1; } - await callerHandler?.(...arguments_); }); try { for (const [index, transition] of transitions.entries()) { @@ -1248,11 +1297,7 @@ const executeLifecycleTransitions = async ( await runtimeIdentity.observe(`${descriptor.id}/${transition.phase}`); } } finally { - if (callerHandler === undefined) { - notificationHandlers.delete(progressMethod); - } else { - notificationHandlers.set(progressMethod, callerHandler); - } + stopObserving(); } return { byPhase }; }; diff --git a/packages/agent-bundle/src/test/index.ts b/packages/agent-bundle/src/test/index.ts index 3d5145f3f..7afe96766 100644 --- a/packages/agent-bundle/src/test/index.ts +++ b/packages/agent-bundle/src/test/index.ts @@ -114,9 +114,11 @@ export type { ContractMatrixClient, ContractMatrixOptions, ContractMatrixFailure, + ContractMatrixProgressSource, ContractMatrixProvenance, ContractMatrixReport, ContractMatrixRestartSession, + ContractProgressNotification, DevEpochContractMatrixOptions, DevEpochContractMatrixSession, DevEpochMcpProvenance, diff --git a/packages/agent-bundle/tests/dev-contract-runner.test.ts b/packages/agent-bundle/tests/dev-contract-runner.test.ts new file mode 100644 index 000000000..be96dc849 --- /dev/null +++ b/packages/agent-bundle/tests/dev-contract-runner.test.ts @@ -0,0 +1,168 @@ +import { expect, it } from '@rstest/core'; + +import { devContractTarget, matrixClient } from '../src/dev/dev-contract-runner.ts'; +import type { McpSessionTraceListener, McpSessionTraceMessage } from '../src/dev/mcp-session/mcp-session-protocol.ts'; +import { contractProgressObserver, type ContractMatrixClient } from '../src/test/contract.ts'; + +const model = (servers: readonly { readonly name: string; readonly targets: readonly string[] }[], targets: readonly string[]) => + ({ + model: { + mcpServers: servers.map((server) => ({ ...server, id: `mcp:${server.name}`, transport: 'stdio' })), + targets: targets.map((name) => ({ name })), + }, + }) as unknown as Parameters[0]; + +it('selects a target whose generated manifest carries the configured server, preferring portable when eligible', () => { + expect(devContractTarget(model([{ name: 'fixture', targets: ['claude', 'portable'] }], ['claude', 'portable']), 'fixture')) + .toBe('portable'); + // A server restricted to one host is absent from the portable manifest: the matrix must open it there instead. + expect(devContractTarget(model([{ name: 'fixture', targets: ['claude'] }], ['claude', 'portable']), 'fixture')) + .toBe('claude'); + expect(devContractTarget(model([{ name: 'fixture', targets: ['codex', 'claude'] }], ['codex', 'claude', 'portable']), 'fixture')) + .toBe('codex'); + // A server the model does not describe (handwritten registration) falls back to the project targets. + expect(devContractTarget(model([], ['claude', 'portable']), 'other')).toBe('portable'); +}); + +it('rejects a matrix whose server is emitted for none of the project targets', () => { + expect(() => devContractTarget(model([{ name: 'fixture', targets: ['cursor'] }], ['claude', 'portable']), 'fixture')) + .toThrow('Development contract matrix server "fixture" is emitted for none of the project\'s targets.'); + expect(() => devContractTarget(model([], []), 'fixture')) + .toThrow('Development contract matrix requires at least one generated target.'); +}); + +interface RecordedRequest { + readonly meta?: unknown; + readonly signal: AbortSignal | undefined; + readonly timeoutMs: number | undefined; +} + +const fakeSession = (timeoutMs: number) => { + const requests: RecordedRequest[] = []; + const listeners = new Set(); + let sequence = 0; + const record = (options: { readonly _meta?: unknown; readonly signal?: AbortSignal; readonly timeoutMs?: number } | undefined) => { + requests.push({ + ...(options?._meta === undefined ? {} : { meta: options._meta }), + signal: options?.signal, + timeoutMs: options?.timeoutMs, + }); + }; + const session = { + callTool: async (options: { readonly _meta?: unknown; readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { + record(options); + return { content: [] }; + }, + emitProgress: (progressToken: string) => { + sequence += 1; + const entry: McpSessionTraceMessage = { + kind: 'progress', + occurredAt: Date.now(), + payload: { progress: 1, progressToken }, + sequence, + }; + for (const listener of listeners) listener(entry); + }, + getPrompt: async (options: { readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { + record(options); + return { messages: [] }; + }, + listPrompts: async (options?: { readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { + record(options); + return []; + }, + listResources: async (options?: { readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { + record(options); + return []; + }, + listTools: async (options?: { readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { + record(options); + return []; + }, + readResource: async (options: { readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { + record(options); + return { contents: [] }; + }, + requests, + subscribeTrace: (_options: { readonly afterSequence?: number }, listener: McpSessionTraceListener) => { + listeners.add(listener); + return { unsubscribe: () => listeners.delete(listener) }; + }, + timeoutMs, + trace: () => ({ entries: [] }), + }; + return session; +}; + +it('applies the session timeout per matrix request instead of one deadline for the whole matrix', async () => { + const session = fakeSession(30_000); + const client = matrixClient(session as unknown as Parameters[0]); + + await client.listTools(); + await client.callTool({ arguments: {}, name: 'version' }); + await client.readResource({ uri: 'memo://x' }, { timeout: 250 }); + + expect(session.requests).toHaveLength(3); + const signals = session.requests.map((request) => request.signal); + expect(signals.every((signal) => signal instanceof AbortSignal && !signal.aborted)).toBe(true); + expect(new Set(signals).size).toBe(3); + expect(session.requests[2]?.timeoutMs).toBe(250); +}); + +it('forwards the lifecycle progress token as the request _meta so generated routes emit progress', async () => { + const session = fakeSession(30_000); + const client = matrixClient(session as unknown as Parameters[0]); + + await client.callTool({ _meta: { progressToken: 'lifecycle:route:0' }, arguments: {}, name: 'transition' }); + await client.callTool({ arguments: {}, name: 'plain' }); + + expect(session.requests[0]?.meta).toEqual({ progressToken: 'lifecycle:route:0' }); + expect(session.requests[1]).not.toHaveProperty('meta'); +}); + +it('exposes live progress notifications through the session trace for lifecycle fixtures', () => { + const session = fakeSession(30_000); + const client = matrixClient(session as unknown as Parameters[0]); + const observe = contractProgressObserver(client); + const seen: unknown[] = []; + const stop = observe((notification) => seen.push(notification.params?.progressToken)); + + session.emitProgress('token-1'); + stop(); + session.emitProgress('token-2'); + + expect(seen).toEqual(['token-1']); +}); + +it('invokes a custom observeProgress method with the client as its receiver', () => { + class InstanceClient { + readonly listeners = new Set<(notification: { readonly params?: { readonly progressToken?: string | number } }) => void>(); + observeProgress(listener: (notification: { readonly params?: { readonly progressToken?: string | number } }) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + } + const client = new InstanceClient(); + const seen: unknown[] = []; + const stop = contractProgressObserver(client as unknown as ContractMatrixClient)((notification) => + seen.push(notification.params?.progressToken)); + for (const listener of client.listeners) listener({ params: { progressToken: 'bound' } }); + stop(); + expect(seen).toEqual(['bound']); + expect(client.listeners.size).toBe(0); +}); + +it('names the missing notification path for a client that is neither an SDK Client nor exposes observeProgress', () => { + const bare = {} as ContractMatrixClient; + expect(() => contractProgressObserver(bare)) + .toThrow('Contract matrix client exposes no progress notification path'); + + const handlers = new Map void>(); + const sdkLike = { _notificationHandlers: handlers } as unknown as ContractMatrixClient; + const seen: unknown[] = []; + const stop = contractProgressObserver(sdkLike)((notification) => seen.push(notification.params?.progressToken)); + handlers.get('notifications/progress')?.({ params: { progressToken: 'sdk' } }); + stop(); + expect(handlers.has('notifications/progress')).toBe(false); + expect(seen).toEqual(['sdk']); +}); diff --git a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts index c3111ffc1..55166a31d 100644 --- a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts +++ b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts @@ -171,6 +171,181 @@ it('runs the contract matrix over a seeded epoch and reports it in the status sn await policy.close(); }); +it('leases the adopted epoch until another epoch replaces it or the policy closes', async () => { + const eventHub = new ProjectEventHub(); + const events: string[] = []; + const adopted: string[] = []; + const policy = new EpochAdoptionPolicy({ + contracts: () => ({ diagnostics: [], fixtures: {}, modulePath: '/project/fixtures.ts' }), + eventHub, + lease: async (epochId) => { + events.push(`lease:${epochId}`); + return { close: async () => { events.push(`release:${epochId}`); } }; + }, + run: async (epochId) => epochId === 'epoch-2' + ? Object.freeze({ + diagnostics: Object.freeze([]), + epochId, + failures: Object.freeze([]), + state: 'failed' as const, + summary: 'Development contract matrix reported 1 violation(s).', + }) + : passed(epochId), + }); + policy.subscribe((epochId) => { + adopted.push(epochId); + events.push(`adopt:${epochId}`); + }); + + publish(eventHub, 'epoch-1'); + await policy.settled(); + // A failing rebuild keeps the previous lease: retention cannot reclaim epoch-1. + publish(eventHub, 'epoch-2'); + await policy.settled(); + expect(events).toEqual(['lease:epoch-1', 'adopt:epoch-1']); + + publish(eventHub, 'epoch-3'); + await policy.settled(); + // The replacement is pinned before it is announced and the old pin is released only afterwards. + expect(events).toEqual(['lease:epoch-1', 'adopt:epoch-1', 'lease:epoch-3', 'adopt:epoch-3', 'release:epoch-1']); + expect(adopted).toEqual(['epoch-1', 'epoch-3']); + + await policy.close(); + expect(events.at(-1)).toBe('release:epoch-3'); +}); + +it('does not adopt an epoch it cannot lease and reports the failure as contract status', async () => { + const eventHub = new ProjectEventHub(); + const adopted: string[] = []; + const statuses: EpochContractEvaluation[] = []; + eventHub.subscribe((event) => { + if (event.type === 'dev.contract.status') statuses.push(event.payload); + }); + const policy = new EpochAdoptionPolicy({ + contracts: () => undefined, + eventHub, + lease: async (epochId) => { + if (epochId === 'epoch-gone') throw new Error('Epoch "epoch-gone" does not exist.'); + return { close: async () => undefined }; + }, + run: async () => { throw new Error('disabled contracts must not run'); }, + }); + policy.subscribe((epochId) => adopted.push(epochId)); + + publish(eventHub, 'epoch-1'); + await policy.settled(); + publish(eventHub, 'epoch-gone'); + await policy.settled(); + + expect(adopted).toEqual(['epoch-1']); + expect(policy.currentEpochId).toBe('epoch-1'); + expect(statuses).toEqual([expect.objectContaining({ + diagnostics: [expect.objectContaining({ code: 'AB7211', message: expect.stringContaining('could not be leased') })], + epochId: 'epoch-gone', + state: 'failed', + summary: 'Adopted epoch could not be leased.', + })]); + await policy.close(); +}); + +it('drains an epoch that arrives while the previous drain is finishing, at every microtask depth of the handoff', async () => { + // The drain's completion handler runs a few microtasks after its loop last + // saw an empty queue. Publish the next epoch from each depth inside that + // window (from the adoption listener) and require that it is still adopted. + for (let depth = 0; depth <= 8; depth += 1) { + const eventHub = new ProjectEventHub(); + const adopted: string[] = []; + const policy = new EpochAdoptionPolicy({ + contracts: () => undefined, + eventHub, + lease: async () => ({ close: async () => undefined }), + run: async () => { throw new Error('disabled contracts must not run'); }, + }); + policy.subscribe((epochId) => { + adopted.push(epochId); + if (epochId !== 'epoch-1') return; + let publishLate = (): void => publish(eventHub, 'epoch-2'); + for (let hop = 0; hop < depth; hop += 1) { + const next = publishLate; + publishLate = () => queueMicrotask(next); + } + publishLate(); + }); + + publish(eventHub, 'epoch-1'); + await policy.settled(); + // Let the deepest publish land (it may fall just after the first settle), then settle again. + await new Promise((resolvePromise) => { setTimeout(resolvePromise, 0); }); + await policy.settled(); + + expect(adopted, `microtask depth ${String(depth)}`).toEqual(['epoch-1', 'epoch-2']); + expect(policy.currentEpochId).toBe('epoch-2'); + await policy.close(); + } +}); + +it('never adopts a candidate superseded while its lease was being acquired, at every microtask depth', async () => { + // Epoch-1 passes; while its lease settles, epoch-2 is published from each + // microtask depth and then fails its contracts. Once epoch-2 has been + // published, epoch-1 is obsolete: it must not be adopted afterwards, so hosts + // never serve a stale epoch that a newer build already replaced. + for (let depth = 0; depth <= 8; depth += 1) { + const eventHub = new ProjectEventHub(); + const adopted: string[] = []; + const releases: string[] = []; + let epoch2Published = false; + const policy = new EpochAdoptionPolicy({ + contracts: () => ({ diagnostics: [], fixtures: {}, modulePath: '/project/fixtures.ts' }), + eventHub, + lease: async (epochId) => { + if (epochId === 'epoch-1') { + let publishLate = (): void => { + epoch2Published = true; + publish(eventHub, 'epoch-2'); + }; + for (let hop = 0; hop < depth; hop += 1) { + const next = publishLate; + publishLate = () => queueMicrotask(next); + } + publishLate(); + } + return { close: async () => { releases.push(epochId); } }; + }, + run: async (epochId) => epochId === 'epoch-1' + ? passed(epochId) + : Object.freeze({ + diagnostics: Object.freeze([]), + epochId, + failures: Object.freeze([]), + state: 'failed' as const, + summary: 'Development contract matrix reported 1 violation(s).', + }), + }); + const adoptedAfterSupersession: string[] = []; + policy.subscribe((epochId) => { + adopted.push(epochId); + if (epoch2Published) adoptedAfterSupersession.push(epochId); + }); + + publish(eventHub, 'epoch-1'); + await policy.settled(); + await new Promise((resolvePromise) => { setTimeout(resolvePromise, 0); }); + await policy.settled(); + + expect(adoptedAfterSupersession, `microtask depth ${String(depth)}`).toEqual([]); + if (adopted.length === 0) { + // Superseded before adoption: discarded, and its lease released. + expect(policy.currentEpochId, `microtask depth ${String(depth)}`).toBeUndefined(); + expect(releases, `microtask depth ${String(depth)}`).toEqual(['epoch-1']); + } else { + // Adopted before epoch-2 existed; the failing epoch-2 leaves it in place. + expect(adopted, `microtask depth ${String(depth)}`).toEqual(['epoch-1']); + expect(policy.currentEpochId, `microtask depth ${String(depth)}`).toBe('epoch-1'); + } + await policy.close(); + } +}); + it('discards a superseded contract result and evaluates only the latest pending epoch', async () => { const eventHub = new ProjectEventHub(); const first = Promise.withResolvers(); diff --git a/packages/agent-bundle/tests/mcp-session-service.test.ts b/packages/agent-bundle/tests/mcp-session-service.test.ts index e42944421..f9d526b48 100644 --- a/packages/agent-bundle/tests/mcp-session-service.test.ts +++ b/packages/agent-bundle/tests/mcp-session-service.test.ts @@ -285,6 +285,7 @@ it('keeps one generated server and plugin-data directory bound to the selected e it('uses the admitted session timeout for initialization, catalog, operations, and restart', async () => { const root = await mkdtemp(join(tmpdir(), 'agent-bundle-persistent-mcp-timeout-')); const observed: Array = []; + const callToolParams: unknown[] = []; const capture = (operation: string, options: { readonly timeout?: number } | undefined): void => { observed.push([operation, options?.timeout]); }; @@ -293,7 +294,8 @@ it('uses the admitted session timeout for initialization, catalog, operations, a const epochStore = await publishFixtureEpoch(root, 'epoch-timeout'); service = new McpSessionService({ createClient: () => ({ - callTool: async (_params, options) => { + callTool: async (params, options) => { + callToolParams.push(params); capture('callTool', options); return { content: [] }; }, @@ -347,6 +349,7 @@ it('uses the admitted session timeout for initialization, catalog, operations, a await session.getPrompt({ name: 'fixture' }); await session.readResource({ uri: 'ui://fixture/resource.txt' }); await session.callTool({ arguments: {}, name: 'fixture' }); + await session.callTool({ _meta: { progressToken: 'lifecycle:fixture:0' }, arguments: {}, name: 'fixture' }); await session.listTools({ timeoutMs: 321 }); await session.restart(); await expect(session.listTools({ timeoutMs: Number.NaN })).rejects.toThrow( @@ -364,9 +367,15 @@ it('uses the admitted session timeout for initialization, catalog, operations, a ['getPrompt', 12_345], ['readResource', 12_345], ['callTool', 12_345], + ['callTool', 12_345], ['listTools', 321], ['connect', 12_345], ]); + // `_meta` reaches the wire request only when the caller supplies it. + expect(callToolParams).toEqual([ + { arguments: {}, name: 'fixture' }, + { _meta: { progressToken: 'lifecycle:fixture:0' }, arguments: {}, name: 'fixture' }, + ]); await session.close(); } finally { await service?.close(); diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 3b5f0e690..c97d1a4c4 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1473,7 +1473,7 @@ it('fsyncs durable catalog publication, validates a no-replace winner, and retai } }); -it('adopts a linked winner while its staging link is still present instead of rejecting the doubly linked sidecar', async () => { +it('waits for a linked winner to release its staging link, then adopts it instead of rejecting the doubly linked sidecar', async () => { const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-linked-winner-')); const catalogDirectory = join(root, 'catalog'); const reference = epoch('epoch-linked-winner', join(root, 'artifact')); @@ -1513,13 +1513,18 @@ it('adopts a linked winner while its staging link is still present instead of re const winning = winner.catalog(reference); await linked; // The winner has linked its staging file into place but has not released - // it yet: the published sidecar is legitimately doubly linked here. + // it yet: the published sidecar is legitimately doubly linked here, and a + // reader must treat it as a publication still in progress. expect((await stat(sidecar)).nlink).toBe(2); - const losing = await loser.catalog(reference); + const losing = loser.catalog(reference); + await expect(Promise.race([ + losing.then(() => 'adopted' as const), + new Promise<'pending'>((resolvePromise) => { setTimeout(() => resolvePromise('pending'), 150); }), + ])).resolves.toBe('pending'); expect((await stat(sidecar)).nlink).toBe(2); releaseWinnerCleanup(); - expect(await winning).toEqual(losing); + expect(await winning).toEqual(await losing); expect((await stat(sidecar)).nlink).toBe(1); expect((await readdir(catalogDirectory)).filter((name) => name.includes('.stage-'))).toEqual([]); await Promise.all([winner.close(), loser.close()]); @@ -1529,6 +1534,357 @@ it('adopts a linked winner while its staging link is still present instead of re } }); +it('never adopts a staged sidecar that its publisher rolls back, and republishes its own instead', async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-withdrawn-winner-')); + const catalogDirectory = join(root, 'catalog'); + const reference = epoch('epoch-withdrawn-winner', join(root, 'artifact')); + const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); + const cleanupFailure = new Error('winner stage cleanup failed'); + let winnerStaging: string | undefined; + let signalLinked!: () => void; + const linked = new Promise((resolvePromise) => { signalLinked = resolvePromise; }); + let releaseWinnerCleanup!: () => void; + const winnerCleanup = new Promise((resolvePromise) => { releaseWinnerCleanup = resolvePromise; }); + const winnerStorage: NativePlaygroundCatalogStorage = { + link: async (source, destination) => { + await link(source, destination); + winnerStaging = String(source); + signalLinked(); + }, + mkdir, + open, + remove: async (path, options) => { + if (String(path) === winnerStaging) { + // The winner's staging cleanup stalls, then fails: the sidecar stays + // doubly linked the whole time and is rolled back afterwards. + await winnerCleanup; + throw cleanupFailure; + } + await rm(path, options); + }, + }; + const serviceFor = (storage?: NativePlaygroundCatalogStorage): NativePlaygroundService => new NativePlaygroundService({ + catalogDirectory, + ...(storage === undefined ? {} : { catalogStorage: storage }), + discover: async () => suite(), + inspectArtifact: async (candidate) => Object.freeze({ + binding: Object.freeze({ manifestPath: 'agent-bundle.manifest.json', source: 'explicit' as const, targetDigests: candidate.epoch.targetDigests }), + root: candidate.root, + }), + planFixture: async () => fixturePlan, + projectRoot: '/project', + }); + const winner = serviceFor(winnerStorage); + const loser = serviceFor(); + try { + const winning = winner.catalog(reference); + await linked; + expect((await stat(sidecar)).nlink).toBe(2); + const losing = loser.catalog(reference); + await expect(Promise.race([ + losing.then(() => 'adopted' as const), + new Promise<'pending'>((resolvePromise) => { setTimeout(() => resolvePromise('pending'), 150); }), + ])).resolves.toBe('pending'); + releaseWinnerCleanup(); + await expect(winning).rejects.toMatchObject({ errors: [cleanupFailure] }); + // The reader saw the publication withdrawn rather than adopting the rolled + // back inode, so it discovered and persisted a singly linked sidecar itself. + const adopted = await losing; + expect((await stat(sidecar)).nlink).toBe(1); + expect(await loser.catalog(reference)).toEqual(adopted); + await Promise.all([winner.close(), loser.close()]); + } finally { + releaseWinnerCleanup(); + await rm(root, { force: true, recursive: true }); + } +}); + +it('withdraws a sidecar whose directory fsync fails before releasing its staging link, so a waiting reader never adopts it', async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-fsync-withdrawn-')); + const catalogDirectory = join(root, 'catalog'); + const reference = epoch('epoch-fsync-withdrawn', join(root, 'artifact')); + const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); + const directoryFailure = new Error('directory sync failed'); + let signalLinked!: () => void; + const linked = new Promise((resolvePromise) => { signalLinked = resolvePromise; }); + let releaseDirectorySync!: () => void; + const directorySync = new Promise((resolvePromise) => { releaseDirectorySync = resolvePromise; }); + const winnerStorage: NativePlaygroundCatalogStorage = { + link: async (source, destination) => { + await link(source, destination); + signalLinked(); + }, + mkdir, + open: async (path, flags, mode) => { + const handle = await open(path, flags, mode); + return new Proxy(handle, { + get(target, property) { + if (property === 'sync' && String(path) === catalogDirectory) { + return async () => { + // The publisher stalls on its directory fsync after the link, then fails it. + await directorySync; + throw directoryFailure; + }; + } + const value = Reflect.get(target, property, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + }, + remove: rm, + } as NativePlaygroundCatalogStorage; + const serviceFor = (storage?: NativePlaygroundCatalogStorage): NativePlaygroundService => new NativePlaygroundService({ + catalogDirectory, + ...(storage === undefined ? {} : { catalogStorage: storage }), + discover: async () => suite(), + inspectArtifact: async (candidate) => Object.freeze({ + binding: Object.freeze({ manifestPath: 'agent-bundle.manifest.json', source: 'explicit' as const, targetDigests: candidate.epoch.targetDigests }), + root: candidate.root, + }), + planFixture: async () => fixturePlan, + projectRoot: '/project', + }); + const winner = serviceFor(winnerStorage); + const loser = serviceFor(); + try { + const winning = winner.catalog(reference); + await linked; + expect((await stat(sidecar)).nlink).toBe(2); + const losing = loser.catalog(reference); + await expect(Promise.race([ + losing.then(() => 'adopted' as const), + new Promise<'pending'>((resolvePromise) => { setTimeout(() => resolvePromise('pending'), 150); }), + ])).resolves.toBe('pending'); + releaseDirectorySync(); + // The rollback's own directory fsync fails the same way, so both are retained. + await expect(winning).rejects.toMatchObject({ errors: [directoryFailure, directoryFailure] }); + const adopted = await losing; + expect((await stat(sidecar)).nlink).toBe(1); + expect((await readdir(catalogDirectory)).filter((name) => name.includes('.stage-'))).toEqual([]); + expect(await loser.catalog(reference)).toEqual(adopted); + await Promise.all([winner.close(), loser.close()]); + } finally { + releaseDirectorySync(); + await rm(root, { force: true, recursive: true }); + } +}); + +it('keeps the staging link when a failed publication cannot roll its sidecar back, so readers never see it as settled', async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-rollback-failed-')); + const catalogDirectory = join(root, 'catalog'); + const reference = epoch('epoch-rollback-failed', join(root, 'artifact')); + const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); + const directoryFailure = new Error('directory sync failed'); + const moveFailure = Object.assign(new Error('rename busy'), { code: 'EBUSY' }); + const storage: NativePlaygroundCatalogStorage = { + link, + mkdir, + move: async () => { throw moveFailure; }, + open: async (path, flags, mode) => { + const handle = await open(path, flags, mode); + return new Proxy(handle, { + get(target, property) { + if (property === 'sync' && String(path) === catalogDirectory) { + return async () => { throw directoryFailure; }; + } + const value = Reflect.get(target, property, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + }, + remove: rm, + } as NativePlaygroundCatalogStorage; + const serviceFor = ( + discover: () => Promise, + catalogStorage?: NativePlaygroundCatalogStorage, + ): NativePlaygroundService => new NativePlaygroundService({ + catalogDirectory, + catalogStagingSettleDeadlineMs: 50, + ...(catalogStorage === undefined ? {} : { catalogStorage }), + discover, + inspectArtifact: async (candidate) => Object.freeze({ + binding: Object.freeze({ manifestPath: 'agent-bundle.manifest.json', source: 'explicit' as const, targetDigests: candidate.epoch.targetDigests }), + root: candidate.root, + }), + planFixture: async () => fixturePlan, + projectRoot: '/project', + }); + try { + const winner = serviceFor(async () => suite(), storage); + await expect(winner.catalog(reference)).rejects.toMatchObject({ + errors: [directoryFailure, moveFailure], + }); + await winner.close(); + // The sidecar survived its failed rollback, so its staging link survives with + // it: the publication still reads as in progress, never as settled. + expect((await stat(sidecar)).nlink).toBe(2); + expect((await readdir(catalogDirectory)).filter((name) => name.includes('.stage-'))).toHaveLength(1); + const reader = serviceFor(async () => { throw new Error('An unsettled catalog must not fall back to discovery.'); }); + await expect(reader.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); + await reader.close(); + } finally { + await rm(root, { force: true, recursive: true }); + } +}); + +const exitedPid = (): number => { + for (let pid = 4_194_000; pid > 1_000; pid -= 1) { + try { + process.kill(pid, 0); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ESRCH') return pid; + } + } + throw new Error('No exited pid was available for the abandoned-staging fixture.'); +}; + +it('recovers a staging link abandoned by an exited publisher after the settle deadline, but never one whose publisher is alive', async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-abandoned-staging-')); + const catalogDirectory = join(root, 'catalog'); + const reference = epoch('epoch-abandoned-staging', join(root, 'artifact')); + const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); + const directorySyncs: string[] = []; + // Fault injection for the recovery fallback chain. Directory fsync failures + // are counted down so that the recovery fsync fails while the fsync that + // makes the compensating guard durable succeeds. + let directorySyncFailuresLeft = 0; + const directorySyncFailure = Object.assign(new Error('directory sync failed'), { code: 'EIO' }); + let relinkFailure: Error | undefined; + let concurrentRestorer = false; + let sidecarRemoveFailure: Error | undefined; + const storage: NativePlaygroundCatalogStorage = { + link: async (source, destination) => { + if (String(destination).endsWith('-orphan')) { + if (relinkFailure !== undefined) throw relinkFailure; + if (concurrentRestorer) { + // Another reader restored the same alias a moment earlier. + await link(source, destination); + throw Object.assign(new Error('link exists'), { code: 'EEXIST' }); + } + } + await link(source, destination); + }, + mkdir, + open: async (path, flags, mode) => { + const handle = await open(path, flags, mode); + return new Proxy(handle, { + get(target, property) { + if (property === 'sync' && String(path) === catalogDirectory) { + return async () => { + directorySyncs.push(String(path)); + if (directorySyncFailuresLeft > 0) { + directorySyncFailuresLeft -= 1; + throw directorySyncFailure; + } + await target.sync(); + }; + } + const value = Reflect.get(target, property, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + }, + remove: async (path, options) => { + if (sidecarRemoveFailure !== undefined && String(path) === sidecar) throw sidecarRemoveFailure; + await rm(path, options); + }, + } as NativePlaygroundCatalogStorage; + const serviceFor = (discover: () => Promise): NativePlaygroundService => new NativePlaygroundService({ + catalogDirectory, + catalogStagingSettleDeadlineMs: 50, + catalogStorage: storage, + discover, + inspectArtifact: async (candidate) => Object.freeze({ + binding: Object.freeze({ manifestPath: 'agent-bundle.manifest.json', source: 'explicit' as const, targetDigests: candidate.epoch.targetDigests }), + root: candidate.root, + }), + planFixture: async () => fixturePlan, + projectRoot: '/project', + }); + const rejectsUnsettled = async (label: string): Promise => { + const reader = serviceFor(async () => { throw new Error(`${label} must not fall back to discovery.`); }); + await expect(reader.catalog(reference)).rejects.toThrow(); + await reader.close(); + }; + try { + const writer = serviceFor(async () => suite()); + const published = await writer.catalog(reference); + await writer.close(); + + // A publisher that is still running owns its staging link: the reader waits out the deadline and rejects. + const live = join(catalogDirectory, `.${reference.epoch.id}.stage-${String(process.pid)}-live`); + await link(sidecar, live); + await rejectsUnsettled('A blocked catalog'); + expect((await stat(sidecar)).nlink).toBe(2); + await rm(live); + + // A recovery whose directory fsync fails has not made the orphan's removal + // durable: the staging guard is restored, fsynced, and the sidecar stays unsettled. + const orphan = join(catalogDirectory, `.${reference.epoch.id}.stage-${String(exitedPid())}-orphan`); + await link(sidecar, orphan); + directorySyncFailuresLeft = 1; + directorySyncs.length = 0; + await rejectsUnsettled('An unsynced recovery'); + expect((await stat(sidecar)).nlink).toBe(2); + expect((await stat(orphan)).ino).toBe((await stat(sidecar)).ino); + expect(directorySyncs).toEqual([catalogDirectory, catalogDirectory]); + + // A concurrent recoverer restored the same alias first: EEXIST on an entry + // that already aliases the sidecar counts as a restored guard. + directorySyncFailuresLeft = 1; + concurrentRestorer = true; + await rejectsUnsettled('A racing recovery'); + concurrentRestorer = false; + expect((await stat(sidecar)).nlink).toBe(2); + expect((await stat(orphan)).ino).toBe((await stat(sidecar)).ino); + + // When the guard cannot be re-linked and the sidecar cannot be unlinked + // either, a fresh fsynced guard under this process's pid keeps it doubly linked. + directorySyncFailuresLeft = 1; + relinkFailure = Object.assign(new Error('relink denied'), { code: 'EPERM' }); + sidecarRemoveFailure = Object.assign(new Error('sidecar busy'), { code: 'EBUSY' }); + await rejectsUnsettled('A guarded recovery'); + expect((await stat(sidecar)).nlink).toBe(2); + const guards = (await readdir(catalogDirectory)).filter((name) => name.includes(`.stage-${String(process.pid)}-guard-`)); + expect(guards).toHaveLength(1); + expect((await stat(join(catalogDirectory, guards[0]!))).ino).toBe((await stat(sidecar)).ino); + // A live-pid guard is honoured by the next reader until this process exits. + await rejectsUnsettled('A guarded catalog'); + await rm(join(catalogDirectory, guards[0]!)); + await link(sidecar, orphan); + relinkFailure = undefined; + sidecarRemoveFailure = undefined; + + // When every fsync keeps failing, the sidecar is withdrawn rather than left + // singly linked: the guard chain never trusts an unsynced step. + directorySyncFailuresLeft = Number.POSITIVE_INFINITY; + await rejectsUnsettled('A withdrawn recovery'); + directorySyncFailuresLeft = 0; + await expect(stat(sidecar)).rejects.toMatchObject({ code: 'ENOENT' }); + const republisher = serviceFor(async () => suite()); + expect(await republisher.catalog(reference)).toEqual(published); + await republisher.close(); + // The restored orphan name still aliases the withdrawn inode; a fresh + // sidecar is singly linked and unaffected by it. + expect((await stat(sidecar)).nlink).toBe(1); + await rm(orphan); + await link(sidecar, orphan); + + // A publisher that exited after link() but before cleanup left a complete, + // fsynced sidecar behind: the reader withdraws the orphan and adopts it. + directorySyncs.length = 0; + const reader = serviceFor(async () => { throw new Error('A recovered catalog must not fall back to discovery.'); }); + expect(await reader.catalog(reference)).toEqual(published); + await reader.close(); + expect((await stat(sidecar)).nlink).toBe(1); + expect((await readdir(catalogDirectory)).filter((name) => name.includes('.stage-'))).toEqual([]); + // The exited publisher may never have flushed the directory after link(); recovery does. + expect(directorySyncs).toEqual([catalogDirectory]); + } finally { + await rm(root, { force: true, recursive: true }); + } +}); + it('still rejects a persisted catalog aliased by a hard link that is not an epoch staging file', async () => { const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-aliased-catalog-')); const catalogDirectory = join(root, 'catalog'); diff --git a/packages/agent-bundle/tests/support/host-install.ts b/packages/agent-bundle/tests/support/host-install.ts index be7f976ec..0a432f18d 100644 --- a/packages/agent-bundle/tests/support/host-install.ts +++ b/packages/agent-bundle/tests/support/host-install.ts @@ -1,6 +1,6 @@ import { execFile as executeFile } from 'node:child_process'; import { access, cp, lstat, mkdir, mkdtemp, readFile, readdir, realpath, rm, symlink, writeFile } from 'node:fs/promises'; -import { tmpdir } from 'node:os'; +import { homedir, tmpdir } from 'node:os'; import { delimiter, dirname, isAbsolute, join, relative, resolve, sep } from 'node:path'; import { promisify } from 'node:util'; import { randomUUID } from 'node:crypto'; @@ -435,12 +435,18 @@ const record = (value: unknown): Readonly> | undefined = const normalizedRelative = (root: string, path: string): string => relative(root, path).split(sep).join('/'); +/** + * Isolates a scenario's home. `os.homedir()` follows `HOME` on POSIX but + * `USERPROFILE` on Windows, so an isolated `HOME` alone would leave a Windows + * developer's real `~/.cursor/plugins/local` as the install destination. + */ const isolatedEnvironment = ( environment: Readonly, values: Readonly, ): NodeJS.ProcessEnv => ({ ...packedNativeEnvironment(environment), ...values, + ...(values.HOME === undefined ? {} : { USERPROFILE: values.HOME }), }); const stringEnvironment = ( @@ -1701,6 +1707,10 @@ export const runClaudeLiveDevSessionProof = async ( options: { readonly environment: Readonly }, ): Promise => { const sessionId = randomUUID(); + // Captured before runLiveHostScenario swaps process.env.HOME for the isolated + // scenario home: the spawned claude turn runs against the developer's real + // home, so the unchanged-state guard must digest that same home. + const normalHome = homedir(); const normalEnvironment = { ...packedNativeEnvironment(options.environment) }; delete normalEnvironment.CLAUDE_CONFIG_DIR; const toolOutputs: string[] = []; @@ -1736,6 +1746,7 @@ export const runClaudeLiveDevSessionProof = async ( timeout: 300_000, }); }, + { homeDirectory: normalHome }, ); assertProof(result !== undefined, `Claude ${liveVersion} inline live development turn did not run.`); const rawOutput = result.stdout.trim();