From b032e85d0ce25f78c448b47b87458b733c09a086 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 10:44:15 +0200 Subject: [PATCH 1/6] fix(orchestrator): bound the whole sweep, not one more call inside it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three unbounded calls have wedged this sweep in a single day, on three different transports. Each was real, each was bounded, and each time the wedge came back one layer down: the FACTORY_STATE Durable Object calls (factory-cloud#78), the relayfile change-feed tail reads (#368, shipped and verified in 0.1.75), then the retry of the now-bounded call. This does not bound a fourth. It makes the class of failure survivable. The property it establishes: NO SWEEP CAN BE IN FLIGHT FOR LONGER THAN ITS BUDGET, whatever it is waiting on. Elapsed time is charged against ONE timer for the whole pass, so it does not matter which await is slow, how many there are, or how many times the sweep retries one of them. The next unbounded call degrades a sweep instead of ending dispatch. WHY A PER-CALL BOUND CANNOT DO THIS. `relayfileOperationTimeoutMs` bounds one relayfile call and cannot see the retry loop around it or a call on another transport. `reconcileTimeoutMs` bounds the CALLER'S WAIT from outside `runOnce()`, so expiry leaves the sweep running and every later cycle coalesces onto the same wedged promise (factory.ts `runOnce()`, the `#runOnceInFlight` branch) — which is why the deployed daemon never recovers. The budget expires from INSIDE `#runOnceWithDiscoveryFence`, so the sweep unwinds, the lease goes back, `#runOnceInFlight` clears, and the next cycle claims a fresh lease. MECHANISM, PLAINLY. `budget.run()` is a race, not a cancellation — the same limitation #368 documented, stated for the same reason. CAN: abandon an in-flight await, from any transport, and unwind the sweep. CANNOT: stop the abandoned work. The socket stays open, the SDK's own retry loop keeps running, and a side effect already in flight still lands. PARTIAL: `budget.signal` aborts at expiry, so anything honouring an AbortSignal is really cancelled — nothing in the sweep consumes it yet (the relayfile client mints its own per-call signal and that file is owned by another lane this week); it is exported so wiring it is one line. `assertNotExpired()` is a between-await check and is worth nothing against a call that never returns, but it does make an abandoned pass unwind at its next loop iteration rather than run to completion beside its replacement. TEARDOWN IS BOUNDED SEPARATELY. On the path that matters the budget is spent by construction, so teardown cannot run under it or the lease would never be released — and releasing it is the half that makes the next cycle clean. An unbounded release would re-create this wedge one layer down. It gets a 30 s deadline; an abandoned release costs an orphaned lease for one expiry window, which a later sweep reclaims (`claim.reclaimedLease`). DEFAULT IS THE EXISTING ENVELOPE, DELIBERATELY. `sweepBudgetMs` defaults to `reconcileTimeoutMs` (90 min) and is clamped to it, so no sweep that survives today is killed by this. The value is a policy dial, the mechanism is the fix. Tightening it has a real cost: the checkpoint commits only at the end, so a budget below realistic cold-mirror hydration (#36 measured 61 min in production) makes a slow boot a loop that never progresses. TESTS (11), must-fire/must-not-fire for each: - must-fire, end to end: a sweep whose first post-claim call never returns is aborted at its budget naming the phase, the lease release is OBSERVED on the store, and the next cycle runs a fresh sweep and dispatches. Fail-first verified by mechanism: with only factory.ts reverted it fails after 4038 ms with "sweep never settled" — the pass never settles, exactly as production. - must-fire, primitive: three 40 ms calls under a 120 ms budget — the third is rejected because the SWEEP is out of time, not because it is slow; a bounded-but-always-failing call inside an unbounded retry loop ends at the budget (the L3 shape) after more than one attempt; the signal aborts; a spent budget refuses to start new work against the dependency it gave up on. - must-not-fire: a healthy sweep under a snug 30 s budget produces results IDENTICAL to an unbounded control (pulled, dispatched, skipped, spawns) — without this the trivial wrong fix, abort everything, passes; a caller's own failure still surfaces as itself and is never re-clothed as a budget expiry; with `sweepBudgetMs: 0` the same hung call stays pending, so every rejection above is attributable to the budget and not to the wrapper. WHAT THIS DOES NOT COVER. - It does not make anything faster or find the hanging call. A wedged dependency still costs one whole budget per cycle. - It does not cancel. See MECHANISM above. - The abandoned pass runs concurrently with the sweep that replaces it if it ever unsticks. It cannot commit a checkpoint (the store's epoch guard) but its in-flight side effects still land. - Two `stop()`/shutdown paths and the `#runOnceWithReadinessDeadline` abandoned-wait bookkeeping are unchanged; a budget expiry reaches them as an ordinary sweep failure. - The default changes no timing. Recovery inside 90 minutes needs either a tighter `sweepBudgetMs` or the L3 retry bound the other lane owns. Co-Authored-By: Claude Opus 5 Session-Id: b1177efc-90da-4ff1-bda0-ef5de1b475e2 Session-Id: b1177efc-90da-4ff1-bda0-ef5de1b475e2 --- src/config/schema.ts | 43 +++ src/orchestrator/factory.ts | 188 +++++++++++-- src/orchestrator/sweep-budget.test.ts | 381 ++++++++++++++++++++++++++ src/orchestrator/sweep-budget.ts | 232 ++++++++++++++++ src/types.ts | 7 + 5 files changed, 826 insertions(+), 25 deletions(-) create mode 100644 src/orchestrator/sweep-budget.test.ts create mode 100644 src/orchestrator/sweep-budget.ts diff --git a/src/config/schema.ts b/src/config/schema.ts index 890ebe1f..c74c3ed1 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -61,6 +61,26 @@ const subscriptionSchema = z.object({ */ export const DEFAULT_READINESS_RECONCILE_TIMEOUT_MS = 90 * 60_000 +/** + * Aggregate budget for one discovery sweep (#372). + * + * Defaults to `DEFAULT_READINESS_RECONCILE_TIMEOUT_MS` on purpose. The two + * numbers describe the same envelope; what changes is the MECHANISM, and the + * mechanism is the deliverable. `reconcileTimeoutMs` rejects the caller's wait + * and leaves `runOnce()` running, so every later cycle coalesces onto the + * wedged pass and the daemon never recovers. The sweep budget rejects from + * inside the fence, so the lease is released and the next cycle starts clean. + * + * Shipping it equal keeps this change free of new timing risk: no sweep that + * survives today is killed by it. Tightening it is a separate, evidence-driven + * decision with a real cost — the sweep commits its checkpoint only at the end, + * so a budget below realistic cold-mirror hydration (#36 measured 61 minutes in + * production) converts a slow boot into a loop that never makes progress, + * which is the same trap `reconcileTimeoutMs` documents above. That is why this + * is a config dial and not a constant. + */ +export const DEFAULT_DISCOVERY_SWEEP_BUDGET_MS = DEFAULT_READINESS_RECONCILE_TIMEOUT_MS + const liveSubscriptionSchema = z.object({ transport: z.enum(['subscribe-and-poll', 'subscribe', 'poll']).default('subscribe-and-poll'), pollIntervalMs: z.number().int().min(50).default(5_000), @@ -84,6 +104,19 @@ const liveSubscriptionSchema = z.object({ */ relayfileOperationTimeoutMs: z.number().int().min(50).max(60 * 60_000) .default(DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS), + /** + * Bounds the WHOLE sweep (#372). + * + * Distinct from both neighbours above, and the only one of the three that is + * agnostic to which dependency hangs. `relayfileOperationTimeoutMs` bounds one + * relayfile call and cannot see a retry loop around it or a call on another + * transport; `reconcileTimeoutMs` bounds the caller's wait and leaves the + * sweep running underneath it. This one is charged against a single timer for + * the entire pass, so a sweep cannot outlive it however many calls, retries + * or transports it is spread across. + */ + sweepBudgetMs: z.number().int().min(50).max(6 * 60 * 60_000) + .default(DEFAULT_DISCOVERY_SWEEP_BUDGET_MS), }).superRefine((value, ctx) => { // A deadline below the interval kills every pass that takes longer than one // tick, which is most of them on a cold mirror. @@ -94,6 +127,16 @@ const liveSubscriptionSchema = z.object({ message: `reconcileTimeoutMs (${value.reconcileTimeoutMs}) must be at least reconcileIntervalMs (${value.reconcileIntervalMs})`, }) } + // The sweep budget has to be the tighter of the two, or the wait gives up + // first and the sweep it abandoned keeps running for the next cycle to + // coalesce onto — the exact behaviour the budget exists to remove. + if (value.sweepBudgetMs > value.reconcileTimeoutMs) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['sweepBudgetMs'], + message: `sweepBudgetMs (${value.sweepBudgetMs}) must not exceed reconcileTimeoutMs (${value.reconcileTimeoutMs})`, + }) + } }).default({}) export const DEFAULT_AGENT_HOLD_TIMEOUT_MS = 4 * 60 * 60_000 diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 73150a74..53b0f551 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3,7 +3,12 @@ import { randomUUID } from 'node:crypto' import { readFile } from 'node:fs/promises' import { dirname, isAbsolute, resolve } from 'node:path' -import { DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, FactoryConfigSchema, type FactoryConfig } from '../config/schema' +import { + DEFAULT_DISCOVERY_SWEEP_BUDGET_MS, + DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, + FactoryConfigSchema, + type FactoryConfig, +} from '../config/schema' import { DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS, RelayfileOperationTimeoutError, @@ -65,6 +70,13 @@ import type { Clock, Logger } from '../ports/system' import type { AgentWorktree, AgentWorktreeManager, AgentWorktreeRepository } from '../ports/worktree' import { factoryWorktreeIssueSlug, factoryWorktreePath } from '../git/agent-worktree' import { InMemoryStateStore } from '../state/in-memory-state-store' +import { + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + DiscoverySweepBudgetExceededError, + startDiscoverySweepBudget, + withSweepTeardownDeadline, + type DiscoverySweepBudget, +} from './sweep-budget' import { dispatchHandedOffToBabysitters, dispatchLifecycleOccupiesSlot, @@ -877,6 +889,21 @@ export class FactoryLoop implements Factory { * returns: a deadline checked between awaits never regains control to check. */ #relayfileOperationTimeoutMs = DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS + /** + * Aggregate budget for ONE sweep (#372). + * + * The third bound on this path and the only transport-agnostic one. + * `#relayfileOperationTimeoutMs` bounds one relayfile call and cannot see the + * retry loop around it; `#readinessReconcileTimeoutMs` bounds the *wait* and + * leaves `runOnce()` running for the next cycle to coalesce onto. This is + * charged against one timer for the whole pass, so however many calls, + * retries and transports the sweep is spread across, it cannot outlive this. + * + * Held at `min(configured, #readinessReconcileTimeoutMs)`: a budget looser + * than the wait would let the wait give up first, which is the behaviour it + * exists to remove. + */ + #discoverySweepBudgetMs = DEFAULT_DISCOVERY_SWEEP_BUDGET_MS // Set for exactly as long as a sweep is running. `state` is derived from // this, so an in-flight pass can no longer masquerade as the last settled one. #readinessReconcileInFlightSinceMs?: number @@ -1121,6 +1148,10 @@ export class FactoryLoop implements Factory { // Also read here, not only in `#startLiveSubscription`: a standalone // `runOnce()` never starts the live subscription and must still be bounded. this.#relayfileOperationTimeoutMs = config.liveSubscription.relayfileOperationTimeoutMs + this.#discoverySweepBudgetMs = Math.min( + config.liveSubscription.sweepBudgetMs, + config.liveSubscription.reconcileTimeoutMs, + ) this.#mount = ports.mount // Resolved role<->state mapping. The CLI injects a name-resolved, per-team // resolution via ports; fall back to one built from explicit stateIds plus @@ -1963,6 +1994,10 @@ export class FactoryLoop implements Factory { // floor here: a deadline under one interval would kill every pass. this.#readinessReconcileTimeoutMs = Math.max(options.reconcileTimeoutMs, options.reconcileIntervalMs) this.#relayfileOperationTimeoutMs = options.relayfileOperationTimeoutMs + // Same re-application as the deadline above: `start()` overrides bypass the + // schema's cross-field check, and a budget looser than the wait is not a + // budget. + this.#discoverySweepBudgetMs = Math.min(options.sweepBudgetMs, this.#readinessReconcileTimeoutMs) this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -2137,6 +2172,7 @@ export class FactoryLoop implements Factory { reconcileTimeoutMs: overrides.reconcileTimeoutMs ?? this.#config.liveSubscription.reconcileTimeoutMs, relayfileOperationTimeoutMs: overrides.relayfileOperationTimeoutMs ?? this.#config.liveSubscription.relayfileOperationTimeoutMs, + sweepBudgetMs: overrides.sweepBudgetMs ?? this.#config.liveSubscription.sweepBudgetMs, } } @@ -3028,17 +3064,62 @@ export class FactoryLoop implements Factory { } } + /** + * One sweep, under one aggregate budget (#372). + * + * The budget wraps the fence rather than the fence's caller, and that is the + * whole point. #296's deadline lives in `#runOnceWithReadinessDeadline`, + * outside `runOnce()`: expiry rejects the wait and leaves the sweep running, + * so the next cycle coalesces onto the same wedged promise (`runOnce()`, the + * `#runOnceInFlight` branch) and the daemon never recovers. Expiring in HERE + * unwinds the body below, which releases the discovery lease on its way out + * and lets `runOnce()` clear `#runOnceInFlight` — so the next cycle claims a + * fresh lease and runs clean. + * + * See `sweep-budget.ts` for what the mechanism can and cannot interrupt. In + * short: it abandons the in-flight await, it does not cancel the call. + */ async #runOnceWithDiscoveryFence(opts: { dryRun?: boolean }): Promise { + const budget = startDiscoverySweepBudget(this.#discoverySweepBudgetMs) + const sweepStartedAtMs = this.#clock.now() + try { + return await this.#runDiscoverySweep(opts, budget) + } catch (error) { + if (error instanceof DiscoverySweepBudgetExceededError) { + this.#increment('discoverySweepBudgetExceeded') + this.#logger.error?.('[factory] discovery sweep aborted at its aggregate budget', { + budgetMs: error.budgetMs, + // The await the sweep was abandoned on. The one diagnostic no + // per-call bound can produce once the sweep is already wedged: it + // says WHICH transport this wedge is on without anyone having to + // guess which layer to bound next. + phase: error.phase, + elapsedMs: this.#elapsedSince(sweepStartedAtMs), + }) + } + throw error + } finally { + budget.dispose() + } + } + + async #runDiscoverySweep( + opts: { dryRun?: boolean }, + budget: DiscoverySweepBudget, + ): Promise { const sweepStartedAtMs = this.#clock.now() if (!(opts.dryRun ?? this.#config.dryRun)) { - await this.#assertFleetControlPlaneAvailable() + // Under the budget, and first, because this is where the 2026-08-25 + // 07:52:59Z wedge sat: a pre-claim probe on a transport neither #351 nor + // #368 covers. The budget does not care which one it is. + await budget.run('fleet-control-plane-probe', () => this.#assertFleetControlPlaneAvailable()) } - let claim = await this.#state.claimDiscoverySweep( + let claim = await budget.run('discovery-lease-claim', () => this.#state.claimDiscoverySweep( this.#workspaceId, this.#discoverySweepOwner, this.#clock.now(), DISCOVERY_SWEEP_LEASE_MS, - ) + )) if (!claim.acquired && claim.reason === 'backoff') { const delayMs = Math.max(0, claim.state.backoffUntilMs - this.#clock.now()) this.#increment('discoveryBackoffWaits') @@ -3047,13 +3128,13 @@ export class FactoryLoop implements Factory { backoffUntilMs: claim.state.backoffUntilMs, consecutiveOverloads: claim.state.consecutiveOverloads, }) - await this.#clock.sleep(delayMs) - claim = await this.#state.claimDiscoverySweep( + await budget.run('discovery-backoff-wait', () => this.#clock.sleep(delayMs)) + claim = await budget.run('discovery-lease-claim', () => this.#state.claimDiscoverySweep( this.#workspaceId, this.#discoverySweepOwner, this.#clock.now(), DISCOVERY_SWEEP_LEASE_MS, - ) + )) } if (!claim.acquired || !claim.lease) { this.#increment('discoverySweepsSkippedInFlight') @@ -3099,13 +3180,16 @@ export class FactoryLoop implements Factory { this.#startDiscoverySweepRenewal(claim.lease.epoch) let leaseReleased = false try { - this.#discoverySession = await this.#prepareDiscoverySession(claim) + this.#discoverySession = await budget.run( + 'discovery-session', + () => this.#prepareDiscoverySession(claim), + ) // #297: a 429 raised anywhere in the sweep used to latch and be rethrown // here, discarding a completed pass — every issue read, every dispatch — // because of one transient shed operation. The work this sweep did is // now kept instead, and the ratchet below records that the dependency is // shedding but still serving. - const report = await this.#performRunOnce(opts) + const report = await budget.run('run-once', () => this.#performRunOnce(opts, budget)) // The exception, and the reason skipping shed units cannot make a sweep // unconditionally green: a sweep that was shed AND got no work unit // through accomplished nothing. There is no progress to preserve, and @@ -3116,17 +3200,23 @@ export class FactoryLoop implements Factory { if (this.#discoveryOverloadError !== undefined && !this.#discoverySweepProgress) { throw this.#discoveryOverloadError } - const checkpoint = await this.#finalizeDiscoveryCheckpoint() + const checkpoint = await budget.run('discovery-checkpoint', () => this.#finalizeDiscoveryCheckpoint()) // Do not clear the durable lease while a renewal can still be waiting on // the same state-file lock. A late renewal that observes the completed // (lease-less) checkpoint is a false lease-loss signal and can poison an // otherwise successful reconcile cycle. - await this.#stopDiscoverySweepRenewal() + await budget.run('discovery-renewal-stop', () => this.#stopDiscoverySweepRenewal()) if (this.#discoverySweepLeaseLost) { throw new Error('discovery sweep lease was lost before checkpoint commit') } const residual = this.#discoveryOverloadOutcome(claim.state.consecutiveOverloads, 'committed') - const completed = await this.#commitDiscoverySweep(claim.lease.epoch, checkpoint, residual) + // Under the budget too. A hung commit is the same class of wedge as a + // hung read, and the epoch guard in the store makes a late one a no-op: + // the teardown below has already released this epoch's lease. + const completed = await budget.run( + 'discovery-commit', + () => this.#commitDiscoverySweep(claim.lease!.epoch, checkpoint, residual), + ) leaseReleased = completed if (!completed) throw new Error('discovery sweep lease was lost before completion') // This is the progress boundary consumed by deployment health. A timer @@ -3141,17 +3231,23 @@ export class FactoryLoop implements Factory { }) return report } catch (error) { - await this.#stopDiscoverySweepRenewal() + await this.#sweepTeardownStep('discovery sweep renewal stop', () => this.#stopDiscoverySweepRenewal()) const overload = relayfileOverload(error) if (overload) { const outcome = this.#discoveryOverloadOutcome(claim.state.consecutiveOverloads, 'aborted', error)! - leaseReleased = await this.#state.deferDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - claim.lease.epoch, - outcome.backoffUntilMs, - outcome.consecutiveOverloads, - ) + // Bounded for the same reason the release below is: this is the other + // path that hands the lease back, and an unbounded one would hold + // `#runOnceInFlight` open past the budget that just expired. + leaseReleased = await withSweepTeardownDeadline( + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + () => this.#state.deferDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + claim.lease!.epoch, + outcome.backoffUntilMs, + outcome.consecutiveOverloads, + ), + ) ?? false this.#logger.warn?.('[factory] Relayfile discovery overloaded; backing off before another sweep', { status: overload.status, reason: overload.reason, @@ -3171,7 +3267,7 @@ export class FactoryLoop implements Factory { } throw error } finally { - await this.#stopDiscoverySweepRenewal() + await this.#sweepTeardownStep('discovery sweep renewal stop', () => this.#stopDiscoverySweepRenewal()) this.#discoverySession = undefined this.#discoverySweepEpoch = undefined this.#discoverySweepStartedAtMs = undefined @@ -3188,15 +3284,48 @@ export class FactoryLoop implements Factory { // happens to run and reset it at the top of this method. this.#discoverySweepLeaseLost = false if (!leaseReleased) { - await this.#state.releaseDiscoverySweep( + // The half of the budget that makes the NEXT cycle clean, so it gets + // its own deadline rather than the spent one: an unbounded release + // would re-create the wedge one layer down, which is the pattern this + // change exists to end. An abandoned release is survivable — the + // durable lease carries its own expiry and a later sweep reclaims it + // as an orphan (`claim.reclaimedLease` above). + await this.#sweepTeardownStep('discovery sweep lease release', () => this.#state.releaseDiscoverySweep( this.#workspaceId, this.#discoverySweepOwner, - claim.lease.epoch, - ) + claim.lease!.epoch, + )) } } } + /** + * One sweep-teardown step, under its own deadline. + * + * Teardown cannot run under the sweep's aggregate budget: on the path that + * matters the budget is already spent, so every step would reject and the + * lease would never be released. It gets a short independent deadline + * instead. Abandoning it costs an orphaned lease for one expiry window; + * NOT bounding it costs the whole invariant, because a hung release holds + * `#runOnceInFlight` open and every later cycle coalesces onto it. + */ + async #sweepTeardownStep(label: string, step: () => Promise): Promise { + const outcome = await withSweepTeardownDeadline( + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + async () => { + await step() + return true as const + }, + ) + if (outcome === undefined) { + this.#increment('discoverySweepTeardownDeadlineExceeded') + this.#logger.warn?.('[factory] discovery sweep teardown step abandoned at its deadline', { + step: label, + timeoutMs: DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + }) + } + } + /** * Commit the sweep, carrying any residual overload backoff into the store. * @@ -3310,7 +3439,10 @@ export class FactoryLoop implements Factory { } } - async #performRunOnce(opts: { dryRun?: boolean } = {}): Promise { + async #performRunOnce( + opts: { dryRun?: boolean } = {}, + budget?: DiscoverySweepBudget, + ): Promise { const dryRun = opts.dryRun ?? this.#config.dryRun const startedAtMs = this.#clock.now() const relayfileWaitWarningsAtStart = this.#counters.relayfileOperationWaitWarnings ?? 0 @@ -3371,6 +3503,12 @@ export class FactoryLoop implements Factory { const issueEntries: Array<{ path: string; issue?: LinearIssue }> = [] for (const path of paths) { + // A between-await check, worth exactly what #368 said such a check is + // worth against a call that never returns: nothing. What it buys is + // the other half — a pass already abandoned at the budget unwinds at + // its next iteration if it ever regains control, instead of running to + // completion beside the sweep that replaced it. + budget?.assertNotExpired('run-once') let issue: LinearIssue | undefined let shed = false try { diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts new file mode 100644 index 00000000..9ba2e91b --- /dev/null +++ b/src/orchestrator/sweep-budget.test.ts @@ -0,0 +1,381 @@ +import { describe, expect, it } from 'vitest' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { + FactoryConfigSchema, + createFactory, + type FactoryConfig, + type LinearIssue, + type TriageDecision, + type TriageEngine, +} from '../index' +import { FakeFleetClient, FakeMountClient } from '../testing' +import { withDeadline } from '../testing/deadline' +import { InMemoryStateStore } from '../state/in-memory-state-store' +import { + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + DiscoverySweepBudgetExceededError, + startDiscoverySweepBudget, + withSweepTeardownDeadline, +} from './sweep-budget' + +/** + * The aggregate sweep budget (#372). + * + * Three unbounded calls have wedged this sweep in one day, on three different + * transports, each one bounded afterwards and each time the wedge returned one + * layer down. So the subject here is not another call. It is the property that + * a sweep cannot be in flight for longer than its budget REGARDLESS of what it + * is waiting on — which is what turns the next unbounded call into a degraded + * sweep instead of the end of dispatch. + * + * Every must-fire below hangs a call that never returns. Before this change + * each one hangs the test itself, at vitest's 5 s default: that is the defect, + * in the same shape production had it. + */ + +const NEVER = (): Promise => new Promise(() => undefined) + +describe('sweep budget primitive', () => { + it('must-fire: abandons an in-flight await that never returns, naming the phase', async () => { + const budget = startDiscoverySweepBudget(30) + try { + await expect(budget.run('discovery-session', NEVER)).rejects.toThrow(DiscoverySweepBudgetExceededError) + } finally { + budget.dispose() + } + }) + + it('must-fire: charges every phase against ONE clock, so the sum cannot exceed the budget', async () => { + // The property a per-call deadline cannot have. Each call here is well + // inside the budget on its own; only their sum is not. + const budget = startDiscoverySweepBudget(120) + const sleep = (ms: number) => new Promise((resolve) => setTimeout(() => resolve('served'), ms)) + try { + expect(await budget.run('discovery-session', () => sleep(40))).toBe('served') + expect(await budget.run('discovery-checkpoint', () => sleep(40))).toBe('served') + // The third 40 ms call is identical to the two that were served. It is + // rejected because the SWEEP is out of time, not because it is slow. + await expect(budget.run('discovery-commit', () => sleep(80))).rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + phase: 'discovery-commit', + budgetMs: 120, + }) + } finally { + budget.dispose() + } + }) + + it('must-fire: a bounded call inside an unbounded retry loop still ends at the budget (L3)', async () => { + // The deployed 0.1.75 shape: #368's per-call deadline fires every time, and + // the retry around it makes the total unbounded anyway. `attempts` is + // deliberately unbounded — the budget is the only thing that stops it. + const budget = startDiscoverySweepBudget(80) + let attempts = 0 + const alwaysRejects = async (): Promise => { + attempts += 1 + await new Promise((resolve) => setTimeout(resolve, 5)) + throw new Error('relayfile getEvents did not respond within 5ms') + } + const retryForever = async (): Promise => { + for (;;) { + try { + return await budget.run('run-once', alwaysRejects) + } catch (error) { + if (error instanceof DiscoverySweepBudgetExceededError) throw error + // Exactly what L3 does: swallow the per-call deadline and go again. + } + } + } + try { + await expect(withDeadline(retryForever(), 4_000, 'retry loop never ended')).rejects + .toThrow(DiscoverySweepBudgetExceededError) + expect(attempts).toBeGreaterThan(1) + } finally { + budget.dispose() + } + }) + + it('must-fire: aborts its signal at expiry, so anything that honours one is really cancelled', async () => { + const budget = startDiscoverySweepBudget(25) + try { + expect(budget.signal.aborted).toBe(false) + await expect(budget.run('run-once', NEVER)).rejects.toThrow(DiscoverySweepBudgetExceededError) + expect(budget.signal.aborted).toBe(true) + expect(budget.signal.reason).toBeInstanceOf(DiscoverySweepBudgetExceededError) + } finally { + budget.dispose() + } + }) + + it('must-fire: refuses to start new work once the budget is spent', async () => { + const budget = startDiscoverySweepBudget(25) + let started = false + try { + await expect(budget.run('run-once', NEVER)).rejects.toThrow(DiscoverySweepBudgetExceededError) + await expect(budget.run('discovery-commit', async () => { + started = true + return 'served' + })).rejects.toThrow(DiscoverySweepBudgetExceededError) + // A spent budget must not issue another request against the dependency + // it just gave up on. + expect(started).toBe(false) + expect(() => budget.assertNotExpired('run-once')).toThrow(DiscoverySweepBudgetExceededError) + } finally { + budget.dispose() + } + }) + + it('must-not-fire: work that finishes inside the budget returns unchanged', async () => { + const budget = startDiscoverySweepBudget(2_000) + try { + const report = { dispatched: ['ar-1'], candidates: 2 } + expect(await budget.run('run-once', async () => report)).toBe(report) + expect(budget.expired()).toBe(false) + expect(budget.signal.aborted).toBe(false) + expect(() => budget.assertNotExpired('run-once')).not.toThrow() + // The caller's own failure still surfaces as itself, never re-clothed as + // a budget expiry — the two have completely different remedies. + const failure = new Error('dispatch failed') + await expect(budget.run('run-once', async () => { throw failure })).rejects.toBe(failure) + } finally { + budget.dispose() + } + }) + + it('must-not-fire: with no budget the same hung call stays pending, so the rejections above are the budget', async () => { + // The control for every must-fire above. Without it a wrapper that + // rejected unconditionally would pass all of them. + const budget = startDiscoverySweepBudget(0) + let settled = false + try { + const pending = budget.run('run-once', NEVER).then( + () => { settled = true }, + () => { settled = true }, + ) + await Promise.race([pending, new Promise((resolve) => setTimeout(resolve, 200))]) + expect(settled).toBe(false) + expect(budget.expired()).toBe(false) + expect(budget.budgetMs).toBeUndefined() + } finally { + budget.dispose() + } + }) +}) + +describe('sweep teardown deadline', () => { + it('must-fire: abandons a teardown step that never returns, rather than holding the sweep open', async () => { + expect(await withDeadline( + withSweepTeardownDeadline(25, NEVER), + 4_000, + 'teardown deadline never fired', + )).toBeUndefined() + }) + + it('must-not-fire: a served teardown returns its value, and a failing one still throws', async () => { + expect(await withSweepTeardownDeadline(2_000, async () => true)).toBe(true) + const failure = new Error('release rejected') + await expect(withSweepTeardownDeadline(2_000, async () => { throw failure })).rejects.toBe(failure) + // A teardown deadline is not a licence to swallow failures that used to + // surface, and it is short enough that it cannot be the thing that wedges. + expect(DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS).toBeLessThan(60_000) + }) +}) + +/* ------------------------------------------------------------------------- */ +/* The same invariant, driven through a real sweep. */ +/* ------------------------------------------------------------------------- */ + +const ready = '11111111-1111-4111-8111-111111111111' +const implementing = '22222222-2222-4222-8222-222222222222' +const done = '33333333-3333-4333-8333-333333333333' +const planning = '44444444-4444-4444-8444-444444444444' + +const config = (sweepBudgetMs: number, registryRoot: string): FactoryConfig => FactoryConfigSchema.parse({ + workspaceId: 'factory-sweep-budget', + repos: { + byLabel: { pear: 'AgentWorkforce/pear' }, + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + triage: { maxImplementers: 4 }, + batchSize: 4, + stateIds: { readyForAgent: ready, agentImplementing: implementing, done, inPlanning: planning }, + verification: { enabled: false }, + loop: { + registryPath: join(registryRoot, 'registry.json'), + heartbeatPath: join(registryRoot, 'heartbeat.json'), + }, + liveSubscription: { sweepBudgetMs }, +}) + +const issuePath = (n: number) => `/linear/issues/AR-${n}__uuid-${n}.json` + +const issueFile = (n: number) => ({ + provider: 'linear', + objectType: 'issue', + objectId: `uuid-${n}`, + payload: { + id: `uuid-${n}`, + identifier: `AR-${n}`, + title: `[factory-e2e] Fix factory issue ${n}`, + description: 'Implement the requested fix in src/orchestrator/factory.ts and verify it with tests.', + stateId: ready, + url: `https://linear.app/agent-relay/issue/AR-${n}/factory-issue-${n}`, + labels: [{ name: 'pear' }], + labelIds: ['label-id-not-used-by-parser'], + team: { key: 'AR', name: 'Agent Relay' }, + project: { name: 'Factory' }, + state: { id: ready, name: 'Ready for Agent' }, + }, +}) + +class StaticTriage implements TriageEngine { + async triage(issue: LinearIssue): Promise { + const number = issue.key.match(/\d+/)?.[0] ?? '0' + return { + issue: { uuid: issue.uuid, key: issue.key, path: issue.path }, + routes: [{ repo: 'AgentWorkforce/pear', clonePath: '/work/pear', rationale: 'test route' }], + scope: 'single', + implementers: [{ + name: `ar-${number}-impl`, + role: 'implementer', + capability: 'spawn:codex', + model: 'codex', + task: `Implement ${issue.key}`, + repo: 'AgentWorkforce/pear', + clonePath: '/work/pear', + node: 'self', + }], + reviewer: { + name: `ar-${number}-review`, + role: 'reviewer', + capability: 'spawn:claude', + model: 'claude', + task: `Review ${issue.key}`, + repo: 'AgentWorkforce/pear', + clonePath: '/work/pear', + node: 'self', + }, + thin: false, + confidence: 'high', + rationale: 'static test decision', + } + } +} + +/** + * A mount whose first watermark read never answers. + * + * `#prepareDiscoverySession` makes this call immediately after the discovery + * lease is claimed, so a sweep that hangs here holds the lease — the position + * the 0.1.74 wedge occupied, and the position every later layer has reoccupied + * on a different transport. + */ +class HangingWatermarkMount extends FakeMountClient { + hang = true + hungCalls = 0 + + override async getEventHighWatermark(opts: { provider?: string } = {}): Promise { + if (this.hang) { + this.hungCalls += 1 + return await NEVER() + } + return await super.getEventHighWatermark(opts) + } +} + +/** A store that records the lease handbacks, so "released" is observed, not inferred. */ +class LeaseWatchingStateStore extends InMemoryStateStore { + readonly released: number[] = [] + + override async releaseDiscoverySweep(workspaceId: string, owner: string, epoch: number): Promise { + this.released.push(epoch) + await super.releaseDiscoverySweep(workspaceId, owner, epoch) + } +} + +describe('a wedged sweep is bounded end to end (#372)', () => { + it('must-fire: aborts at the budget, releases the lease, and the NEXT cycle runs clean', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-')) + const mount = new HangingWatermarkMount({ + [issuePath(901)]: issueFile(901), + }) + const fleet = new FakeFleetClient() + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + const factory = createFactory(config(150, root), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + + try { + // BEFORE this change this call never settles and the test dies at + // vitest's 5 s default — the production defect, not an assertion detail. + // The 4 s guard is inside that default so the failure names itself. + await expect(withDeadline(factory.runOnce(), 4_000, 'sweep never settled')) + .rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + // The abandoned await is named, which is the diagnostic no per-call + // bound can produce once the sweep is already wedged. + phase: 'discovery-session', + budgetMs: 150, + }) + expect(mount.hungCalls).toBe(1) + // The lease is handed back on the way out. Without this the next cycle + // has nothing to claim and the wedge simply moves. + expect(stateStore.released).toHaveLength(1) + + // The next cycle. It must start a NEW sweep rather than coalesce onto + // the abandoned one — the failure mode #296's deadline left behind. + mount.hang = false + const report = await withDeadline(factory.runOnce(), 4_000, 'next cycle never ran') + expect(report.pulled).toHaveLength(1) + expect(report.dispatched).toHaveLength(1) + expect(report.discoveryDeferred).toBeUndefined() + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-901-impl-pear', 'ar-901-review']) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-not-fire: a sweep that completes inside its budget is not aborted and its result is unchanged', async () => { + // Without this the trivial wrong fix — abort every sweep immediately — + // passes the must-fire above. + const run = async (sweepBudgetMs: number) => { + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-ok-')) + const fleet = new FakeFleetClient() + const factory = createFactory(config(sweepBudgetMs, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901), [issuePath(902)]: issueFile(902) }), + fleet, + stateStore: new InMemoryStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + const report = await withDeadline(factory.runOnce(), 4_000, 'healthy sweep never settled') + return { + pulled: report.pulled.map((issue) => issue.key).sort(), + dispatched: report.dispatched.length, + skipped: report.skipped.length, + spawns: fleet.spawns.map((spawn) => spawn.name), + } + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + } + + // A snug budget, and an effectively unbounded control. Identical results + // are what "the budget did not change this sweep" means. + const budgeted = await run(30_000) + const control = await run(90 * 60_000) + expect(budgeted.dispatched).toBeGreaterThan(0) + expect(budgeted).toEqual(control) + }) +}) diff --git a/src/orchestrator/sweep-budget.ts b/src/orchestrator/sweep-budget.ts new file mode 100644 index 00000000..bf7ae0f8 --- /dev/null +++ b/src/orchestrator/sweep-budget.ts @@ -0,0 +1,232 @@ +/** + * One aggregate deadline for one discovery sweep. + * + * ## Why a per-call bound is not enough + * + * Three separate unbounded dependency calls have wedged this sweep in a single + * day, each one real, each one bounded, and each time the wedge came back one + * layer down: + * + * - the FACTORY_STATE Durable Object calls (factory-cloud#78), + * - the relayfile change-feed tail reads (#368, shipped in 0.1.75), + * - the retry of the now-bounded call, which is itself unbounded. + * + * The pattern is structural, not incidental. A per-call deadline bounds ONE + * await; a sweep is thousands of awaits plus every retry loop between them, so + * "every call is bounded" never adds up to "the sweep is bounded". The + * deployed 0.1.75 proves it: `readinessReconcile.state` reached `retrying`, + * which means #368's deadline genuinely fires, and `inFlightMs` still climbed + * with wall clock to 31 minutes on a 60-second interval. + * + * This module bounds the total instead. Time is charged against ONE timer for + * the whole sweep, so it does not matter which await is slow, how many there + * are, or how many times the sweep retries one of them: the sum cannot exceed + * the budget. The next unbounded call is then a degraded sweep rather than the + * end of dispatch. + * + * ## What it can and cannot interrupt + * + * `run()` is a race, not a cancellation — the same honest limitation #368 + * documented for `withRelayfileCallDeadline`, and stated here for the same + * reason: the next lane over has to be able to reason about what is left. + * + * It CAN abandon an in-flight await. Expiry rejects the sweep from inside + * `#runOnceWithDiscoveryFence`, so the sweep's own teardown runs: the lease is + * released and `#runOnceInFlight` is cleared, which is what lets the next cycle + * start a fresh sweep instead of coalescing onto the wedged one. That is the + * whole difference from the #296 sweep deadline, which rejected the *caller's + * wait* and left `runOnce()` running for every later cycle to coalesce onto. + * + * It CANNOT stop the abandoned work. The socket stays open, the SDK's own + * retry loop keeps running, and any side effect already in flight still lands. + * Two things narrow that: + * + * - `signal` aborts at expiry, so anything downstream that honours an + * `AbortSignal` is really cancelled. Nothing in the sweep consumes it yet + * (the relayfile client mints its own per-call signal, and that file is + * owned by another lane this week) — it is exported so the wiring is a + * one-line change rather than a redesign. + * - `assertNotExpired()` is a between-await check, which is worth exactly + * what #368 said it was worth — nothing against a call that never returns + * — but it does make an abandoned pass unwind at its next loop iteration + * if it ever regains control, instead of running to completion beside the + * sweep that replaced it. + */ + +/** The phases a sweep can be abandoned in. A closed set: see the error below. */ +export type DiscoverySweepPhase = + | 'fleet-control-plane-probe' + | 'discovery-lease-claim' + | 'discovery-backoff-wait' + | 'discovery-session' + | 'run-once' + | 'discovery-checkpoint' + | 'discovery-renewal-stop' + | 'discovery-commit' + +/** + * A sweep that did not finish inside its aggregate budget. + * + * Built only from code-controlled values — one integer and one literal from the + * closed set above — because it is persisted verbatim into the operator-facing + * `readinessReconcile.lastError`, and the class name reaches the + * unauthenticated health surface through the `error-class` allowlist. + * + * The phase is the diagnostic that matters: it names the await the sweep was + * abandoned on, which is the one thing no per-call bound could report once the + * sweep was already wedged. + */ +export class DiscoverySweepBudgetExceededError extends Error { + readonly code = 'FACTORY_DISCOVERY_SWEEP_BUDGET_EXCEEDED' + + constructor( + readonly budgetMs: number, + readonly phase: DiscoverySweepPhase, + ) { + super(`discovery sweep exceeded its ${budgetMs}ms budget while waiting on ${phase}`) + this.name = 'DiscoverySweepBudgetExceededError' + } +} + +/** The budget to apply, or `undefined` when the caller configured none. */ +export const discoverySweepBudgetMs = (timeoutMs: number | undefined): number | undefined => + timeoutMs !== undefined && Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : undefined + +export interface DiscoverySweepBudget { + /** The applied budget, or `undefined` when the sweep is deliberately unbounded. */ + readonly budgetMs?: number + /** Aborted the moment the budget is spent. Reason is the error below. */ + readonly signal: AbortSignal + /** True once the budget is spent, whether or not anything is awaiting. */ + expired(): boolean + /** Await `start()`, or abandon that wait once the sweep's budget is spent. */ + run(phase: DiscoverySweepPhase, start: () => Promise): Promise + /** Throw if the budget is already spent. A between-await check; see above. */ + assertNotExpired(phase: DiscoverySweepPhase): void + /** Releases the timer. Always call it, or a settled sweep leaves one pending. */ + dispose(): void +} + +const BUDGET_EXPIRED = Symbol('discovery-sweep-budget-expired') + +type SweepOutcome = { ok: true; value: T } | { ok: false; error: unknown } + +/** + * Start the clock for one sweep. + * + * Deliberately not `AbortSignal.timeout()`: its timer cannot be cancelled, so a + * sweep that finishes in four seconds would leave a ninety-minute timer and a + * retained signal behind it. `dispose()` clears the timer the moment the sweep + * settles. + * + * One timer and one expiry promise are shared by every `run()` on this budget. + * That sharing is the aggregate property: a sweep does not get a fresh budget + * per await, so elapsed time accumulates across phases and across retries. + */ +export function startDiscoverySweepBudget(timeoutMs: number | undefined): DiscoverySweepBudget { + const budgetMs = discoverySweepBudgetMs(timeoutMs) + const controller = new AbortController() + let expired = false + let timer: ReturnType | undefined + let expire: () => void = () => undefined + // Never rejects, so an unawaited race arm cannot surface as an unhandled + // rejection when the sweep finishes first. + const expiry = new Promise((resolve) => { + expire = () => resolve(BUDGET_EXPIRED) + }) + + if (budgetMs !== undefined) { + timer = setTimeout(() => { + expired = true + // `run()` supplies the phase on the throw; the abort reason cannot know + // which await was in flight, so it names the sweep as a whole. + controller.abort(new DiscoverySweepBudgetExceededError(budgetMs, 'run-once')) + expire() + }, budgetMs) + timer.unref?.() + } + + const assertNotExpired = (phase: DiscoverySweepPhase): void => { + if (expired && budgetMs !== undefined) throw new DiscoverySweepBudgetExceededError(budgetMs, phase) + } + + return { + ...(budgetMs === undefined ? {} : { budgetMs }), + signal: controller.signal, + expired: () => expired, + assertNotExpired, + async run(phase: DiscoverySweepPhase, start: () => Promise): Promise { + if (budgetMs === undefined) return await start() + // Decided before the call is made, so an already-spent budget cannot + // start new work against the dependency it is abandoning. + assertNotExpired(phase) + // The outcome is folded once. A late rejection from the abandoned call + // then has a handler attached and cannot crash the process. + const inFlight: Promise> = start().then( + (value) => ({ ok: true, value }) as const, + (error: unknown) => ({ ok: false, error }) as const, + ) + const outcome = await Promise.race | typeof BUDGET_EXPIRED>([inFlight, expiry]) + if (outcome === BUDGET_EXPIRED) throw new DiscoverySweepBudgetExceededError(budgetMs, phase) + if (outcome.ok) return outcome.value + throw outcome.error + }, + dispose: () => { + if (timer) clearTimeout(timer) + }, + } +} + +/** + * How long one sweep-teardown step may take. + * + * Teardown cannot run under the aggregate budget: by the time it runs the + * budget is spent by construction, so every step would reject and the lease + * would never be released — which is the half of the fix that makes the next + * cycle clean. It gets its own, much shorter, deadline instead, because an + * unbounded release call would re-create the very wedge this module exists to + * end, one layer further down. + */ +export const DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS = 30_000 + +const TEARDOWN_TIMED_OUT = Symbol('discovery-sweep-teardown-timed-out') + +/** + * Run one teardown step under its own deadline. + * + * Returns `undefined` if the step did not finish in time; the caller decides + * what an abandoned teardown means. Errors propagate exactly as they did + * before this wrapper existed — a teardown deadline is not a licence to start + * swallowing failures that used to surface. + * + * An abandoned lease release is survivable and is why this returns rather than + * throws: the durable lease carries its own expiry, so it is reclaimed as an + * orphan by a later sweep instead of being held forever. + */ +export async function withSweepTeardownDeadline( + timeoutMs: number, + step: () => Promise, +): Promise { + const budget = discoverySweepBudgetMs(timeoutMs) + if (budget === undefined) return await step() + + let timer: ReturnType | undefined + try { + const inFlight: Promise> = step().then( + (value) => ({ ok: true, value }) as const, + (error: unknown) => ({ ok: false, error }) as const, + ) + const outcome = await Promise.race | typeof TEARDOWN_TIMED_OUT>([ + inFlight, + new Promise((resolve) => { + timer = setTimeout(() => resolve(TEARDOWN_TIMED_OUT), budget) + timer.unref?.() + }), + ]) + if (outcome === TEARDOWN_TIMED_OUT) return undefined + if (outcome.ok) return outcome.value + throw outcome.error + } finally { + if (timer) clearTimeout(timer) + } +} diff --git a/src/types.ts b/src/types.ts index f7ef0659..b36b7e7d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -134,6 +134,13 @@ export interface FactoryLiveSubscriptionOptions { * which no deadline checked *between* awaits can reach. */ relayfileOperationTimeoutMs: number + /** + * Aggregate budget for one whole sweep (#372). Bounds what neither neighbour + * can: a bounded call wrapped in an unbounded retry, or a hang on a transport + * nobody has bounded yet. Expiry aborts the sweep from inside its own fence, + * so the discovery lease is released and the next cycle starts clean. + */ + sweepBudgetMs: number } /** From 5c656f98e553ce9fd29a6697cb07b61360d238a0 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 10:53:37 +0200 Subject: [PATCH 2/6] test(orchestrator): point the abandoned-wait suite at the disabled-budget backstop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The seven `bounded readiness reconciliation` cases assert on a sweep that the readiness deadline abandoned and that is STILL RUNNING. The aggregate budget makes that state unreachable at its default — it aborts the sweep at or before that deadline, so there is nothing left in flight to observe. That is the fix, not a regression. Each now passes `sweepBudgetMs: 0`, which selects the pre-#372 backstop those assertions are actually about: the #296/#301 abandoned-wait accounting, still the behaviour when the budget is disabled and still the shape a sweep degrades to if a teardown path cannot be abandoned. `0` as the disable value is the same control idiom #368 used for `operationTimeoutMs`. Adds the positive counterpart, which the redirected cases can no longer state: a live daemon whose first post-claim call never returns still completes `start()` and `stop()`, because the sweep is aborted rather than abandoned. Fail-first verified by mechanism against `origin/main`'s factory.ts: it fails after 4044 ms with `start never returned`. That is deliverable B demonstrated rather than argued — `#deferLiveEventDrain = false` sits in a `finally` around that unbounded `runOnce()` (main factory.ts:1983/2021/2039), so a wedged startup backfill also kills the live-event dispatch path, which is why a hung sweep meant zero dispatch instead of stale dispatch. Also documents a gap the shutdown test exposed and this PR does NOT close: `#startLiveSubscription` reads the event high-watermark before the backfill and outside any sweep, so that read is bounded only by the per-call relayfile deadline. Co-Authored-By: Claude Opus 5 Session-Id: b1177efc-90da-4ff1-bda0-ef5de1b475e2 --- src/orchestrator/factory.test.ts | 28 ++++++++++++----- src/orchestrator/sweep-budget.test.ts | 45 ++++++++++++++++++++++++++- 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 97d3c411..f54c7113 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -15523,6 +15523,20 @@ describe('FactoryLoop', () => { // took neither: the subsystem reported `healthy` with zero failures while // dispatching nothing, and only a process restart recovered it. describe('bounded readiness reconciliation', () => { + /** + * Selects the pre-#372 backstop these assertions are about. + * + * The aggregate sweep budget aborts a wedged sweep at or before the + * readiness deadline, so at its default there IS no abandoned-but-still- + * running sweep left for this block to observe — which is the fix, covered + * in `sweep-budget.test.ts`. What survives underneath it is what this block + * has always tested: the #296/#301 abandoned-wait accounting, which is + * still the behaviour when the budget is disabled, and still the shape a + * sweep degrades to if a future teardown path cannot be abandoned. `0` + * disables the budget, the same idiom #368 used for its own control. + */ + const NO_SWEEP_BUDGET = { sweepBudgetMs: 0 } as const + class HangingDiscoveryStateStore extends InMemoryStateStore { hangClaims = false readonly hangStarted: Promise @@ -15576,7 +15590,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16134,7 +16148,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) let stopped = false let stopping: Promise | undefined @@ -16182,7 +16196,7 @@ describe('FactoryLoop', () => { // possible by carrying the abandoned sweep's own start time. await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 1_000 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 1_000, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16221,7 +16235,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 400 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 400, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16262,7 +16276,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16307,7 +16321,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) let stopped = false let stopping: Promise | undefined @@ -16370,7 +16384,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 100, reconcileTimeoutMs: 200 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 100, reconcileTimeoutMs: 200, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts index 9ba2e91b..93bfbf96 100644 --- a/src/orchestrator/sweep-budget.test.ts +++ b/src/orchestrator/sweep-budget.test.ts @@ -276,13 +276,24 @@ class StaticTriage implements TriageEngine { */ class HangingWatermarkMount extends FakeMountClient { hang = true + /** + * How many watermark reads to serve before hanging. + * + * `#startLiveSubscription` reads the watermark once itself, BEFORE the + * startup backfill and outside the sweep — that read is bounded by the + * per-call relayfile deadline, not by this budget. Serving it is what puts + * the hang inside the sweep, which is the subject here. + */ + serveFirst = 0 + served = 0 hungCalls = 0 override async getEventHighWatermark(opts: { provider?: string } = {}): Promise { - if (this.hang) { + if (this.hang && this.served >= this.serveFirst) { this.hungCalls += 1 return await NEVER() } + this.served += 1 return await super.getEventHighWatermark(opts) } } @@ -344,6 +355,38 @@ describe('a wedged sweep is bounded end to end (#372)', () => { } }) + it('must-fire: a live daemon whose sweep is wedged still shuts down, because nothing is left in flight', async () => { + // The counterpart to the #296/#301 abandoned-wait tests, which assert that + // `stop()` must NOT complete while a sweep abandoned by the readiness + // deadline is still running. That state is real, and it is exactly what + // held the deployed daemon: the sweep outlives every wait on it. With the + // budget it cannot happen — the sweep is aborted, not abandoned — so + // shutdown has nothing to drain. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-stop-')) + const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) + mount.serveFirst = 1 + const factory = createFactory(config(200, root), { + mount, + fleet: new FakeFleetClient(), + stateStore: new LeaseWatchingStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + // The startup backfill is a discovery pass like any other and wedges on + // the same call. Before this change `start()` itself never returns. + await withDeadline(factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }), 4_000, 'start never returned') + expect(mount.hungCalls).toBeGreaterThan(0) + await withDeadline(factory.stop(), 4_000, 'stop never completed') + } finally { + await factory.stop().catch(() => undefined) + await rm(root, { recursive: true, force: true }) + } + }) + it('must-not-fire: a sweep that completes inside its budget is not aborted and its result is unchanged', async () => { // Without this the trivial wrong fix — abort every sweep immediately — // passes the must-fire above. From 2c2dd86eecbe23d7487ca68243dd4a98e95cdbf4 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 11:19:03 +0200 Subject: [PATCH 3/6] =?UTF-8?q?fix(orchestrator):=20answer=20the=20review?= =?UTF-8?q?=20=E2=80=94=20derive=20the=20default,=20fence=20the=20abandone?= =?UTF-8?q?d=20pass?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five findings, all valid at 5c656f98, each with its own must-fire/must-not-fire. 1. THE DANGEROUS ONE. A fixed 90-minute `sweepBudgetMs` default is ABOVE any config that had already tightened `reconcileTimeoutMs`, so the cross-field check rejected it and `FactoryConfigSchema.parse` threw — Factory would not have started. It also silently capped a config that loosened the timeout above 90 minutes. The omitted budget is now derived from its SIBLING in a `.transform()`, never from a constant, and `resolvedSweepBudgetMs` is the one rule the schema and the orchestrator's `start()` clamp both use. 2. An abandoned `#performRunOnce` could write a stale tree listing into the REPLACEMENT sweep's checkpoint: `#rememberDiscoveryTree` reads the shared `#discoverySession` fresh, and by the time a late continuation resolves that is the next sweep's. `#isStaleDiscoveryContinuation()` compares the `discoveryEnumerationPass` epoch — an AsyncLocalStorage store, so it follows the async continuation and still carries the epoch that ISSUED the read — against the live one. It is the same fence the tree-read counters already used. Applied to the checkpoint write and to overload attribution, so a 429 that arrives after its sweep was abandoned cannot drive the replacement's ratchet. 3. The dispatch loop gets the same budget guard as the read loop, so a pass abandoned during enumeration cannot dispatch after its lease went back. 4. A lease claimed after the budget gave up on the claim was stranded: nobody would renew, commit or release it, so every later sweep deferred for a whole lease window. A compensating release is now attached to the abandoned claim. Fail-first verified by mechanism — with the compensation ablated the test fails with `stranded lease was never released`. 5. Unref'd deadline timers let Node exit before the budget fires. Under a one-shot `runOnce()` whose only pending work is a promise nothing else references, the command would return without reporting the wedge or releasing the lease. Both deadline timers are referenced now; they live for at most one budget and `dispose()` clears them from a `finally`. Also, on the review's reading of a comment: the pre-backfill watermark read is bounded neither by the sweep budget NOR by anything in the orchestrator — `#currentEventHighWatermark` (factory.ts:2192) awaits the mount directly under a bare try/catch. What bounds it in production is one layer lower, the deployed client's own `#bounded()` (relayfile-cloud-mount-client.ts:1057, #368). The comment now says which layer, because a `MountClient` without that deadline has no bound here at all. And the e2e must-fire no longer risks blaming a phase string for a timing stall: the budget has 400 ms of headroom over two in-memory calls, and `hungCalls` is asserted before the phase so a mis-timed run names the real cause. Co-Authored-By: Claude Opus 5 Session-Id: b1177efc-90da-4ff1-bda0-ef5de1b475e2 --- src/config/schema.ts | 40 +++++-- src/orchestrator/factory.ts | 100 +++++++++++++++--- src/orchestrator/sweep-budget.test.ts | 146 +++++++++++++++++++++++--- src/orchestrator/sweep-budget.ts | 12 ++- 4 files changed, 260 insertions(+), 38 deletions(-) diff --git a/src/config/schema.ts b/src/config/schema.ts index c74c3ed1..6e3370d9 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -62,11 +62,13 @@ const subscriptionSchema = z.object({ export const DEFAULT_READINESS_RECONCILE_TIMEOUT_MS = 90 * 60_000 /** - * Aggregate budget for one discovery sweep (#372). + * Aggregate budget for one discovery sweep (#372), when nothing narrows it. * - * Defaults to `DEFAULT_READINESS_RECONCILE_TIMEOUT_MS` on purpose. The two - * numbers describe the same envelope; what changes is the MECHANISM, and the - * mechanism is the deliverable. `reconcileTimeoutMs` rejects the caller's wait + * Equal to `DEFAULT_READINESS_RECONCILE_TIMEOUT_MS` on purpose — and a config + * that omits `sweepBudgetMs` tracks whatever `reconcileTimeoutMs` it set, not + * this constant (see `resolvedSweepBudgetMs`). The two numbers describe the + * same envelope; what changes is the MECHANISM, and the mechanism is the + * deliverable. `reconcileTimeoutMs` rejects the caller's wait * and leaves `runOnce()` running, so every later cycle coalesces onto the * wedged pass and the daemon never recovers. The sweep budget rejects from * inside the fence, so the lease is released and the next cycle starts clean. @@ -81,6 +83,17 @@ export const DEFAULT_READINESS_RECONCILE_TIMEOUT_MS = 90 * 60_000 */ export const DEFAULT_DISCOVERY_SWEEP_BUDGET_MS = DEFAULT_READINESS_RECONCILE_TIMEOUT_MS +/** + * The effective budget for a config, given what it did or did not set. + * + * Exported so the orchestrator's field initializer and the schema agree on one + * rule rather than two that happen to match today. + */ +export const resolvedSweepBudgetMs = ( + sweepBudgetMs: number | undefined, + reconcileTimeoutMs: number, +): number => Math.min(sweepBudgetMs ?? reconcileTimeoutMs, reconcileTimeoutMs) + const liveSubscriptionSchema = z.object({ transport: z.enum(['subscribe-and-poll', 'subscribe', 'poll']).default('subscribe-and-poll'), pollIntervalMs: z.number().int().min(50).default(5_000), @@ -115,8 +128,7 @@ const liveSubscriptionSchema = z.object({ * the entire pass, so a sweep cannot outlive it however many calls, retries * or transports it is spread across. */ - sweepBudgetMs: z.number().int().min(50).max(6 * 60 * 60_000) - .default(DEFAULT_DISCOVERY_SWEEP_BUDGET_MS), + sweepBudgetMs: z.number().int().min(50).max(6 * 60 * 60_000).optional(), }).superRefine((value, ctx) => { // A deadline below the interval kills every pass that takes longer than one // tick, which is most of them on a cold mirror. @@ -129,15 +141,25 @@ const liveSubscriptionSchema = z.object({ } // The sweep budget has to be the tighter of the two, or the wait gives up // first and the sweep it abandoned keeps running for the next cycle to - // coalesce onto — the exact behaviour the budget exists to remove. - if (value.sweepBudgetMs > value.reconcileTimeoutMs) { + // coalesce onto — the exact behaviour the budget exists to remove. Checked + // only when it was set explicitly: an omitted one is derived below and + // cannot violate this. + if (value.sweepBudgetMs !== undefined && value.sweepBudgetMs > value.reconcileTimeoutMs) { ctx.addIssue({ code: z.ZodIssueCode.custom, path: ['sweepBudgetMs'], message: `sweepBudgetMs (${value.sweepBudgetMs}) must not exceed reconcileTimeoutMs (${value.reconcileTimeoutMs})`, }) } -}).default({}) +}).transform((value) => ({ + ...value, + // Derived from the SIBLING, never from a constant. A fixed 90-minute default + // would reject every config that already tightened `reconcileTimeoutMs` + // below it — the schema throws, so Factory would not start — and would + // silently cap every config that loosened it above. "Omitted" means "the + // same envelope as the wait", whatever that wait is configured to be. + sweepBudgetMs: value.sweepBudgetMs ?? value.reconcileTimeoutMs, +})).default({}) export const DEFAULT_AGENT_HOLD_TIMEOUT_MS = 4 * 60 * 60_000 diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 53b0f551..b672e5ec 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -7,6 +7,7 @@ import { DEFAULT_DISCOVERY_SWEEP_BUDGET_MS, DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, FactoryConfigSchema, + resolvedSweepBudgetMs, type FactoryConfig, } from '../config/schema' import { @@ -1148,7 +1149,7 @@ export class FactoryLoop implements Factory { // Also read here, not only in `#startLiveSubscription`: a standalone // `runOnce()` never starts the live subscription and must still be bounded. this.#relayfileOperationTimeoutMs = config.liveSubscription.relayfileOperationTimeoutMs - this.#discoverySweepBudgetMs = Math.min( + this.#discoverySweepBudgetMs = resolvedSweepBudgetMs( config.liveSubscription.sweepBudgetMs, config.liveSubscription.reconcileTimeoutMs, ) @@ -1997,7 +1998,10 @@ export class FactoryLoop implements Factory { // Same re-application as the deadline above: `start()` overrides bypass the // schema's cross-field check, and a budget looser than the wait is not a // budget. - this.#discoverySweepBudgetMs = Math.min(options.sweepBudgetMs, this.#readinessReconcileTimeoutMs) + this.#discoverySweepBudgetMs = resolvedSweepBudgetMs( + options.sweepBudgetMs, + this.#readinessReconcileTimeoutMs, + ) this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -3103,6 +3107,49 @@ export class FactoryLoop implements Factory { } } + /** + * Claim the sweep lease under the budget, compensating for a claim that + * lands after we stopped waiting for it. + * + * The budget abandons the wait, not the call, so the store can still persist + * a lease for a claim this sweep has already given up on — and that lease + * would be held by an owner that will never renew, commit or release it, so + * every later sweep defers until it expires. It is self-expiring and a later + * sweep reclaims it as an orphan, which makes this a latency fix rather than + * a correctness one; the latency is one whole lease window with no discovery, + * which is the thing this PR exists to stop paying. + */ + async #claimDiscoverySweepUnderBudget(budget: DiscoverySweepBudget): Promise { + const claim = this.#state.claimDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + this.#clock.now(), + DISCOVERY_SWEEP_LEASE_MS, + ) + try { + return await budget.run('discovery-lease-claim', () => claim) + } catch (error) { + if (!(error instanceof DiscoverySweepBudgetExceededError)) throw error + void claim.then( + async (late) => { + if (!late.acquired || !late.lease) return + this.#increment('discoverySweepStrandedClaimsReleased') + this.#logger.warn?.('[factory] releasing a discovery lease that was claimed after the sweep budget expired', { + epoch: late.lease.epoch, + budgetMs: error.budgetMs, + }) + await this.#sweepTeardownStep('stranded discovery lease release', () => this.#state.releaseDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + late.lease!.epoch, + )) + }, + () => undefined, + ).catch(() => undefined) + throw error + } + } + async #runDiscoverySweep( opts: { dryRun?: boolean }, budget: DiscoverySweepBudget, @@ -3114,12 +3161,7 @@ export class FactoryLoop implements Factory { // #368 covers. The budget does not care which one it is. await budget.run('fleet-control-plane-probe', () => this.#assertFleetControlPlaneAvailable()) } - let claim = await budget.run('discovery-lease-claim', () => this.#state.claimDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - this.#clock.now(), - DISCOVERY_SWEEP_LEASE_MS, - )) + let claim = await this.#claimDiscoverySweepUnderBudget(budget) if (!claim.acquired && claim.reason === 'backoff') { const delayMs = Math.max(0, claim.state.backoffUntilMs - this.#clock.now()) this.#increment('discoveryBackoffWaits') @@ -3129,12 +3171,7 @@ export class FactoryLoop implements Factory { consecutiveOverloads: claim.state.consecutiveOverloads, }) await budget.run('discovery-backoff-wait', () => this.#clock.sleep(delayMs)) - claim = await budget.run('discovery-lease-claim', () => this.#state.claimDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - this.#clock.now(), - DISCOVERY_SWEEP_LEASE_MS, - )) + claim = await this.#claimDiscoverySweepUnderBudget(budget) } if (!claim.acquired || !claim.lease) { this.#increment('discoverySweepsSkippedInFlight') @@ -3583,6 +3620,10 @@ export class FactoryLoop implements Factory { } for (const { issue } of issueEntries) { + // The dispatch half of the same fence as the read loop above. Without + // it a pass abandoned during enumeration would go on to dispatch after + // its lease had been handed back, racing the sweep that replaced it. + budget?.assertNotExpired('run-once') await this.#refreshLiveHeartbeatIfDue() if (!issue) { continue @@ -4840,9 +4881,35 @@ export class FactoryLoop implements Factory { return paths ? [...paths] : undefined } + /** + * True when this call is a continuation of a sweep that has already ended. + * + * #372: the aggregate budget abandons the WAIT, not the call, so a read + * issued by an aborted sweep can still resolve — by which time the shared + * `#discoverySession` and `#discoverySweepEpoch` may belong to the sweep + * that replaced it. `discoveryEnumerationPass` is an `AsyncLocalStorage`, so + * its store follows the async continuation and still carries the epoch that + * ISSUED the read; comparing the two is what tells a live write from a + * late one. + * + * Only when a store exists: a caller outside a discovery pass legitimately + * has none, and treating that as stale would silence the live event drain. + */ + #isStaleDiscoveryContinuation(): boolean { + const issuingPass = discoveryEnumerationPass.getStore() + return issuingPass !== undefined && issuingPass.epoch !== this.#discoverySweepEpoch + } + async #rememberDiscoveryTree(prefix: string, paths: string[]): Promise { const session = this.#discoverySession if (!session) return + // Committing this listing would put a tree from an abandoned pass into the + // replacement sweep's checkpoint, under a watermark that claims to describe + // the replacement. That is checkpoint corruption, and it outlives the sweep. + if (this.#isStaleDiscoveryContinuation()) { + this.#increment('discoveryStaleTreeWritesDropped') + return + } const uniquePaths = new Set() for (let index = 0; index < paths.length; index += 1) { uniquePaths.add(paths[index]!) @@ -4938,7 +5005,10 @@ export class FactoryLoop implements Factory { return result } catch (error) { const overload = relayfileOverload(error) - if (overload && this.#discoverySweepEpoch !== undefined) { + // The stale check for the same reason as the tree write above: a 429 that + // arrives after its sweep was abandoned is not this sweep's evidence, and + // attributing it here would drive the replacement's overload ratchet. + if (overload && this.#discoverySweepEpoch !== undefined && !this.#isStaleDiscoveryContinuation()) { this.#discoveryOverloadError ??= error this.#discoverySweepOverloads += 1 if (overload.retryAfterSeconds !== undefined) { diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts index 93bfbf96..d6dfbba5 100644 --- a/src/orchestrator/sweep-budget.test.ts +++ b/src/orchestrator/sweep-budget.test.ts @@ -3,6 +3,11 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { + DEFAULT_DISCOVERY_SWEEP_BUDGET_MS, + DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, + resolvedSweepBudgetMs, +} from '../config/schema' import { FactoryConfigSchema, createFactory, @@ -14,6 +19,7 @@ import { import { FakeFleetClient, FakeMountClient } from '../testing' import { withDeadline } from '../testing/deadline' import { InMemoryStateStore } from '../state/in-memory-state-store' +import type { DiscoverySweepClaim } from '../ports/state' import { DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, DiscoverySweepBudgetExceededError, @@ -184,6 +190,50 @@ describe('sweep teardown deadline', () => { }) }) +describe('the configured budget', () => { + const live = (liveSubscription: Record) => FactoryConfigSchema.parse({ + workspaceId: 'factory-sweep-budget-config', + repos: { + byLabel: { pear: 'AgentWorkforce/pear' }, + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + stateIds: { readyForAgent: ready, agentImplementing: implementing, done, inPlanning: planning }, + liveSubscription, + }).liveSubscription + + it('must-not-fire: a config that already tightened reconcileTimeoutMs still parses', () => { + // The regression that would have taken Factory down on deploy rather than + // fixing it: a fixed 90-minute default is ABOVE such a config's timeout, so + // the cross-field check rejects it and `FactoryConfigSchema.parse` throws + // before the daemon can start. + expect(live({ reconcileTimeoutMs: 5 * 60_000 }).sweepBudgetMs).toBe(5 * 60_000) + }) + + it('must-not-fire: a config that loosened reconcileTimeoutMs is not silently capped at 90 minutes', () => { + expect(live({ reconcileTimeoutMs: 3 * 60 * 60_000 }).sweepBudgetMs).toBe(3 * 60 * 60_000) + }) + + it('must-not-fire: an omitted budget tracks the default timeout', () => { + expect(live({}).sweepBudgetMs).toBe(DEFAULT_DISCOVERY_SWEEP_BUDGET_MS) + expect(DEFAULT_DISCOVERY_SWEEP_BUDGET_MS).toBe(DEFAULT_READINESS_RECONCILE_TIMEOUT_MS) + }) + + it('must-fire: an explicit budget looser than the wait is still rejected', () => { + // A budget the wait gives up on first is not a budget: the sweep it + // abandoned keeps running and the next cycle coalesces onto it. + expect(() => live({ reconcileTimeoutMs: 60_000, sweepBudgetMs: 120_000 })).toThrow(/sweepBudgetMs/) + expect(live({ reconcileTimeoutMs: 120_000, sweepBudgetMs: 60_000 }).sweepBudgetMs).toBe(60_000) + }) + + it('must-not-fire: resolving is the same rule on both sides of the schema', () => { + expect(resolvedSweepBudgetMs(undefined, 5 * 60_000)).toBe(5 * 60_000) + expect(resolvedSweepBudgetMs(60_000, 5 * 60_000)).toBe(60_000) + // `start()` overrides bypass the schema, so the clamp has to hold here too. + expect(resolvedSweepBudgetMs(10 * 60_000, 5 * 60_000)).toBe(5 * 60_000) + }) +}) + /* ------------------------------------------------------------------------- */ /* The same invariant, driven through a real sweep. */ /* ------------------------------------------------------------------------- */ @@ -280,9 +330,15 @@ class HangingWatermarkMount extends FakeMountClient { * How many watermark reads to serve before hanging. * * `#startLiveSubscription` reads the watermark once itself, BEFORE the - * startup backfill and outside the sweep — that read is bounded by the - * per-call relayfile deadline, not by this budget. Serving it is what puts - * the hang inside the sweep, which is the subject here. + * startup backfill and outside the sweep, so no sweep budget covers it. + * Nothing in the orchestrator bounds it either: `#currentEventHighWatermark` + * (factory.ts:2192) awaits `mount.getEventHighWatermark()` under a bare + * try/catch, not through `#withRelayfileOperation`. What bounds it in + * production is one layer lower — the deployed client's own `#bounded()` + * (relayfile-cloud-mount-client.ts:1057, #368), fed from + * `relayfileOperationTimeoutMs` at cli/fleet.ts:2215 — so a `MountClient` + * without that deadline has none here at all. Serving the read is what puts + * the hang inside the sweep, which is the subject of this file. */ serveFirst = 0 served = 0 @@ -301,6 +357,20 @@ class HangingWatermarkMount extends FakeMountClient { /** A store that records the lease handbacks, so "released" is observed, not inferred. */ class LeaseWatchingStateStore extends InMemoryStateStore { readonly released: number[] = [] + /** Milliseconds a claim takes to answer. Long enough and the budget gives up first. */ + claimDelayMs = 0 + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + if (this.claimDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, this.claimDelayMs)) + } + return await super.claimDiscoverySweep(workspaceId, owner, nowMs, leaseMs) + } override async releaseDiscoverySweep(workspaceId: string, owner: string, epoch: number): Promise { this.released.push(epoch) @@ -316,7 +386,7 @@ describe('a wedged sweep is bounded end to end (#372)', () => { }) const fleet = new FakeFleetClient() const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) - const factory = createFactory(config(150, root), { + const factory = createFactory(config(400, root), { mount, fleet, stateStore, @@ -328,15 +398,22 @@ describe('a wedged sweep is bounded end to end (#372)', () => { // BEFORE this change this call never settles and the test dies at // vitest's 5 s default — the production defect, not an assertion detail. // The 4 s guard is inside that default so the failure names itself. - await expect(withDeadline(factory.runOnce(), 4_000, 'sweep never settled')) - .rejects.toMatchObject({ - name: 'DiscoverySweepBudgetExceededError', - // The abandoned await is named, which is the diagnostic no per-call - // bound can produce once the sweep is already wedged. - phase: 'discovery-session', - budgetMs: 150, - }) + const aborted = await withDeadline( + factory.runOnce().then(() => undefined, (error: unknown) => error), + 4_000, + 'sweep never settled', + ) + expect(aborted).toMatchObject({ name: 'DiscoverySweepBudgetExceededError', budgetMs: 400 }) + // Asserted BEFORE the phase, so a mis-timed run says which sweep phase + // actually ran out rather than blaming a string. One hung call is the + // proof that the budget reached `#prepareDiscoverySession`; the two + // phases ahead of it here are a fake fleet roster and an in-memory lease + // claim, and 400 ms is roughly three orders of magnitude of headroom + // over both. expect(mount.hungCalls).toBe(1) + // The abandoned await is named, which is the diagnostic no per-call + // bound can produce once the sweep is already wedged. + expect((aborted as { phase?: string }).phase).toBe('discovery-session') // The lease is handed back on the way out. Without this the next cycle // has nothing to claim and the wedge simply moves. expect(stateStore.released).toHaveLength(1) @@ -387,6 +464,45 @@ describe('a wedged sweep is bounded end to end (#372)', () => { } }) + it('must-fire: a lease that lands after the budget gave up on the claim is handed straight back', async () => { + // The budget abandons the WAIT, not the call, so the store can persist a + // lease for a claim this sweep has already left. Nobody would renew, + // commit or release it, so every later sweep would defer for a full lease + // window — the wedge this PR exists to stop, one layer down. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-claim-')) + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + stateStore.claimDelayMs = 600 + const factory = createFactory(config(150, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901) }), + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + try { + await expect(withDeadline(factory.runOnce(), 4_000, 'sweep never settled')) + .rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + phase: 'discovery-lease-claim', + }) + // The claim was still in flight when the budget expired; the compensation + // fires when it lands. + await withDeadline( + (async () => { + while (stateStore.released.length === 0) { + await new Promise((resolve) => setTimeout(resolve, 25)) + } + })(), + 4_000, + 'stranded lease was never released', + ) + expect(stateStore.released).toHaveLength(1) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it('must-not-fire: a sweep that completes inside its budget is not aborted and its result is unchanged', async () => { // Without this the trivial wrong fix — abort every sweep immediately — // passes the must-fire above. @@ -407,6 +523,11 @@ describe('a wedged sweep is bounded end to end (#372)', () => { dispatched: report.dispatched.length, skipped: report.skipped.length, spawns: fleet.spawns.map((spawn) => spawn.name), + // The risk the stale-continuation fence introduces is the opposite of + // the one it removes: a fence that mistook a LIVE write for a late + // one would silently drop this sweep's own tree listings and the + // checkpoint would go empty. Zero, on a sweep that enumerated. + staleTreeWritesDropped: factory.status().counters.discoveryStaleTreeWritesDropped ?? 0, } } finally { await factory.stop() @@ -419,6 +540,7 @@ describe('a wedged sweep is bounded end to end (#372)', () => { const budgeted = await run(30_000) const control = await run(90 * 60_000) expect(budgeted.dispatched).toBeGreaterThan(0) + expect(budgeted.staleTreeWritesDropped).toBe(0) expect(budgeted).toEqual(control) }) }) diff --git a/src/orchestrator/sweep-budget.ts b/src/orchestrator/sweep-budget.ts index bf7ae0f8..75b17ab1 100644 --- a/src/orchestrator/sweep-budget.ts +++ b/src/orchestrator/sweep-budget.ts @@ -143,7 +143,14 @@ export function startDiscoverySweepBudget(timeoutMs: number | undefined): Discov controller.abort(new DiscoverySweepBudgetExceededError(budgetMs, 'run-once')) expire() }, budgetMs) - timer.unref?.() + // Deliberately NOT unref'd, unlike the per-call deadline in + // `relayfile-operation-timeout.ts`. That one is created thousands of times + // per sweep and can afford to lose a race with process exit; this one is + // the guarantee. Under a one-shot `runOnce()` whose only pending work is a + // promise nothing else references, an unref'd timer lets Node exit before + // the budget fires — the command would return without ever reporting the + // wedge or releasing the lease. It lives for at most one budget and + // `dispose()` clears it from a `finally` on every path. } const assertNotExpired = (phase: DiscoverySweepPhase): void => { @@ -219,8 +226,9 @@ export async function withSweepTeardownDeadline( const outcome = await Promise.race | typeof TEARDOWN_TIMED_OUT>([ inFlight, new Promise((resolve) => { + // Referenced, for the same reason as the budget timer above: this is + // the deadline that guarantees the lease is handed back. timer = setTimeout(() => resolve(TEARDOWN_TIMED_OUT), budget) - timer.unref?.() }), ]) if (outcome === TEARDOWN_TIMED_OUT) return undefined From 462aa3e28017798b5c1e6f5e6e38360f58e4b675 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 11:43:42 +0200 Subject: [PATCH 4/6] fix(orchestrator): give shutdown a lever on the sweep budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two findings at 2c2dd86, both this PR's own lesson recurring inside this PR. 1. THE BOUND BECAME THE WEDGE. `stop()` deliberately outlives the sweep it started (#301, the `#readinessReconcileAbandonedWait` drain), so a wedged sweep makes shutdown exactly as long as the sweep budget — 90 minutes at the default. Referencing the timer did not create that (before this PR the drain was unbounded, so shutdown was unbounded too), but it is the same trap the teardown deadline already answers one layer down, and an operator restarting a wedged container is the person who pays. NOT fixed by `unref()`. That is the trivially wrong version: it also lets Node exit before the budget fires, so a one-shot `runOnce()` returns having neither reported the wedge nor released the lease — the P2 that made the timer referenced in the first place. The two asks are in tension and only a shutdown-specific path satisfies both. `stop()` now arms a grace timer over the drain; after `STOP_TEARDOWN_TIMEOUT_MS` it calls `budget.expire()` on every in-flight sweep, routing them into the ordinary abort path — lease released, teardown bounded — instead of holding the process. The grace is what keeps an ordinary restart from discarding a sweep that was about to commit. A sweep that starts while `#stopping` is already set is expired immediately, so it cannot hand shutdown a fresh 90-minute budget. 2. The lease claim is now issued INSIDE the budget callback, so a spent budget rejects the phase without opening a lease it could only hand straight back. A lease taken after expiry makes every later sweep defer — the same "later cycles wait on a pass that is already over" failure this PR's own comparison names in `reconcileTimeoutMs`. Pairs, and their fail-first, verified by ablation: - must-fire: a live daemon whose PERIODIC sweep wedges under a 60 s budget still completes `stop()` inside 4 s. With the grace ablated it hangs to the vitest timeout — shutdown waiting out the budget, which is the defect. - must-not-fire: the budget timer appears in `process.getActiveResourcesInfo()` while a sweep runs and is gone the moment `dispose()` runs. This is what fails for the `unref()` version — that list contains only resources KEEPING THE EVENT LOOP ALIVE, so an unref'd timer never appears — and it also pins the other half: a settled sweep leaves nothing behind, which is what makes a referenced 90-minute timer affordable. - must-fire: a sweep aborted in the fleet-probe phase opens no lease and releases none. Scoped honestly in the test: every entry into `#claimDiscoverySweepUnderBudget` is preceded by a `budget.run` that throws first, so "already spent on entry" is a microtask race rather than a reachable state, and moving the claim inside the callback closes it by construction. The guarantee that does the work — `budget.run` never invokes its thunk once spent — is asserted directly on the primitive. - must-not-fire: a healthy sweep still claims exactly once and dispatches. The trivially wrong way to stop a spent budget claiming is to stop claiming. Co-Authored-By: Claude Opus 5 Session-Id: b1177efc-90da-4ff1-bda0-ef5de1b475e2 --- src/orchestrator/factory.ts | 94 ++++++++++++++---- src/orchestrator/sweep-budget.test.ts | 133 ++++++++++++++++++++++++++ src/orchestrator/sweep-budget.ts | 28 ++++-- 3 files changed, 231 insertions(+), 24 deletions(-) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index b672e5ec..75b3ced9 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -905,6 +905,16 @@ export class FactoryLoop implements Factory { * exists to remove. */ #discoverySweepBudgetMs = DEFAULT_DISCOVERY_SWEEP_BUDGET_MS + /** + * The budgets of every sweep currently in flight. + * + * `stop()` drains the sweep it started (see `#readinessReconcileAbandonedWait` + * below), so a wedged sweep holds shutdown open for the whole budget — 90 + * minutes at the default, and unbounded before this budget existed. A set + * rather than one field because a live sweep and a mismatched dry-run one can + * be in flight at the same time. + */ + readonly #discoverySweepBudgets = new Set() // Set for exactly as long as a sweep is running. `state` is derived from // this, so an in-flight pass can no longer masquerade as the last settled one. #readinessReconcileInFlightSinceMs?: number @@ -1585,15 +1595,24 @@ export class FactoryLoop implements Factory { this.#readinessReconcileTimer = undefined if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) this.#previewSweepTimer = undefined - await this.#readinessReconcileInFlight - // #301 review: the deadline ends the *wait*, so `#readinessReconcileInFlight` - // can settle with its `runOnce()` still live. Shutdown releases dispatch - // lifecycle leases and disposes ports below, and `#isPassFatalFailure` only - // fences a stopping sweep once something in it throws — so a sweep whose - // dependency recovers cleanly would otherwise dispatch through torn-down - // state. Draining here restores exactly the pre-deadline shutdown contract: - // stop() outlives the sweep it started. - await this.#readinessReconcileAbandonedWait + // #372: the drain below is unbounded in the one case that matters — a + // wedged sweep — so shutdown inherits the sweep budget, 90 minutes at the + // default. Spending it after one teardown window bounds shutdown without + // discarding a sweep that was about to finish. + const releaseSweepBudgetGrace = this.#cutSweepBudgetsShortForStop() + try { + await this.#readinessReconcileInFlight + // #301 review: the deadline ends the *wait*, so `#readinessReconcileInFlight` + // can settle with its `runOnce()` still live. Shutdown releases dispatch + // lifecycle leases and disposes ports below, and `#isPassFatalFailure` only + // fences a stopping sweep once something in it throws — so a sweep whose + // dependency recovers cleanly would otherwise dispatch through torn-down + // state. Draining here restores exactly the pre-deadline shutdown contract: + // stop() outlives the sweep it started. + await this.#readinessReconcileAbandonedWait + } finally { + releaseSweepBudgetGrace() + } await this.#previewSweepInFlight this.#stoppingHeartbeatRefreshActive = await this.#stopLiveHeartbeat('stopping') try { @@ -3085,7 +3104,11 @@ export class FactoryLoop implements Factory { */ async #runOnceWithDiscoveryFence(opts: { dryRun?: boolean }): Promise { const budget = startDiscoverySweepBudget(this.#discoverySweepBudgetMs) + this.#discoverySweepBudgets.add(budget) const sweepStartedAtMs = this.#clock.now() + // A sweep that starts while shutdown is already draining would otherwise + // get a full fresh budget to hold `stop()` open with. + if (this.#stopping) budget.expire() try { return await this.#runDiscoverySweep(opts, budget) } catch (error) { @@ -3103,10 +3126,40 @@ export class FactoryLoop implements Factory { } throw error } finally { + this.#discoverySweepBudgets.delete(budget) budget.dispose() } } + /** + * Cut every in-flight sweep's budget short once shutdown starts draining. + * + * `stop()` deliberately outlives the sweep it started (#301), so a wedged + * sweep makes shutdown as long as the budget. The grace window is what keeps + * an ordinary restart from throwing away a sweep that was about to commit; + * after it, expiry routes the sweep into exactly the abort path a real expiry + * takes — lease released, teardown bounded — instead of holding the process. + * + * Returns the cancel for the grace timer. Always call it: on a prompt + * shutdown there is nothing to cut short. + */ + #cutSweepBudgetsShortForStop(): () => void { + if (this.#discoverySweepBudgets.size === 0) return () => undefined + const timer = setTimeout(() => { + for (const budget of this.#discoverySweepBudgets) { + if (budget.expired()) continue + this.#increment('discoverySweepBudgetsCutShortForStop') + this.#logger.warn?.('[factory] shutdown is draining a sweep; spending its budget now', { + graceMs: STOP_TEARDOWN_TIMEOUT_MS, + budgetMs: budget.budgetMs, + }) + budget.expire() + } + }, STOP_TEARDOWN_TIMEOUT_MS) + timer.unref?.() + return () => clearTimeout(timer) + } + /** * Claim the sweep lease under the budget, compensating for a claim that * lands after we stopped waiting for it. @@ -3120,16 +3173,23 @@ export class FactoryLoop implements Factory { * which is the thing this PR exists to stop paying. */ async #claimDiscoverySweepUnderBudget(budget: DiscoverySweepBudget): Promise { - const claim = this.#state.claimDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - this.#clock.now(), - DISCOVERY_SWEEP_LEASE_MS, - ) + // Issued INSIDE the budget callback, so a budget that is already spent + // rejects the phase without opening a lease it could only hand straight + // back. The handle is kept out here because the compensation below needs + // the promise the wait was abandoned on. + let claim: Promise | undefined try { - return await budget.run('discovery-lease-claim', () => claim) + return await budget.run('discovery-lease-claim', () => { + claim = this.#state.claimDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + this.#clock.now(), + DISCOVERY_SWEEP_LEASE_MS, + ) + return claim + }) } catch (error) { - if (!(error instanceof DiscoverySweepBudgetExceededError)) throw error + if (!(error instanceof DiscoverySweepBudgetExceededError) || claim === undefined) throw error void claim.then( async (late) => { if (!late.acquired || !late.lease) return diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts index d6dfbba5..04d4ea7c 100644 --- a/src/orchestrator/sweep-budget.test.ts +++ b/src/orchestrator/sweep-budget.test.ts @@ -151,6 +151,24 @@ describe('sweep budget primitive', () => { } }) + it('must-not-fire: the budget timer holds the event loop, and stops holding it the moment the sweep settles', async () => { + // The pair for the shutdown lever below. The trivially wrong way to stop a + // deadline timer holding the process open is to `unref()` it — which also + // lets Node exit before the budget fires, so a one-shot `runOnce()` returns + // having neither reported the wedge nor released the lease. This assertion + // fails for that version: `getActiveResourcesInfo()` lists only resources + // that are KEEPING THE EVENT LOOP ALIVE, so an unref'd timer never appears. + const timers = () => process.getActiveResourcesInfo().filter((resource) => resource === 'Timeout').length + const before = timers() + const budget = startDiscoverySweepBudget(90 * 60_000) + expect(timers()).toBe(before + 1) + expect(await budget.run('run-once', async () => 'served')).toBe('served') + budget.dispose() + // ...and the other half: a settled sweep leaves nothing behind, which is + // what makes keeping it referenced affordable at a 90-minute budget. + expect(timers()).toBe(before) + }) + it('must-not-fire: with no budget the same hung call stays pending, so the rejections above are the budget', async () => { // The control for every must-fire above. Without it a wrapper that // rejected unconditionally would pass all of them. @@ -354,9 +372,22 @@ class HangingWatermarkMount extends FakeMountClient { } } +/** A fleet whose control-plane probe is slow enough to spend a tight budget. */ +class SlowRosterFleet extends FakeFleetClient { + rosterDelayMs = 0 + + override async roster(): ReturnType { + if (this.rosterDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, this.rosterDelayMs)) + } + return await super.roster() + } +} + /** A store that records the lease handbacks, so "released" is observed, not inferred. */ class LeaseWatchingStateStore extends InMemoryStateStore { readonly released: number[] = [] + claims = 0 /** Milliseconds a claim takes to answer. Long enough and the budget gives up first. */ claimDelayMs = 0 @@ -366,6 +397,7 @@ class LeaseWatchingStateStore extends InMemoryStateStore { nowMs: number, leaseMs: number, ): Promise { + this.claims += 1 if (this.claimDelayMs > 0) { await new Promise((resolve) => setTimeout(resolve, this.claimDelayMs)) } @@ -464,6 +496,107 @@ describe('a wedged sweep is bounded end to end (#372)', () => { } }) + it('must-fire: shutdown does not inherit the sweep budget when a sweep is wedged', async () => { + // `stop()` deliberately outlives the sweep it started (#301), so without a + // shutdown lever a wedged sweep makes shutdown as long as the budget — 90 + // minutes at the default. A 60 s budget here stands in for that: if + // shutdown waited for it, the 4 s guard below fires. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-drain-')) + const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) + // Serve the pre-backfill read AND the startup backfill's own, so the + // daemon comes up healthy and it is a LATER, periodic sweep that wedges. + mount.serveFirst = 2 + const factory = createFactory(config(60_000, root), { + mount, + fleet: new FakeFleetClient(), + stateStore: new LeaseWatchingStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + await withDeadline(factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }), 4_000, 'start never returned') + await withDeadline( + (async () => { + while (mount.hungCalls === 0) await new Promise((resolve) => setTimeout(resolve, 25)) + })(), + 4_000, + 'no periodic sweep ever wedged', + ) + await withDeadline(factory.stop(), 4_000, 'stop waited for the whole sweep budget') + } finally { + await factory.stop().catch(() => undefined) + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-fire: a sweep aborted before the claim phase never opens a lease', async () => { + // A lease taken after the budget expired makes every later sweep defer — + // the same "later cycles wait on a pass that is already over" failure the + // PR's table pins on `reconcileTimeoutMs`. + // + // Scope, stated because it is narrower than it looks: this drives the + // ordinary path, where the budget is spent in an EARLIER phase and that + // phase's own `budget.run` throws first. Every entry into + // `#claimDiscoverySweepUnderBudget` is preceded by such a `budget.run`, so + // "already spent on entry" is not reachable by construction — it is the + // microtask gap between that check and the claim, which `stop()` made + // reachable when it gained the ability to spend a budget asynchronously. + // Issuing the claim inside the budget callback closes it by construction; + // the guarantee that makes that work is asserted directly on the primitive + // above ("refuses to start new work once the budget is spent", which + // asserts the thunk is never invoked). + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-spent-')) + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + const fleet = new SlowRosterFleet() + // Longer than the budget, so the budget is already spent by the time the + // claim phase is entered. + fleet.rosterDelayMs = 500 + const factory = createFactory(config(150, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901) }), + fleet, + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + try { + await expect(withDeadline(factory.runOnce(), 4_000, 'sweep never settled')) + .rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + phase: 'fleet-control-plane-probe', + }) + expect(stateStore.claims).toBe(0) + expect(stateStore.released).toHaveLength(0) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-not-fire: a healthy sweep still claims exactly once and hands the lease back', async () => { + // The pair for the test above: the trivially wrong way to stop a spent + // budget from claiming is to stop claiming. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-claims-')) + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + const factory = createFactory(config(30_000, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901) }), + fleet: new SlowRosterFleet(), + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + try { + const report = await withDeadline(factory.runOnce(), 4_000, 'healthy sweep never settled') + expect(report.dispatched).toHaveLength(1) + expect(stateStore.claims).toBe(1) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it('must-fire: a lease that lands after the budget gave up on the claim is handed straight back', async () => { // The budget abandons the WAIT, not the call, so the store can persist a // lease for a claim this sweep has already left. Nobody would renew, diff --git a/src/orchestrator/sweep-budget.ts b/src/orchestrator/sweep-budget.ts index 75b17ab1..f5d0f744 100644 --- a/src/orchestrator/sweep-budget.ts +++ b/src/orchestrator/sweep-budget.ts @@ -103,6 +103,15 @@ export interface DiscoverySweepBudget { run(phase: DiscoverySweepPhase, start: () => Promise): Promise /** Throw if the budget is already spent. A between-await check; see above. */ assertNotExpired(phase: DiscoverySweepPhase): void + /** + * Spend the budget now. + * + * Shutdown's lever. `stop()` drains the sweep it started, so a wedged sweep + * would otherwise hold shutdown open for the whole budget — 90 minutes by + * default. Expiring it turns that into the sweep's ordinary abort path: + * lease released, teardown bounded, `stop()` free to continue. + */ + expire(): void /** Releases the timer. Always call it, or a settled sweep leaves one pending. */ dispose(): void } @@ -135,14 +144,18 @@ export function startDiscoverySweepBudget(timeoutMs: number | undefined): Discov expire = () => resolve(BUDGET_EXPIRED) }) + const spend = (): void => { + if (expired || budgetMs === undefined) return + expired = true + if (timer) clearTimeout(timer) + // `run()` supplies the phase on the throw; the abort reason cannot know + // which await was in flight, so it names the sweep as a whole. + controller.abort(new DiscoverySweepBudgetExceededError(budgetMs, 'run-once')) + expire() + } + if (budgetMs !== undefined) { - timer = setTimeout(() => { - expired = true - // `run()` supplies the phase on the throw; the abort reason cannot know - // which await was in flight, so it names the sweep as a whole. - controller.abort(new DiscoverySweepBudgetExceededError(budgetMs, 'run-once')) - expire() - }, budgetMs) + timer = setTimeout(spend, budgetMs) // Deliberately NOT unref'd, unlike the per-call deadline in // `relayfile-operation-timeout.ts`. That one is created thousands of times // per sweep and can afford to lose a race with process exit; this one is @@ -162,6 +175,7 @@ export function startDiscoverySweepBudget(timeoutMs: number | undefined): Discov signal: controller.signal, expired: () => expired, assertNotExpired, + expire: spend, async run(phase: DiscoverySweepPhase, start: () => Promise): Promise { if (budgetMs === undefined) return await start() // Decided before the call is made, so an already-spent budget cannot From 1e9d2ec94e199b99438be78ccdb22ed031afc8cb Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 14:09:28 +0200 Subject: [PATCH 5/6] fix(orchestrator): arm the shutdown lever before shutdown's first await MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers both open review threads on #374. `stop()` armed the sweep-budget grace timer only after awaiting `#heldAgentDeadlineSweepInFlight`, so an unrelated in-flight held-agent sweep silently extended a wedged discovery sweep's reprieve from `grace` to `held-agent sweep duration + grace` — unbounded if that sweep never returns, which is precisely the bound this change exists to provide. The grace is a timer; arming it costs nothing, so it now starts the clock at the moment shutdown starts, and the teardown it guards moved inside its `try` so the timer is still cleared on every path. Covered by a new must-fire that observes WHEN the lever arms rather than that it exists: a discovery sweep is wedged, a held-agent sweep is parked mid-release through the fleet, and the shutdown counter is read 3.2s into `stop()` — past the 2.5s grace, far short of the 60s budget. It fails `expected undefined to be 1` against the previous ordering. Also wraps the one test that arms the real 90-minute timer in `try/finally`. Without it a failing assertion left a *referenced* `Timeout` in the worker, so the failure would present as a hung suite instead of a named assertion. Proven with a temporary must-fire / must-not-fire pair: the old shape leaves the timer active after a throw, the new one does not. Co-Authored-By: Claude Opus 5 Session-Id: 372b13bc-44a2-45a5-b5cc-aa228ccca39d --- src/orchestrator/factory.ts | 33 ++++--- src/orchestrator/sweep-budget.test.ts | 124 +++++++++++++++++++++++++- 2 files changed, 141 insertions(+), 16 deletions(-) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 75b3ced9..10f205b6 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -1583,24 +1583,33 @@ export class FactoryLoop implements Factory { if (this.#heldAgentDeadlineTimer) clearTimeout(this.#heldAgentDeadlineTimer) this.#heldAgentDeadlineTimer = undefined this.#heldAgentDeadlineDueAtMs = undefined - await this.#heldAgentDeadlineSweepInFlight - for (const timer of this.#dispatchLifecycleRetryTimers.values()) clearTimeout(timer) - this.#dispatchLifecycleRetryTimers.clear() - this.#abandonedDispatchReasons.clear() - this.#dispatchLifecycleCapacityWaits.clear() - this.#dispatchLifecycleOwnershipWaitLogged.clear() - if (this.#completionSweepTimer) clearTimeout(this.#completionSweepTimer) - this.#completionSweepTimer = undefined - if (this.#readinessReconcileTimer) clearTimeout(this.#readinessReconcileTimer) - this.#readinessReconcileTimer = undefined - if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) - this.#previewSweepTimer = undefined // #372: the drain below is unbounded in the one case that matters — a // wedged sweep — so shutdown inherits the sweep budget, 90 minutes at the // default. Spending it after one teardown window bounds shutdown without // discarding a sweep that was about to finish. + // + // Armed BEFORE the first shutdown await, not just before the sweep drain + // (cubic-dev-ai, #374 review). The grace is a timer, so arming it costs + // nothing and starts the clock at the moment shutdown starts; arming it + // after `#heldAgentDeadlineSweepInFlight` made the lever's start depend on + // an UNRELATED in-flight sweep finishing first, so a slow held-agent pass + // simply added its own latency to the wedged discovery sweep's reprieve — + // in the limit the lever never arms at all, which is the bound this whole + // change exists to provide. const releaseSweepBudgetGrace = this.#cutSweepBudgetsShortForStop() try { + await this.#heldAgentDeadlineSweepInFlight + for (const timer of this.#dispatchLifecycleRetryTimers.values()) clearTimeout(timer) + this.#dispatchLifecycleRetryTimers.clear() + this.#abandonedDispatchReasons.clear() + this.#dispatchLifecycleCapacityWaits.clear() + this.#dispatchLifecycleOwnershipWaitLogged.clear() + if (this.#completionSweepTimer) clearTimeout(this.#completionSweepTimer) + this.#completionSweepTimer = undefined + if (this.#readinessReconcileTimer) clearTimeout(this.#readinessReconcileTimer) + this.#readinessReconcileTimer = undefined + if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) + this.#previewSweepTimer = undefined await this.#readinessReconcileInFlight // #301 review: the deadline ends the *wait*, so `#readinessReconcileInFlight` // can settle with its `runOnce()` still live. Shutdown releases dispatch diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts index 04d4ea7c..58eab57e 100644 --- a/src/orchestrator/sweep-budget.test.ts +++ b/src/orchestrator/sweep-budget.test.ts @@ -161,9 +161,19 @@ describe('sweep budget primitive', () => { const timers = () => process.getActiveResourcesInfo().filter((resource) => resource === 'Timeout').length const before = timers() const budget = startDiscoverySweepBudget(90 * 60_000) - expect(timers()).toBe(before + 1) - expect(await budget.run('run-once', async () => 'served')).toBe('served') - budget.dispose() + // `dispose()` has to run on the failing path too, not just the passing one. + // This is the ONE test in the file that deliberately arms the real 90-minute + // timer, so an assertion that throws before an unguarded `dispose()` would + // leave a *referenced* Timeout in the worker — the exact resource the next + // line proves keeps the event loop alive. The failure would then present as + // a hung suite instead of a named assertion, which is the worst way to + // report it (cubic-dev-ai, #374 review). + try { + expect(timers()).toBe(before + 1) + expect(await budget.run('run-once', async () => 'served')).toBe('served') + } finally { + budget.dispose() + } // ...and the other half: a settled sweep leaves nothing behind, which is // what makes keeping it referenced affordable at a 90-minute budget. expect(timers()).toBe(before) @@ -261,7 +271,11 @@ const implementing = '22222222-2222-4222-8222-222222222222' const done = '33333333-3333-4333-8333-333333333333' const planning = '44444444-4444-4444-8444-444444444444' -const config = (sweepBudgetMs: number, registryRoot: string): FactoryConfig => FactoryConfigSchema.parse({ +const config = ( + sweepBudgetMs: number, + registryRoot: string, + dispatch?: { agentHoldTimeoutMs?: number; agentlessHoldTimeoutMs?: number }, +): FactoryConfig => FactoryConfigSchema.parse({ workspaceId: 'factory-sweep-budget', repos: { byLabel: { pear: 'AgentWorkforce/pear' }, @@ -277,6 +291,7 @@ const config = (sweepBudgetMs: number, registryRoot: string): FactoryConfig => F heartbeatPath: join(registryRoot, 'heartbeat.json'), }, liveSubscription: { sweepBudgetMs }, + ...(dispatch ? { dispatch } : {}), }) const issuePath = (n: number) => `/linear/issues/AR-${n}__uuid-${n}.json` @@ -384,6 +399,42 @@ class SlowRosterFleet extends FakeFleetClient { } } +/** + * A fleet that parks the held-agent deadline sweep partway through, so a sweep + * UNRELATED to discovery is genuinely in flight when `stop()` is called. + * + * `#sweepHeldAgentDeadlines` releases the agents it reaps through + * `fleet.release(name, reason)`, so blocking that one reason is enough to hold + * `#heldAgentDeadlineSweepInFlight` open for as long as the test wants, + * without a timer or a sleep deciding the outcome. + */ +class HeldSweepParkingFleet extends FakeFleetClient { + static readonly HELD_PAST_DEADLINE = 'held-past-deadline' + + #open?: () => void + // Once unparked, STAY unparked: a held dispatch releases every one of its + // agents, so a one-shot gate would park the second release forever and the + // test would fail on shutdown rather than on the assertion it exists for. + #opened = false + #entered!: () => void + /** Resolves once the held-agent sweep is parked inside its release. */ + readonly parked: Promise = new Promise((resolve) => { this.#entered = resolve }) + + override async release(name: string, reason?: string): Promise { + if (reason === HeldSweepParkingFleet.HELD_PAST_DEADLINE && !this.#opened) { + this.#entered() + await new Promise((resolve) => { this.#open = resolve }) + } + await super.release(name, reason) + } + + /** Lets the parked held-agent sweep finish, and keeps it free thereafter. */ + unpark(): void { + this.#opened = true + this.#open?.() + } +} + /** A store that records the lease handbacks, so "released" is observed, not inferred. */ class LeaseWatchingStateStore extends InMemoryStateStore { readonly released: number[] = [] @@ -532,6 +583,71 @@ describe('a wedged sweep is bounded end to end (#372)', () => { } }) + it( + 'must-fire: the shutdown lever arms even while an unrelated held-agent sweep is still in flight', + { timeout: 30_000 }, + async () => { + // The pair for the test above. That one proves the lever exists; this one + // proves WHEN it is armed, which is the half a passing shutdown cannot + // distinguish (cubic-dev-ai, #374 review). + // + // `stop()` arms the grace timer and then drains. Arming it after + // `#heldAgentDeadlineSweepInFlight` made the wedged discovery sweep's + // reprieve equal to `held-agent sweep duration + grace` instead of just + // `grace` — an unrelated subsystem silently extending the one bound this + // change exists to provide, without limit if that sweep never returns. + // + // The grace itself is `STOP_TEARDOWN_TIMEOUT_MS` (factory.ts), 2.5 s. The + // observation below sits at 3.2 s: past the grace, and far short of the + // 60 s budget, so a run that reads the counter as set cannot be a budget + // that simply expired on its own. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-held-')) + const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) + // As above: serve the pre-backfill read and the backfill's own, so the + // daemon comes up healthy and a LATER periodic sweep is the one that wedges. + mount.serveFirst = 2 + const fleet = new HeldSweepParkingFleet() + // 50 ms puts the dispatched agents past their hold deadline almost + // immediately, so the held-agent sweep is running well before shutdown. + const factory = createFactory(config(60_000, root, { agentHoldTimeoutMs: 50 }), { + mount, + fleet, + stateStore: new LeaseWatchingStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + await withDeadline(factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }), 8_000, 'start never returned') + // Both preconditions are OBSERVED, not slept for: a discovery sweep is + // wedged, and a held-agent sweep is parked mid-release. + await withDeadline( + (async () => { + while (mount.hungCalls === 0) await new Promise((resolve) => setTimeout(resolve, 25)) + })(), + 8_000, + 'no periodic sweep ever wedged', + ) + await withDeadline(fleet.parked, 8_000, 'the held-agent deadline sweep never started') + + const stopping = factory.stop() + await new Promise((resolve) => setTimeout(resolve, 3_200)) + // Before the fix this is `undefined`: `stop()` is still parked on the + // held-agent sweep and has not armed the timer that spends the budget. + expect(factory.status().counters.discoverySweepBudgetsCutShortForStop).toBe(1) + + fleet.unpark() + await withDeadline(stopping, 8_000, 'stop never completed') + } finally { + fleet.unpark() + await factory.stop().catch(() => undefined) + await rm(root, { recursive: true, force: true }) + } + }, + ) + it('must-fire: a sweep aborted before the claim phase never opens a lease', async () => { // A lease taken after the budget expired makes every later sweep defer — // the same "later cycles wait on a pass that is already over" failure the From 1b48ff0238656f7339dd62b1cc0c2755031133c4 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 14:21:28 +0200 Subject: [PATCH 6/6] test(orchestrator): poll for the shutdown lever instead of racing its grace MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The new must-fire read the counter after a fixed 3.2s wait against a 2.5s grace — a few hundred milliseconds of headroom, which on a loaded worker is a new flake. This suite already carries two (#342, #373), and adding a third inside the PR whose subject is a wedge is the wrong trade. Polling costs the discrimination nothing: the held-agent sweep stays parked until the test releases it, so against the previous ordering `stop()` never reaches the arming call at all and the poll can only end in its own deadline. Re-measured both directions — pre-fix ordering: exit=1, "the shutdown lever never armed while an unrelated held-agent sweep was in flight"; with the fix: exit=0. Co-Authored-By: Claude Opus 5 Session-Id: 372b13bc-44a2-45a5-b5cc-aa228ccca39d --- src/orchestrator/sweep-budget.test.ts | 31 ++++++++++++++++++++------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts index 58eab57e..71c24da3 100644 --- a/src/orchestrator/sweep-budget.test.ts +++ b/src/orchestrator/sweep-budget.test.ts @@ -585,7 +585,9 @@ describe('a wedged sweep is bounded end to end (#372)', () => { it( 'must-fire: the shutdown lever arms even while an unrelated held-agent sweep is still in flight', - { timeout: 30_000 }, + // Generous on purpose: every bound inside is a failure deadline, so this is + // only ever consumed by a run that is already failing. + { timeout: 60_000 }, async () => { // The pair for the test above. That one proves the lever exists; this one // proves WHEN it is armed, which is the half a passing shutdown cannot @@ -597,10 +599,9 @@ describe('a wedged sweep is bounded end to end (#372)', () => { // `grace` — an unrelated subsystem silently extending the one bound this // change exists to provide, without limit if that sweep never returns. // - // The grace itself is `STOP_TEARDOWN_TIMEOUT_MS` (factory.ts), 2.5 s. The - // observation below sits at 3.2 s: past the grace, and far short of the - // 60 s budget, so a run that reads the counter as set cannot be a budget - // that simply expired on its own. + // The grace itself is `STOP_TEARDOWN_TIMEOUT_MS` (factory.ts), 2.5 s, and + // the budget here is 60 s — so a counter that is set can only have been + // set by the lever, never by a budget that expired on its own. const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-held-')) const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) // As above: serve the pre-backfill read and the backfill's own, so the @@ -633,9 +634,23 @@ describe('a wedged sweep is bounded end to end (#372)', () => { await withDeadline(fleet.parked, 8_000, 'the held-agent deadline sweep never started') const stopping = factory.stop() - await new Promise((resolve) => setTimeout(resolve, 3_200)) - // Before the fix this is `undefined`: `stop()` is still parked on the - // held-agent sweep and has not armed the timer that spends the budget. + // POLLED, not slept (cubic-dev-ai, #374 review). A fixed wait just past + // the 2.5 s grace leaves a few hundred ms of headroom, which on a loaded + // worker is a new flake — in a suite that already carries two (#342, + // #373) and in a PR whose whole subject is a wedge. Polling costs the + // discrimination nothing: the held-agent sweep stays parked until + // `unpark()` below, so against the previous ordering `stop()` never + // reaches the arming call at all and this can only end in its own + // deadline. The margin disappears; the must-fire does not. + await withDeadline( + (async () => { + while (factory.status().counters.discoverySweepBudgetsCutShortForStop === undefined) { + await new Promise((resolve) => setTimeout(resolve, 25)) + } + })(), + 12_000, + 'the shutdown lever never armed while an unrelated held-agent sweep was in flight', + ) expect(factory.status().counters.discoverySweepBudgetsCutShortForStop).toBe(1) fleet.unpark()