From 7e34582063ca1e15f2657c4af4a9d471b0e2c20d Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 07:09:01 +0000 Subject: [PATCH 01/16] fix: address late review threads on merged PRs #368/#373/#374/#377/#378/#379/#385/#390 - dev: lease the adopted epoch in EpochAdoptionPolicy until replaced or closed; select the contract-matrix target from the server's own target list; apply the session timeout per matrix request; observe lifecycle progress through the session trace via a new ContractMatrixClient.observeProgress seam (#385) - playground: wait for a hard-link catalog publisher to release its staging link before adopting the sidecar; return to discovery when the publication is rolled back (#377) - build: run the Agent Plugins byte lane over portable/ during ordinary artifact validation; reject every forbidden control character in header values (#373) - events/hooks: Codex PostToolUse accepts any present JSON tool_response (#378) - api: project only contract fields of adapter capability rows in inspect (#390) - tests/support: digest the real Claude home in the live session guard; isolate USERPROFILE alongside HOME (#374) - docs: Claude plugin-root cwd exception, parked-pin trigger independence, provider typing contract, portable validation moments (#368/#379/#382/#373) --- .changeset/late-review-thread-fixes.md | 31 ++++ docs/effect-conventions.md | 14 +- packages/agent-bundle/README.md | 3 +- .../src/dev/dev-contract-runner.ts | 68 +++++++-- .../src/dev/epoch-adoption-policy.ts | 84 +++++++++-- .../playground/native-playground-service.ts | 56 ++++++-- .../agent-bundle/src/dev/workbench-server.ts | 1 + packages/agent-bundle/src/test/contract.ts | 80 ++++++++--- packages/agent-bundle/src/test/index.ts | 2 + packages/agent-bundle/tests/api.test.ts | 38 +++++ .../tests/dev-contract-runner.test.ts | 134 ++++++++++++++++++ .../tests/epoch-adoption-policy.test.ts | 77 ++++++++++ .../tests/native-playground-service.test.ts | 78 +++++++++- .../tests/support/host-install.ts | 13 +- 14 files changed, 612 insertions(+), 67 deletions(-) create mode 100644 .changeset/late-review-thread-fixes.md create mode 100644 packages/agent-bundle/tests/dev-contract-runner.test.ts diff --git a/.changeset/late-review-thread-fixes.md b/.changeset/late-review-thread-fixes.md new file mode 100644 index 000000000..5c336abe9 --- /dev/null +++ b/.changeset/late-review-thread-fixes.md @@ -0,0 +1,31 @@ +--- +"agent-bundle": patch +--- + +Address the post-merge review findings on the dev epoch gate, native catalog, +portable validation, Codex hooks, and inspection: + +- `EpochAdoptionPolicy` leases the adopted epoch until another epoch replaces + it or the policy closes, so store retention cannot delete the advertised + last-good build during a run of failing rebuilds; an epoch that cannot be + leased is not adopted and the failure is published as `AB7211` status. +- The dev contract matrix opens the configured server on a target whose + manifest actually carries it, applies the session timeout per request instead + of once for the whole matrix, and observes live progress through the session + trace; lifecycle fixtures no longer depend on the SDK client's private + `_notificationHandlers` map (`ContractMatrixClient` gains an optional + `observeProgress` seam, exported as `ContractMatrixProgressSource`). +- A Native Playground catalog reader now waits for a hard-link publisher to + release its staging link before adopting the sidecar, and returns to + discovery when that publication is rolled back instead of caching a withdrawn + epoch. +- Ordinary `build`/`validate --artifact` runs the Agent Plugins byte lane over + the emitted `portable/` tree (`AB6035`–`AB6037`), so standard-invalid + documents fail before publication rather than only under `--host-validation`; + header values reject every forbidden control character, not just CR/LF/NUL. +- Codex `PostToolUse` accepts any present JSON `tool_response`, matching the + pinned `"tool_response": true` schema, in both the event projection and the + generated native hook wrapper; Claude keeps the object check. +- `inspect` projects only the contract fields of an adapter capability row, so + extension fields on JavaScript adapters cannot shadow the capability name or + break `--json` serialization. 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..55dca94b5 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,61 @@ 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({ 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 +154,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 +186,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..71f8593f7 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,15 +149,9 @@ 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, }); @@ -155,12 +179,20 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { this.#pending = undefined; await this.#processing; this.#listeners.clear(); + const lease = this.#currentLease; + this.#currentLease = undefined; + await lease?.close(); } async #drain(): Promise { while (!this.#closed && this.#pending !== undefined) { const candidate = this.#pending; this.#pending = undefined; + if (candidate.contracts === undefined) { + this.#latestEvaluation = undefined; + await this.#adopt(candidate); + continue; + } let evaluation: EpochContractEvaluation; try { evaluation = await this.#run(candidate.epochId, candidate.contracts); @@ -174,19 +206,47 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { payload: evaluation, type: 'dev.contract.status', }); - if (evaluation.state === 'passed') this.#adopt(candidate.epochId); + if (evaluation.state === 'passed') await this.#adopt(candidate); } } - #adopt(epochId: string): void { - this.#currentEpochId = epochId; + /** + * 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. A candidate that cannot be leased is not adopted: hosts keep the + * previous epoch and the failure is published as contract status. + */ + async #adopt(candidate: PendingEpoch): Promise { + let lease: EpochAdoptionLease | undefined; + if (this.#lease !== undefined) { + try { + lease = await this.#lease(candidate.epochId); + } catch (error) { + if (this.#closed || candidate.sequence !== this.#sequence) return; + this.#latestEvaluation = leaseFailedEvaluation(candidate.epochId, error); + this.#eventHub.publish({ + epochId: candidate.epochId, + payload: this.#latestEvaluation, + type: 'dev.contract.status', + }); + return; + } + if (this.#closed || candidate.sequence !== this.#sequence) { + await lease.close().catch(() => undefined); + return; + } + } + const previous = this.#currentLease; + this.#currentLease = lease; + this.#currentEpochId = candidate.epochId; for (const listener of this.#listeners) { try { - listener(epochId); + listener(candidate.epochId); } catch { // 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/playground/native-playground-service.ts b/packages/agent-bundle/src/dev/playground/native-playground-service.ts index ff48b256c..c2712d023 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -204,6 +204,9 @@ 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 rejecting the sidecar. */ +const stagingPublicationSettleDeadlineMs = 5_000; +const stagingPublicationPollMs = 10; const maximumCatalogSnapshotNodes = 65_536; const maximumFixtureEntries = 4_096; const maximumSnapshotDepth = 16; @@ -1020,8 +1023,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,15 +1052,35 @@ 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, as + * does a staging link that never settles within the deadline. */ - 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.'); + if (metadata.nlink !== 2 || !(await this.#stagingLinkAccountsFor(path, metadata))) { + if ((await file.stat()).nlink === 1) return (await this.#sidecarStillLinked(path, metadata)) ? 'settled' : 'withdrawn'; + throw invalid(); + } + const deadline = Date.now() + stagingPublicationSettleDeadlineMs; + for (;;) { + await new Promise((resolvePromise) => { setTimeout(resolvePromise, stagingPublicationPollMs); }); + const current = await file.stat(); + if (current.nlink < 1) return 'withdrawn'; + if (current.nlink === 1) return (await this.#sidecarStillLinked(path, metadata)) ? 'settled' : 'withdrawn'; + if (current.nlink !== 2 || !(await this.#stagingLinkAccountsFor(path, metadata))) throw invalid(); + if (!(await this.#sidecarStillLinked(path, metadata))) return 'withdrawn'; + if (Date.now() >= deadline) throw invalid(); + } + } + + async #stagingLinkAccountsFor(path: string, metadata: Stats): Promise { const directory = dirname(path); const stagingPrefix = `.${basename(path, '.json')}.stage-`; for (const entry of await readdir(directory)) { @@ -1068,7 +1092,17 @@ export class NativePlaygroundService { if (!isErrno(error, 'ENOENT')) throw error; } } - return (await file.stat()).nlink === 1; + return false; + } + + 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( 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..07ab43b11 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,42 @@ 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'] => { + if (typeof client.observeProgress === 'function') return client.observeProgress; + 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 +1260,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 +1296,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/api.test.ts b/packages/agent-bundle/tests/api.test.ts index 0fdb49463..109c0f23b 100644 --- a/packages/agent-bundle/tests/api.test.ts +++ b/packages/agent-bundle/tests/api.test.ts @@ -340,6 +340,44 @@ it('accepts the public claude.dependencies config surface and plans its manifest } }); +it('fails an ordinary portable build on Agent Plugins normative-text violations without --host-validation', async () => { + const root = await createProject(); + try { + await writeFile(join(root, 'agent-bundle.config.ts'), [ + 'export default {', + " plugin: { name: 'portable-standard', version: '1.0.0' },", + " targets: ['portable'],", + ' mcp: { servers: { remote: {', + " headers: { 'X-Tenant': 'a', 'x-tenant': 'b' },", + " transport: 'streamable-http',", + " url: 'http://mcp.example.test/mcp',", + ' } } },', + '};', + '', + ].join('\n')); + + // The pinned schemas accept this document; only the standard's text forbids it. + await expect(build({ output: join(root, 'artifact-out'), root })).rejects.toMatchObject({ + diagnostics: expect.arrayContaining([ + expect.objectContaining({ + code: 'AB6036', + message: expect.stringContaining('repeats header "X-Tenant" under different casing'), + severity: 'error', + target: 'portable', + }), + expect.objectContaining({ + code: 'AB6036', + message: expect.stringContaining('uses plain HTTP against non-loopback host "mcp.example.test"'), + severity: 'error', + target: 'portable', + }), + ]), + }); + } finally { + await rm(join(root, '..'), { force: true, recursive: true }); + } +}); + it('reports one modern-MCP source diagnostic for a legacy SSE declaration', async () => { const root = await createProject(); try { 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..47d48fd7e --- /dev/null +++ b/packages/agent-bundle/tests/dev-contract-runner.test.ts @@ -0,0 +1,134 @@ +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 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 signal?: AbortSignal; readonly timeoutMs?: number } | undefined) => { + requests.push({ signal: options?.signal, timeoutMs: options?.timeoutMs }); + }; + const session = { + callTool: async (options: { 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('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('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..ff612f47b 100644 --- a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts +++ b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts @@ -171,6 +171,83 @@ 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('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/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 3b5f0e690..a9d0983ac 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,71 @@ 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('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(); From 3298444caf078e06d58f67c441b8aa04f36964ef Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 07:46:20 +0000 Subject: [PATCH 02/16] fix(dev): lease before publishing contract status; align fixtures with the portable byte lane - EpochAdoptionPolicy acquires the epoch lease before publishing a passed dev.contract.status and announces adoption synchronously with it, so a status reader never sees "passed" for an epoch that is not yet adopted - hooks.test: Codex PostToolUse accepts a string tool_response and rejects a missing one; Claude keeps the object check - mcp-session-service/public-api-packed fixtures: Agent Plugins forbids placeholders in headers and non-bare/non-./ commands, and ordinary artifact validation now enforces the standard, so the fixtures carry a literal header and a bare `node` command --- .../src/dev/epoch-adoption-policy.ts | 90 ++++++++++--------- 1 file changed, 50 insertions(+), 40 deletions(-) diff --git a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts index 71f8593f7..12c3fdf3c 100644 --- a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts +++ b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts @@ -184,64 +184,74 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { 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; - if (candidate.contracts === undefined) { - this.#latestEvaluation = undefined; - await this.#adopt(candidate); - continue; + 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 evaluation: EpochContractEvaluation; - try { - evaluation = await this.#run(candidate.epochId, candidate.contracts); - } catch (error) { - evaluation = failedEvaluation(candidate.epochId, error); + 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; } - 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') await this.#adopt(candidate); + 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); + } + } + + 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. A candidate that cannot be leased is not adopted: hosts keep the - * previous epoch and the failure is published as contract status. + * unleased. */ - async #adopt(candidate: PendingEpoch): Promise { - let lease: EpochAdoptionLease | undefined; - if (this.#lease !== undefined) { - try { - lease = await this.#lease(candidate.epochId); - } catch (error) { - if (this.#closed || candidate.sequence !== this.#sequence) return; - this.#latestEvaluation = leaseFailedEvaluation(candidate.epochId, error); - this.#eventHub.publish({ - epochId: candidate.epochId, - payload: this.#latestEvaluation, - type: 'dev.contract.status', - }); - return; - } - if (this.#closed || candidate.sequence !== this.#sequence) { - await lease.close().catch(() => undefined); - return; - } - } + async #adopt(epochId: string, lease: EpochAdoptionLease | undefined): Promise { const previous = this.#currentLease; this.#currentLease = lease; - this.#currentEpochId = candidate.epochId; + this.#currentEpochId = epochId; for (const listener of this.#listeners) { try { - listener(candidate.epochId); + listener(epochId); } catch { // Adoption consumers own their async failure reporting; one cannot starve its peers. } From 2976cccbeb06be4c357b21229442b1b348ec87e1 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 07:47:55 +0000 Subject: [PATCH 03/16] chore(changeset): drop the Codex tool_response bullet already released by #404 --- .changeset/late-review-thread-fixes.md | 33 ++++++++++++-------------- 1 file changed, 15 insertions(+), 18 deletions(-) diff --git a/.changeset/late-review-thread-fixes.md b/.changeset/late-review-thread-fixes.md index 5c336abe9..a9d666939 100644 --- a/.changeset/late-review-thread-fixes.md +++ b/.changeset/late-review-thread-fixes.md @@ -3,29 +3,26 @@ --- Address the post-merge review findings on the dev epoch gate, native catalog, -portable validation, Codex hooks, and inspection: +portable validation, and inspection (#408): -- `EpochAdoptionPolicy` leases the adopted epoch until another epoch replaces - it or the policy closes, so store retention cannot delete the advertised +- `agent-bundle dev` 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; an epoch that cannot be leased is not adopted and the failure is published as `AB7211` status. -- The dev contract matrix opens the configured server on a target whose +- The `dev.contracts` matrix opens the configured server on a target whose manifest actually carries it, applies the session timeout per request instead of once for the whole matrix, and observes live progress through the session trace; lifecycle fixtures no longer depend on the SDK client's private - `_notificationHandlers` map (`ContractMatrixClient` gains an optional - `observeProgress` seam, exported as `ContractMatrixProgressSource`). -- A Native Playground catalog reader now waits for a hard-link publisher to - release its staging link before adopting the sidecar, and returns to - discovery when that publication is rolled back instead of caching a withdrawn - epoch. -- Ordinary `build`/`validate --artifact` runs the Agent Plugins byte lane over - the emitted `portable/` tree (`AB6035`–`AB6037`), so standard-invalid + `_notificationHandlers` map (`ContractMatrixClient` from `agent-bundle/test` + gains an optional `observeProgress` seam, exported as + `ContractMatrixProgressSource`). +- A Native Playground catalog reader waits for a hard-link publisher to release + its staging link before adopting the sidecar, and returns to discovery when + that publication is rolled back instead of caching a withdrawn epoch. +- `agent-bundle build` and `validate --artifact` run the Agent Plugins byte lane + over the emitted `portable/` tree (`AB6035`–`AB6037`), so standard-invalid documents fail before publication rather than only under `--host-validation`; header values reject every forbidden control character, not just CR/LF/NUL. -- Codex `PostToolUse` accepts any present JSON `tool_response`, matching the - pinned `"tool_response": true` schema, in both the event projection and the - generated native hook wrapper; Claude keeps the object check. -- `inspect` projects only the contract fields of an adapter capability row, so - extension fields on JavaScript adapters cannot shadow the capability name or - break `--json` serialization. +- `agent-bundle inspect` projects only the contract fields of an adapter + capability row, so extension fields on JavaScript adapters cannot shadow the + capability name or break `--json` serialization. From f7011168af959029aaf7738402cdda6f4c34edee Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 07:50:21 +0000 Subject: [PATCH 04/16] fix(playground,test): withdraw a failed catalog publication before releasing its staging link; bind custom observeProgress - #persistSnapshot rolls the sidecar back while the staging link still exists when the post-link directory fsync fails, so a concurrent reader keeps seeing an in-progress publication until the path is withdrawn instead of adopting a briefly singly linked file - contractProgressObserver invokes a client's observeProgress method with the client as receiver --- .../playground/native-playground-service.ts | 11 ++- packages/agent-bundle/src/test/contract.ts | 3 +- .../tests/dev-contract-runner.test.ts | 18 +++++ .../tests/native-playground-service.test.ts | 70 +++++++++++++++++++ 4 files changed, 100 insertions(+), 2 deletions(-) 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 c2712d023..976653d8b 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1149,10 +1149,19 @@ export class NativePlaygroundService { try { await handle.close(); } 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 (primary !== undefined && created && publicationIdentity !== undefined) { + try { await this.#publicationReceipt(path, publicationIdentity, true, true).rollback(); } + catch (error) { cleanupFailures.push(error); } + created = false; + } 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/test/contract.ts b/packages/agent-bundle/src/test/contract.ts index 07ab43b11..272901454 100644 --- a/packages/agent-bundle/src/test/contract.ts +++ b/packages/agent-bundle/src/test/contract.ts @@ -1225,7 +1225,8 @@ const sdkProgressObserver = ( export const contractProgressObserver = ( client: ContractMatrixClient, ): ContractMatrixProgressSource['observeProgress'] => { - if (typeof client.observeProgress === 'function') return client.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( diff --git a/packages/agent-bundle/tests/dev-contract-runner.test.ts b/packages/agent-bundle/tests/dev-contract-runner.test.ts index 47d48fd7e..40f4ac9e0 100644 --- a/packages/agent-bundle/tests/dev-contract-runner.test.ts +++ b/packages/agent-bundle/tests/dev-contract-runner.test.ts @@ -118,6 +118,24 @@ it('exposes live progress notifications through the session trace for lifecycle 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)) diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index a9d0983ac..24a9433b8 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1599,6 +1599,76 @@ it('never adopts a staged sidecar that its publisher rolls back, and republishes } }); +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('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'); From 3d19529c0f40250591e2300ce10b9c8e0e865c7e Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 08:10:03 +0000 Subject: [PATCH 05/16] chore(changeset): one-paragraph summary ending with the PR reference --- .changeset/late-review-thread-fixes.md | 41 +++++++++++--------------- 1 file changed, 17 insertions(+), 24 deletions(-) diff --git a/.changeset/late-review-thread-fixes.md b/.changeset/late-review-thread-fixes.md index a9d666939..a91fd52c8 100644 --- a/.changeset/late-review-thread-fixes.md +++ b/.changeset/late-review-thread-fixes.md @@ -2,27 +2,20 @@ "agent-bundle": patch --- -Address the post-merge review findings on the dev epoch gate, native catalog, -portable validation, and inspection (#408): - -- `agent-bundle dev` 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; an epoch that cannot be - leased is not adopted and the failure is published as `AB7211` status. -- The `dev.contracts` matrix opens the configured server on a target whose - manifest actually carries it, applies the session timeout per request instead - of once for the whole matrix, and observes live progress through the session - trace; lifecycle fixtures no longer depend on the SDK client's private - `_notificationHandlers` map (`ContractMatrixClient` from `agent-bundle/test` - gains an optional `observeProgress` seam, exported as - `ContractMatrixProgressSource`). -- A Native Playground catalog reader waits for a hard-link publisher to release - its staging link before adopting the sidecar, and returns to discovery when - that publication is rolled back instead of caching a withdrawn epoch. -- `agent-bundle build` and `validate --artifact` run the Agent Plugins byte lane - over the emitted `portable/` tree (`AB6035`–`AB6037`), so standard-invalid - documents fail before publication rather than only under `--host-validation`; - header values reject every forbidden control character, not just CR/LF/NUL. -- `agent-bundle inspect` projects only the contract fields of an adapter - capability row, so extension fields on JavaScript adapters cannot shadow the - capability name or break `--json` serialization. +`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, and observes lifecycle progress through the session trace +(`ContractMatrixClient` from `agent-bundle/test` gains an optional +`observeProgress` seam, `ContractMatrixProgressSource`). Native Playground +catalog readers wait for a hard-link publisher to release its staging link +before adopting the sidecar and return to discovery when that publication is +rolled back. `agent-bundle build` and `validate --artifact` run the Agent +Plugins byte lane over the emitted `portable/` tree (`AB6035`–`AB6037`) so +standard-invalid documents fail before publication rather than only under +`--host-validation`, and header values reject every forbidden control character. +`agent-bundle inspect` projects only the contract fields of an adapter +capability row, so JavaScript adapter extension fields cannot shadow the +capability name or break `--json`. (#408) From 0eb4c32b621888399b2dda374b8b6745535d2b46 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 08:32:40 +0000 Subject: [PATCH 06/16] fix(playground): recover a catalog staging link abandoned by an exited publisher After the settle deadline, a matching .stage--* link whose publisher pid no longer exists is an abandoned publication of an already fsynced sidecar: withdraw the orphan and adopt the sidecar instead of rejecting the epoch forever. A live publisher's staging link is still never yanked. Adds the @internal catalogStagingSettleDeadlineMs seam for deterministic tests. --- .../playground/native-playground-service.ts | 62 +++++++++++++++---- .../tests/native-playground-service.test.ts | 54 ++++++++++++++++ 2 files changed, 104 insertions(+), 12 deletions(-) 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 976653d8b..b87288c86 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,9 +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 rejecting the sidecar. */ +/** 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; @@ -601,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; @@ -608,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 })); @@ -1059,40 +1079,58 @@ export class NativePlaygroundService { * 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, as - * does a staging link that never settles within the deadline. + * 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 #awaitStagedPublication(file: FileHandle, path: string, metadata: Stats): Promise<'settled' | 'withdrawn'> { const invalid = () => new Error('Native Playground catalog snapshot is invalid.'); - if (metadata.nlink !== 2 || !(await this.#stagingLinkAccountsFor(path, metadata))) { - if ((await file.stat()).nlink === 1) return (await this.#sidecarStillLinked(path, metadata)) ? 'settled' : 'withdrawn'; + 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() + stagingPublicationSettleDeadlineMs; + 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 (await this.#sidecarStillLinked(path, metadata)) ? 'settled' : 'withdrawn'; - if (current.nlink !== 2 || !(await this.#stagingLinkAccountsFor(path, metadata))) throw invalid(); + 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) throw invalid(); + 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 }); + if ((await file.stat()).nlink === 1) return settledOrWithdrawn(); + throw invalid(); } } - async #stagingLinkAccountsFor(path: string, metadata: Stats): Promise { + /** 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 false; + return matches; } async #sidecarStillLinked(path: string, metadata: Stats): Promise { diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 24a9433b8..f4f81eae4 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1669,6 +1669,60 @@ it('withdraws a sidecar whose directory fsync fails before releasing its staging } }); +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 serviceFor = (discover: () => Promise): NativePlaygroundService => new NativePlaygroundService({ + catalogDirectory, + catalogStagingSettleDeadlineMs: 50, + 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 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); + const blocked = serviceFor(async () => { throw new Error('A blocked catalog must not fall back to discovery.'); }); + await expect(blocked.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); + await blocked.close(); + expect((await stat(sidecar)).nlink).toBe(2); + await rm(live); + + // A publisher that exited after link() but before cleanup left a complete, + // fsynced sidecar behind: the reader withdraws the orphan and adopts it. + await link(sidecar, join(catalogDirectory, `.${reference.epoch.id}.stage-${String(exitedPid())}-orphan`)); + 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([]); + } 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'); From f7b7b99d130534894c234a8524aee8a88eef45ea Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 08:52:26 +0000 Subject: [PATCH 07/16] fix(playground): fsync the catalog directory after withdrawing an abandoned staging link --- .../playground/native-playground-service.ts | 3 +++ .../tests/native-playground-service.test.ts | 25 +++++++++++++++++++ 2 files changed, 28 insertions(+) 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 b87288c86..a9e71a295 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1111,6 +1111,9 @@ export class NativePlaygroundService { 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. + await this.#syncCatalogDirectory(dirname(path)); if ((await file.stat()).nlink === 1) return settledOrWithdrawn(); throw invalid(); } diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index f4f81eae4..1f736c305 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1685,9 +1685,31 @@ it('recovers a staging link abandoned by an exited publisher after the settle de 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[] = []; + const storage: NativePlaygroundCatalogStorage = { + link, + 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)); + await target.sync(); + }; + } + const value = Reflect.get(target, property, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + }, + remove: rm, + } 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 }), @@ -1713,11 +1735,14 @@ it('recovers a staging link abandoned by an exited publisher after the settle de // A publisher that exited after link() but before cleanup left a complete, // fsynced sidecar behind: the reader withdraws the orphan and adopts it. await link(sidecar, join(catalogDirectory, `.${reference.epoch.id}.stage-${String(exitedPid())}-orphan`)); + 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 }); } From 15baf3fc3f38089157c47cae6f9a4691bc1d30b7 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 10:14:51 +0000 Subject: [PATCH 08/16] fix(dev): forward request _meta (progress token) through McpSession and the dev matrix client Lifecycle fixtures pass their generated progressToken as params._meta; the session adapter and McpSession.callTool dropped it, so generated routes never enabled sendProgress and every progress-gated lifecycle fixture failed the dev matrix. McpSessionToolCallOptions and McpClient.callTool now carry _meta. --- .../src/dev/dev-contract-runner.ts | 3 +++ .../src/dev/mcp-session/mcp-session-types.ts | 11 +++++++++- .../src/dev/mcp-session/mcp-session.ts | 6 ++++- .../tests/dev-contract-runner.test.ts | 22 ++++++++++++++++--- .../tests/mcp-session-service.test.ts | 11 +++++++++- 5 files changed, 47 insertions(+), 6 deletions(-) diff --git a/packages/agent-bundle/src/dev/dev-contract-runner.ts b/packages/agent-bundle/src/dev/dev-contract-runner.ts index 55dca94b5..b9bfae455 100644 --- a/packages/agent-bundle/src/dev/dev-contract-runner.ts +++ b/packages/agent-bundle/src/dev/dev-contract-runner.ts @@ -94,6 +94,9 @@ type MatrixSession = Pick< 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: requestSignal(session, options), 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/tests/dev-contract-runner.test.ts b/packages/agent-bundle/tests/dev-contract-runner.test.ts index 40f4ac9e0..be96dc849 100644 --- a/packages/agent-bundle/tests/dev-contract-runner.test.ts +++ b/packages/agent-bundle/tests/dev-contract-runner.test.ts @@ -32,6 +32,7 @@ it('rejects a matrix whose server is emitted for none of the project targets', ( }); interface RecordedRequest { + readonly meta?: unknown; readonly signal: AbortSignal | undefined; readonly timeoutMs: number | undefined; } @@ -40,11 +41,15 @@ const fakeSession = (timeoutMs: number) => { const requests: RecordedRequest[] = []; const listeners = new Set(); let sequence = 0; - const record = (options: { readonly signal?: AbortSignal; readonly timeoutMs?: number } | undefined) => { - requests.push({ signal: options?.signal, timeoutMs: options?.timeoutMs }); + 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 signal?: AbortSignal; readonly timeoutMs?: number }) => { + callTool: async (options: { readonly _meta?: unknown; readonly signal?: AbortSignal; readonly timeoutMs?: number }) => { record(options); return { content: [] }; }, @@ -104,6 +109,17 @@ it('applies the session timeout per matrix request instead of one deadline for t 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]); 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(); From bf1ba3e31d6fdb72b946baa8d8e2b38a928dc62f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 10:48:31 +0000 Subject: [PATCH 09/16] fix(dev,playground): restart the adoption drain after a handoff race; keep the staging link when a sidecar rollback fails - EpochAdoptionPolicy reschedules its drain from the completion handler when a candidate arrived between the loop's last empty check and #processing being cleared, and settled() waits through restarts - #persistSnapshot releases the staging link after a failed publication only once the owned sidecar is confirmed withdrawn, so a rollback failure never leaves a singly linked sidecar for readers to adopt --- .../src/dev/epoch-adoption-policy.ts | 16 ++++- .../playground/native-playground-service.ts | 14 +++-- .../tests/epoch-adoption-policy.test.ts | 36 ++++++++++++ .../tests/native-playground-service.test.ts | 58 +++++++++++++++++++ 4 files changed, 118 insertions(+), 6 deletions(-) diff --git a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts index 12c3fdf3c..5657fd209 100644 --- a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts +++ b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts @@ -155,8 +155,19 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { 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(); }); } @@ -168,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 { 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 a9e71a295..1b63535ee 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1162,13 +1162,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.'); } @@ -1193,14 +1194,19 @@ export class NativePlaygroundService { // 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. + // 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); } } - try { await this.#catalogStorage.remove(temporary, { force: true }); } - catch (error) { cleanupFailures.push(error); } } if (primary === undefined && cleanupFailures.length > 0 && created && publicationIdentity !== undefined) { try { await this.#publicationReceipt(path, publicationIdentity, true, true).rollback(); } diff --git a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts index ff612f47b..3082747d8 100644 --- a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts +++ b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts @@ -248,6 +248,42 @@ it('does not adopt an epoch it cannot lease and reports the failure as contract 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('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/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 1f736c305..3d832cbfd 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1669,6 +1669,64 @@ it('withdraws a sidecar whose directory fsync fails before releasing its staging } }); +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 { From 56c017e92af52ed9bf5fe40a79ba6de039617556 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 11:01:27 +0000 Subject: [PATCH 10/16] ci: retrigger checks for the rebased head From 3bfc760b345aa671ae3b4e9d9bff9f9151c7c72c Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 11:04:40 +0000 Subject: [PATCH 11/16] chore: drop the portable byte-lane changes superseded by #406; keep the _meta assertions --- .changeset/late-review-thread-fixes.md | 18 +++++------- packages/agent-bundle/tests/api.test.ts | 38 ------------------------- 2 files changed, 7 insertions(+), 49 deletions(-) diff --git a/.changeset/late-review-thread-fixes.md b/.changeset/late-review-thread-fixes.md index a91fd52c8..1f8144556 100644 --- a/.changeset/late-review-thread-fixes.md +++ b/.changeset/late-review-thread-fixes.md @@ -7,15 +7,11 @@ 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, and observes lifecycle progress through the session trace +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`). Native Playground -catalog readers wait for a hard-link publisher to release its staging link -before adopting the sidecar and return to discovery when that publication is -rolled back. `agent-bundle build` and `validate --artifact` run the Agent -Plugins byte lane over the emitted `portable/` tree (`AB6035`–`AB6037`) so -standard-invalid documents fail before publication rather than only under -`--host-validation`, and header values reject every forbidden control character. -`agent-bundle inspect` projects only the contract fields of an adapter -capability row, so JavaScript adapter extension fields cannot shadow the -capability name or break `--json`. (#408) +`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/packages/agent-bundle/tests/api.test.ts b/packages/agent-bundle/tests/api.test.ts index 109c0f23b..0fdb49463 100644 --- a/packages/agent-bundle/tests/api.test.ts +++ b/packages/agent-bundle/tests/api.test.ts @@ -340,44 +340,6 @@ it('accepts the public claude.dependencies config surface and plans its manifest } }); -it('fails an ordinary portable build on Agent Plugins normative-text violations without --host-validation', async () => { - const root = await createProject(); - try { - await writeFile(join(root, 'agent-bundle.config.ts'), [ - 'export default {', - " plugin: { name: 'portable-standard', version: '1.0.0' },", - " targets: ['portable'],", - ' mcp: { servers: { remote: {', - " headers: { 'X-Tenant': 'a', 'x-tenant': 'b' },", - " transport: 'streamable-http',", - " url: 'http://mcp.example.test/mcp',", - ' } } },', - '};', - '', - ].join('\n')); - - // The pinned schemas accept this document; only the standard's text forbids it. - await expect(build({ output: join(root, 'artifact-out'), root })).rejects.toMatchObject({ - diagnostics: expect.arrayContaining([ - expect.objectContaining({ - code: 'AB6036', - message: expect.stringContaining('repeats header "X-Tenant" under different casing'), - severity: 'error', - target: 'portable', - }), - expect.objectContaining({ - code: 'AB6036', - message: expect.stringContaining('uses plain HTTP against non-loopback host "mcp.example.test"'), - severity: 'error', - target: 'portable', - }), - ]), - }); - } finally { - await rm(join(root, '..'), { force: true, recursive: true }); - } -}); - it('reports one modern-MCP source diagnostic for a legacy SSE declaration', async () => { const root = await createProject(); try { From c9bd5a2fd75f5a0d540168a26814bea89ea43154 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 11:25:20 +0000 Subject: [PATCH 12/16] fix(playground): restore the staging guard when a recovery fsync fails --- .../playground/native-playground-service.ts | 23 ++++++++++++++++++- .../tests/native-playground-service.test.ts | 15 +++++++++++- 2 files changed, 36 insertions(+), 2 deletions(-) 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 1b63535ee..de9aff37f 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1113,12 +1113,33 @@ export class NativePlaygroundService { 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. - await this.#syncCatalogDirectory(dirname(path)); + 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 restored = await Promise.allSettled( + staging.map((entry) => this.#catalogStorage.link(path, join(dirname(path), entry))), + ); + if (restored.every((outcome) => outcome.status === 'fulfilled')) return; + try { await this.#catalogStorage.remove(path, { force: true }); } + catch { + // Nothing further can make this publication unsettled; the caller rejects it. + } + } + /** The epoch's own staging entries that alias this sidecar's inode. */ async #stagingLinksFor(path: string, metadata: Stats): Promise { const directory = dirname(path); diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 3d832cbfd..06dd661a4 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1744,6 +1744,7 @@ it('recovers a staging link abandoned by an exited publisher after the settle de const reference = epoch('epoch-abandoned-staging', join(root, 'artifact')); const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); const directorySyncs: string[] = []; + let directorySyncFailure: Error | undefined; const storage: NativePlaygroundCatalogStorage = { link, mkdir, @@ -1754,6 +1755,7 @@ it('recovers a staging link abandoned by an exited publisher after the settle de if (property === 'sync' && String(path) === catalogDirectory) { return async () => { directorySyncs.push(String(path)); + if (directorySyncFailure !== undefined) throw directorySyncFailure; await target.sync(); }; } @@ -1790,9 +1792,20 @@ it('recovers a staging link abandoned by an exited publisher after the settle de 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 and the sidecar stays unsettled. + const orphan = join(catalogDirectory, `.${reference.epoch.id}.stage-${String(exitedPid())}-orphan`); + await link(sidecar, orphan); + directorySyncFailure = Object.assign(new Error('directory sync failed'), { code: 'EIO' }); + const unsynced = serviceFor(async () => { throw new Error('An unsynced recovery must not fall back to discovery.'); }); + await expect(unsynced.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); + await unsynced.close(); + expect((await stat(sidecar)).nlink).toBe(2); + expect((await stat(orphan)).ino).toBe((await stat(sidecar)).ino); + directorySyncFailure = undefined; + // A publisher that exited after link() but before cleanup left a complete, // fsynced sidecar behind: the reader withdraws the orphan and adopts it. - await link(sidecar, join(catalogDirectory, `.${reference.epoch.id}.stage-${String(exitedPid())}-orphan`)); 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); From 2e6c8d117ba731627f8b513fffcb38cba33a8d28 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 11:52:19 +0000 Subject: [PATCH 13/16] fix(playground): accept a concurrently restored staging guard (EEXIST aliasing the sidecar) during recovery --- .../dev/playground/native-playground-service.ts | 16 +++++++++++++--- .../tests/native-playground-service.test.ts | 13 +++++++++++++ 2 files changed, 26 insertions(+), 3 deletions(-) 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 de9aff37f..7b7d242b7 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1130,9 +1130,7 @@ export class NativePlaygroundService { * staging links it removed, and if even that fails, withdraw the sidecar. */ async #restoreStagingGuard(path: string, staging: readonly string[]): Promise { - const restored = await Promise.allSettled( - staging.map((entry) => this.#catalogStorage.link(path, join(dirname(path), entry))), - ); + const restored = await Promise.allSettled(staging.map((entry) => this.#restoreStagingLink(path, entry))); if (restored.every((outcome) => outcome.status === 'fulfilled')) return; try { await this.#catalogStorage.remove(path, { force: true }); } catch { @@ -1140,6 +1138,18 @@ export class NativePlaygroundService { } } + /** 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); diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 06dd661a4..afa0eef53 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1802,6 +1802,19 @@ it('recovers a staging link abandoned by an exited publisher after the settle de await unsynced.close(); expect((await stat(sidecar)).nlink).toBe(2); expect((await stat(orphan)).ino).toBe((await stat(sidecar)).ino); + + // Two readers recovering the same orphan while fsync fails: the second one's + // guard re-link meets EEXIST from the first, which is the same alias, so the + // sidecar is kept rather than withdrawn. + const [firstRacer, secondRacer] = [ + serviceFor(async () => { throw new Error('A racing recovery must not fall back to discovery.'); }), + serviceFor(async () => { throw new Error('A racing recovery must not fall back to discovery.'); }), + ]; + const raced = await Promise.allSettled([firstRacer.catalog(reference), secondRacer.catalog(reference)]); + expect(raced.map((outcome) => outcome.status)).toEqual(['rejected', 'rejected']); + await Promise.all([firstRacer.close(), secondRacer.close()]); + expect((await stat(sidecar)).nlink).toBe(2); + expect((await stat(orphan)).ino).toBe((await stat(sidecar)).ino); directorySyncFailure = undefined; // A publisher that exited after link() but before cleanup left a complete, From b641117d1748f052a9b1b8b727dd7e99d70a041f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 12:16:41 +0000 Subject: [PATCH 14/16] fix(dev): recheck supersession after the adoption lease settles --- .../src/dev/epoch-adoption-policy.ts | 6 ++ .../tests/epoch-adoption-policy.test.ts | 62 +++++++++++++++++++ 2 files changed, 68 insertions(+) diff --git a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts index 5657fd209..b9c96ec71 100644 --- a/packages/agent-bundle/src/dev/epoch-adoption-policy.ts +++ b/packages/agent-bundle/src/dev/epoch-adoption-policy.ts @@ -221,6 +221,12 @@ export class EpochAdoptionPolicy implements EpochAdoptionSource { 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; + } } this.#latestEvaluation = evaluation; if (evaluation !== undefined) { diff --git a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts index 3082747d8..55166a31d 100644 --- a/packages/agent-bundle/tests/epoch-adoption-policy.test.ts +++ b/packages/agent-bundle/tests/epoch-adoption-policy.test.ts @@ -284,6 +284,68 @@ it('drains an epoch that arrives while the previous drain is finishing, at every } }); +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(); From 7cfbc6797b14888ad1a393b36a12202e005ad6d1 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 12:39:12 +0000 Subject: [PATCH 15/16] fix(playground): keep a fresh pid-owned guard when recovery can neither re-link nor withdraw the sidecar --- .../playground/native-playground-service.ts | 21 +++++++++--- .../tests/native-playground-service.test.ts | 34 +++++++++++++++++-- 2 files changed, 49 insertions(+), 6 deletions(-) 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 7b7d242b7..eaba4c692 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1131,10 +1131,23 @@ export class NativePlaygroundService { */ async #restoreStagingGuard(path: string, staging: readonly string[]): Promise { const restored = await Promise.allSettled(staging.map((entry) => this.#restoreStagingLink(path, entry))); - if (restored.every((outcome) => outcome.status === 'fulfilled')) return; - try { await this.#catalogStorage.remove(path, { force: true }); } - catch { - // Nothing further can make this publication unsettled; the caller rejects it. + const failures = restored.flatMap((outcome) => (outcome.status === 'rejected' ? [outcome.reason] : [])); + if (failures.length === 0) return; + try { + await this.#catalogStorage.remove(path, { force: true }); + return; + } catch (error) { + failures.push(error); + } + // 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(dirname(path), `.${basename(path, '.json')}.stage-${process.pid}-guard-${Math.random().toString(16).slice(2)}`); + try { + await this.#catalogStorage.link(path, guard); + } catch (error) { + failures.push(error); + throw new AggregateError(failures, 'Native Playground catalog recovery could not keep the sidecar guarded.', { cause: error }); } } diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index afa0eef53..06826dfd2 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1745,8 +1745,15 @@ it('recovers a staging link abandoned by an exited publisher after the settle de const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); const directorySyncs: string[] = []; let directorySyncFailure: Error | undefined; + // Fault injection for the recovery fallback chain: fail re-linking the + // removed staging names and unlinking the sidecar, but let a fresh guard link. + let relinkFailure: Error | undefined; + let sidecarRemoveFailure: Error | undefined; const storage: NativePlaygroundCatalogStorage = { - link, + link: async (source, destination) => { + if (relinkFailure !== undefined && String(destination).endsWith('-orphan')) throw relinkFailure; + await link(source, destination); + }, mkdir, open: async (path, flags, mode) => { const handle = await open(path, flags, mode); @@ -1764,7 +1771,10 @@ it('recovers a staging link abandoned by an exited publisher after the settle de }, }); }, - remove: rm, + 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, @@ -1815,6 +1825,26 @@ it('recovers a staging link abandoned by an exited publisher after the settle de await Promise.all([firstRacer.close(), secondRacer.close()]); 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 guard under this process's pid keeps it doubly linked. + relinkFailure = Object.assign(new Error('relink denied'), { code: 'EPERM' }); + sidecarRemoveFailure = Object.assign(new Error('sidecar busy'), { code: 'EBUSY' }); + const guarded = serviceFor(async () => { throw new Error('A guarded recovery must not fall back to discovery.'); }); + await expect(guarded.catalog(reference)).rejects.toThrow(); + await guarded.close(); + 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. + const held = serviceFor(async () => { throw new Error('A guarded catalog must not fall back to discovery.'); }); + await expect(held.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); + await held.close(); + await rm(join(catalogDirectory, guards[0]!)); + await link(sidecar, orphan); + relinkFailure = undefined; + sidecarRemoveFailure = undefined; directorySyncFailure = undefined; // A publisher that exited after link() but before cleanup left a complete, From d0861a12c19433b44e46f5e51750202c3a8948ed Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 13:02:00 +0000 Subject: [PATCH 16/16] fix(playground): fsync every compensating recovery guard before trusting it --- .../playground/native-playground-service.ts | 43 +++++---- .../tests/native-playground-service.test.ts | 87 ++++++++++++------- 2 files changed, 82 insertions(+), 48 deletions(-) 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 eaba4c692..a1c01f424 100644 --- a/packages/agent-bundle/src/dev/playground/native-playground-service.ts +++ b/packages/agent-bundle/src/dev/playground/native-playground-service.ts @@ -1130,25 +1130,36 @@ export class NativePlaygroundService { * staging links it removed, and if even that fails, withdraw the sidecar. */ async #restoreStagingGuard(path: string, staging: readonly string[]): Promise { - const restored = await Promise.allSettled(staging.map((entry) => this.#restoreStagingLink(path, entry))); - const failures = restored.flatMap((outcome) => (outcome.status === 'rejected' ? [outcome.reason] : [])); - if (failures.length === 0) return; - try { - await this.#catalogStorage.remove(path, { force: true }); - return; - } catch (error) { - failures.push(error); - } + 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(dirname(path), `.${basename(path, '.json')}.stage-${process.pid}-guard-${Math.random().toString(16).slice(2)}`); - try { - await this.#catalogStorage.link(path, guard); - } catch (error) { - failures.push(error); - throw new AggregateError(failures, 'Native Playground catalog recovery could not keep the sidecar guarded.', { cause: error }); - } + 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. */ diff --git a/packages/agent-bundle/tests/native-playground-service.test.ts b/packages/agent-bundle/tests/native-playground-service.test.ts index 06826dfd2..c97d1a4c4 100644 --- a/packages/agent-bundle/tests/native-playground-service.test.ts +++ b/packages/agent-bundle/tests/native-playground-service.test.ts @@ -1744,14 +1744,24 @@ it('recovers a staging link abandoned by an exited publisher after the settle de const reference = epoch('epoch-abandoned-staging', join(root, 'artifact')); const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`); const directorySyncs: string[] = []; - let directorySyncFailure: Error | undefined; - // Fault injection for the recovery fallback chain: fail re-linking the - // removed staging names and unlinking the sidecar, but let a fresh guard link. + // 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 (relinkFailure !== undefined && String(destination).endsWith('-orphan')) throw relinkFailure; + 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, @@ -1762,7 +1772,10 @@ it('recovers a staging link abandoned by an exited publisher after the settle de if (property === 'sync' && String(path) === catalogDirectory) { return async () => { directorySyncs.push(String(path)); - if (directorySyncFailure !== undefined) throw directorySyncFailure; + if (directorySyncFailuresLeft > 0) { + directorySyncFailuresLeft -= 1; + throw directorySyncFailure; + } await target.sync(); }; } @@ -1788,6 +1801,11 @@ it('recovers a staging link abandoned by an exited publisher after the settle de 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); @@ -1796,56 +1814,61 @@ it('recovers a staging link abandoned by an exited publisher after the settle de // 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); - const blocked = serviceFor(async () => { throw new Error('A blocked catalog must not fall back to discovery.'); }); - await expect(blocked.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); - await blocked.close(); + 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 and the sidecar stays unsettled. + // 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); - directorySyncFailure = Object.assign(new Error('directory sync failed'), { code: 'EIO' }); - const unsynced = serviceFor(async () => { throw new Error('An unsynced recovery must not fall back to discovery.'); }); - await expect(unsynced.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); - await unsynced.close(); + 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); - - // Two readers recovering the same orphan while fsync fails: the second one's - // guard re-link meets EEXIST from the first, which is the same alias, so the - // sidecar is kept rather than withdrawn. - const [firstRacer, secondRacer] = [ - serviceFor(async () => { throw new Error('A racing recovery must not fall back to discovery.'); }), - serviceFor(async () => { throw new Error('A racing recovery must not fall back to discovery.'); }), - ]; - const raced = await Promise.allSettled([firstRacer.catalog(reference), secondRacer.catalog(reference)]); - expect(raced.map((outcome) => outcome.status)).toEqual(['rejected', 'rejected']); - await Promise.all([firstRacer.close(), secondRacer.close()]); + 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 guard under this process's pid keeps it doubly linked. + // 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' }); - const guarded = serviceFor(async () => { throw new Error('A guarded recovery must not fall back to discovery.'); }); - await expect(guarded.catalog(reference)).rejects.toThrow(); - await guarded.close(); + 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. - const held = serviceFor(async () => { throw new Error('A guarded catalog must not fall back to discovery.'); }); - await expect(held.catalog(reference)).rejects.toThrow('catalog snapshot is invalid'); - await held.close(); + await rejectsUnsettled('A guarded catalog'); await rm(join(catalogDirectory, guards[0]!)); await link(sidecar, orphan); relinkFailure = undefined; sidecarRemoveFailure = undefined; - directorySyncFailure = 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.