From a1b5bd6bf14cc72dc64ad70e303cc0de3336f3da Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 11:44:13 +0200 Subject: [PATCH 1/4] fix(orchestrator): skip per-item dispatch failures instead of aborting the pass (#292) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One work unit whose dispatch threw a non-whitelisted error aborted the entire run-once pass, every pass, so nothing behind it was ever dispatched. Production Cloud Factory wedged this way for hours with readinessReconcile degraded on "Refusing to dispatch : dispatch lifecycle is already terminal". Narrow: the dispatch-lifecycle claim refusal is now a typed DispatchLifecycleClaimRefusedError, recorded as a skip with a clear reason alongside its sibling conditions rather than escaping the loop. Class fix: the per-item catch is inverted. Failures raised while processing one work unit skip that unit by default; only the conditions named in #isPassFatalFailure abort the sweep — the discovery sweep lease being lost, Relayfile overload, shutdown, and an open fleet control-plane circuit. Builtin error types are deliberately not in that set: Node reports network failures as `TypeError: fetch failed`, the exact per-item fault behind #291. A consecutive-unclassified-failure fuse keeps the loop from swallowing a pass-wide fault disguised as a run of per-item ones, and a skipped failure now reaps the dispatch-failure handoffs runLoop's catch used to reap, so half-spawned agents cannot leak. Co-Authored-By: Claude Opus 5 Session-Id: a84a310e-4e12-4d54-8cf6-6daa1ac9032e --- src/orchestrator/factory.test.ts | 151 +++++++++++++++++++++++- src/orchestrator/factory.ts | 195 ++++++++++++++++++++++++++++--- 2 files changed, 331 insertions(+), 15 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 73bd2c1..423a50d 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -49,6 +49,7 @@ import type { ConversationMessage, ConversationSessionState, DiscoverySweepClaim import { DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD, DEFAULT_FLEET_ROSTER_TIMEOUT_MS, + FleetControlPlaneCircuitOpenError, } from '../fleet/control-plane-circuit' describe('fleet control-plane admission', () => { @@ -4300,7 +4301,13 @@ describe('FactoryLoop', () => { probePrGhRunner: async () => ({ stdout: '[]' }), }) - await expect(factory.runOnce()).rejects.toThrow('transient triage failure') + // #292: the transient failure costs issue 58 this pass and nothing more — + // it is recorded as a skip instead of aborting the sweep. The exemption + // this test is about is still consumed, so the next pass recovers it. + await expect(factory.runOnce()).resolves.toMatchObject({ + dispatched: [], + skipped: [{ issue: { key: '58' }, reason: 'dispatch failed: transient triage failure' }], + }) await expect(factory.runOnce()).resolves.toMatchObject({ dispatched: [{ issue: { key: '58' } }], }) @@ -4448,6 +4455,148 @@ describe('FactoryLoop', () => { }, ) + // #292: one work unit must not decide the fate of the others. Each case + // below is a per-item failure on issue 59 followed by an untouched, ready + // issue 60 that must still be dispatched in the SAME pass. + describe('per-item dispatch failures do not abort the readiness pass (#292)', () => { + const blockedPath = githubIssuePath('AgentWorkforce', 'pear', 59) + const freshPath = githubIssuePath('AgentWorkforce', 'pear', 60) + + const twoReadyIssues = () => new FakeMountClient({ + [blockedPath]: githubIssueFile(59, { labels: ['factory', 'pear'] }), + [freshPath]: githubIssueFile(60, { labels: ['factory', 'pear'] }), + }) + + // Reproduces the production wedge: the surface issue is open and + // gate-eligible while its durable dispatch-lifecycle record is terminal, + // so `#claimDispatchLifecycle` cannot acquire and refuses the dispatch. + class TerminalLifecycleStateStore extends InMemoryStateStore { + constructor(readonly terminalIssueKey: string) { + super({ batchSize: 4 }) + } + + override async claimDispatchLifecycle( + workspaceId: string, + key: string, + seed: DispatchLifecycle, + owner: string, + nowMs: number, + leaseMs: number, + ) { + if (seed.issue.key === this.terminalIssueKey) { + return { + key, + acquired: false as const, + created: false, + lifecycle: { ...seed, phase: 'complete' as const }, + } + } + return await super.claimDispatchLifecycle(workspaceId, key, seed, owner, nowMs, leaseMs) + } + } + + // Per-item failures raised from triage stand in for every non-lifecycle + // fault that is about one unit (#291's `fetch failed` roster lookup is the + // production example) without coupling the test to a dispatch internal. + class FailingTriage extends StaticTriage { + constructor(readonly failFor: (issue: LinearIssue) => Error | undefined) { + super() + } + + override async triage(issue: LinearIssue): Promise { + const failure = this.failFor(issue) + if (failure) throw failure + return await super.triage(issue) + } + } + + it('skips a work unit whose dispatch lifecycle is already terminal and dispatches the rest of the pass', async () => { + const mount = twoReadyIssues() + const fleet = new LocalLifecycleFleetClient() + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet, + stateStore: new TerminalLifecycleStateStore('59'), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + }) + + const report = await factory.runOnce() + + expect(report.skipped).toContainEqual({ + issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: blockedPath }, + reason: 'dispatch lifecycle already terminal', + }) + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-60-impl-pear', + 'ar-60-review-pear', + ]) + }) + + it('skips a work unit whose per-item failure is unclassified and dispatches the rest of the pass', async () => { + const mount = twoReadyIssues() + const fleet = new LocalLifecycleFleetClient() + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet, + // Node reports a network failure as `TypeError: fetch failed`, which is + // exactly the shape that wedged the second instance (#291). + triage: new FailingTriage((issue) => issue.key === '59' + ? new TypeError('fetch failed') + : undefined), + githubWriteback: new RecordingGithubWriteback(), + }) + + const report = await factory.runOnce() + + expect(report.skipped).toContainEqual({ + issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: blockedPath }, + reason: 'dispatch failed: fetch failed', + }) + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(factory.status().counters.dispatchItemFailuresSkipped).toBe(1) + }) + + // Must-not-fire control. Without this the fix could degrade into swallowing + // every failure and the suite would not notice. + it('still aborts the pass when the fleet control plane is globally unavailable', async () => { + const mount = twoReadyIssues() + const fleet = new LocalLifecycleFleetClient() + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet, + triage: new FailingTriage((issue) => issue.key === '59' + ? new FleetControlPlaneCircuitOpenError(0) + : undefined), + githubWriteback: new RecordingGithubWriteback(), + }) + + await expect(factory.runOnce()).rejects.toThrow(/circuit/) + expect(fleet.spawns).toEqual([]) + }) + + // Must-not-fire control. A pass-wide fault that arrives disguised as a + // string of per-item faults must still surface as a failed pass rather + // than a green report full of skips. + it('still aborts the pass once unclassified per-item failures repeat', async () => { + const paths = [71, 72, 73, 74, 75, 76].map((number) => githubIssuePath('AgentWorkforce', 'pear', number)) + const mount = new FakeMountClient(Object.fromEntries( + paths.map((path, index) => [path, githubIssueFile(71 + index, { labels: ['factory', 'pear'] })]), + )) + const fleet = new LocalLifecycleFleetClient() + const factory = createFactory(config({ issueSource: 'github', batchSize: 5 }), { + mount, + fleet, + triage: new FailingTriage(() => new Error('state store is unreachable')), + githubWriteback: new RecordingGithubWriteback(), + }) + + await expect(factory.runOnce()).rejects.toThrow(/consecutive unclassified dispatch failures/) + expect(fleet.spawns).toEqual([]) + }) + }) + it('adopts a same-repo legacy PR and wakes its babysitter when REST metadata later becomes conflicting', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-orphan-open-pr-')) try { diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index c2dca8d..572f92a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -505,6 +505,26 @@ class DispatchLifecycleOwnedElsewhereError extends Error { } } +/** + * The durable dispatch-lifecycle claim was refused for one work unit: its + * record is already terminal, or another publisher currently holds the lease. + * Both are facts about that single unit — the rest of the pass is unaffected — + * so the readiness loop skips it and keeps going (#292). + * + * Typed rather than left as a plain `Error` so the loop can classify it by + * construction instead of by matching on `Refusing to dispatch ...` text. + */ +class DispatchLifecycleClaimRefusedError extends Error { + constructor( + readonly issueKey: string, + readonly refusal: 'terminal' | 'owned-elsewhere', + message: string, + ) { + super(message) + this.name = 'DispatchLifecycleClaimRefusedError' + } +} + const realClock: Clock = { now: () => Date.now(), sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), @@ -743,6 +763,11 @@ export class FactoryLoop implements Factory { #discoverySweepRenewTimer?: ReturnType #discoverySweepRenewalInFlight?: Promise #discoverySweepLeaseLost = false + // Registry/heartbeat paths the in-flight runLoop iteration would use. A + // per-item dispatch failure now skips instead of aborting the pass (#292), + // so the loop's catch no longer runs the failure-handoff reaper for it; the + // pass reaps inline and must write to the same paths runLoop would. + #loopReapPaths?: { heartbeatPath: string; registryPath: string } #discoveryOverloadError?: unknown #resolvedIssueSource?: IssueSource #integrationInstructions?: string @@ -2423,6 +2448,8 @@ export class FactoryLoop implements Factory { reason: entry.reason, }) } + // Backstop for the skip-by-default catch below: see #292. + let consecutiveUnclassifiedFailures = 0 let lastReadyReadProgressAtMs = this.#clock.now() let readyIssueReads = 0 @@ -2531,17 +2558,10 @@ export class FactoryLoop implements Factory { const decision = await this.triageIssue(issue) triaged.push(decision) - let result: DispatchResult - try { - result = await this.dispatch(decision, { dryRun }) - } catch (error) { - if (!(error instanceof LiveDispatchStateChangedError)) throw error - recordSkip({ issue: decision.issue, reason: 'live state changed during dispatch' }) - this.#logger.info?.('[factory] skipped issue whose live state changed during dispatch', { - issue: decision.issue.key, - }) - continue - } + const result = await this.dispatch(decision, { dryRun }) + // A completed dispatch — even one that parks or escalates the issue — + // proves the pipeline still works, so the fuse below starts over. + consecutiveUnclassifiedFailures = 0 if (result.agents.length === 0 && !dryRun) { const reason = result.hold?.kind === 'dependency-cycle' ? `dependency cycle detected: ${result.hold.cycle?.join(' -> ') ?? 'unknown cycle'}` @@ -2552,6 +2572,45 @@ export class FactoryLoop implements Factory { } else { dispatched.push(result) } + } catch (error) { + // #292: issues in a pass are independent work units, so a failure + // that is about ONE unit costs that unit and nothing else. Only the + // conditions named in `#isPassFatalFailure` — the ones where + // continuing the pass is meaningless — abort the whole sweep. + if (this.#isPassFatalFailure(error)) throw error + if (!isClassifiedPerItemDispatchFailure(error)) { + consecutiveUnclassifiedFailures += 1 + // A pass-wide fault can arrive disguised as a run of per-item + // faults. Skipping every unit would then hand back a green report + // that dispatched nothing, which is the same silent wedge #292 is + // about, wearing the opposite costume. Fail the pass loudly so + // `readinessReconcile.lastError` carries the cause. + if (consecutiveUnclassifiedFailures >= UNCLASSIFIED_DISPATCH_FAILURE_LIMIT) { + throw contextualError( + `Aborting readiness pass after ${consecutiveUnclassifiedFailures} consecutive unclassified dispatch failures`, + error, + ) + } + this.#increment('dispatchItemFailuresSkipped') + this.#logger.warn?.('[factory] skipped a work unit whose dispatch failed; continuing the pass', { + issue: issueRef(issue).key, + consecutiveUnclassifiedFailures, + error: describeError(error).errorMessage, + }) + this.#error(error, issueRef(issue)) + // The failure may have left half-spawned agents behind. runLoop's + // catch used to reap them because this error aborted the pass; + // now that the pass survives, the reap has to happen here or the + // agents leak until the next failed iteration. + await this.#reapDispatchFailureHandoffsNow() + } else { + this.#logger.info?.('[factory] skipped a work unit that cannot be dispatched right now', { + issue: issueRef(issue).key, + error: describeError(error).errorMessage, + }) + } + recordSkip({ issue: issueRef(issue), reason: perItemDispatchSkipReason(error) }) + continue } finally { if (recoveredIdentity) this.#reconciledGithubInProgress.delete(recoveredIdentity) } @@ -2584,6 +2643,55 @@ export class FactoryLoop implements Factory { } } + /** + * Whether a failure raised while processing ONE work unit must abort the + * whole readiness pass instead of skipping that unit. + * + * The default is the opposite, and that inversion is the fix for #292. + * Issues in a pass are independent work units: a failure that is *about one + * unit* — its dispatch-lifecycle record, its live state, a provider fault on + * its own writeback — costs that unit and nothing else. Before this, every + * error except `LiveDispatchStateChangedError` escaped the `for` loop, so a + * single issue whose lifecycle record had gone terminal stopped all dispatch + * indefinitely, every pass. + * + * A condition belongs here only when continuing the pass is meaningless or + * actively harmful — when the failure is about the *pass*, not the item: + * + * - The discovery sweep lease is gone. Another process now owns this + * workspace's sweep, so every remaining read throws the same way and each + * one would be recorded as an ordinary per-issue skip. The run report + * would then claim a clean pass over work this process no longer has the + * right to touch. + * - Relayfile signalled overload for this sweep. The backend is shedding + * load; grinding through the remaining units makes it worse, and + * `#runOnceWithDiscoveryFence` is going to rethrow this at the fence + * anyway. + * - The factory is stopping. Teardown is in progress and dispatching more + * agents now leaks them past the shutdown deadline. + * - The fleet control-plane circuit is open. Dispatch is globally paused — + * this is the same condition `#assertFleetControlPlaneAvailable` refuses + * to *start* a pass on, so it must also stop one already in flight. + * + * Deliberately NOT here: JavaScript builtin error types. Classifying + * "programmer faults" such as `TypeError` as fatal is the obvious next rule + * and it is a trap — Node reports a failed `fetch` as `TypeError: fetch + * failed`, which is precisely the transient per-item roster lookup that + * wedged the second instance (#291). A rule keyed on builtin types would + * have preserved that outage verbatim. + * + * Everything else — a refused lifecycle claim, a terminal lifecycle record, + * a transient network fault on one issue — is per-item: record a skip and + * keep going. The consecutive-failure fuse in `#performRunOnce` is the + * backstop for a pass-wide fault that does not announce itself as one. + */ + #isPassFatalFailure(error: unknown): boolean { + if (this.#discoverySweepLeaseLost || this.#discoveryOverloadError !== undefined || this.#stopping) { + return true + } + return isPassFatalDispatchError(error) + } + #startDiscoverySweepRenewal(epoch: number): void { this.#discoverySweepRenewTimer = setInterval(() => { if (this.#discoverySweepRenewalInFlight || this.#discoverySweepLeaseLost) return @@ -3555,6 +3663,7 @@ export class FactoryLoop implements Factory { ))) const heartbeatPath = opts.heartbeatPath ?? this.#config.loop.heartbeatPath const registryPath = opts.registryPath ?? this.#config.loop.registryPath + this.#loopReapPaths = { heartbeatPath, registryPath } const reports: IterationReport[] = [] let consecutiveFailures = 0 let completed = false @@ -3611,6 +3720,7 @@ export class FactoryLoop implements Factory { completed = true return reports } finally { + this.#loopReapPaths = undefined if (!completed) { await this.#writeLoopHeartbeat(heartbeatPath, registryPath, 'stopping', reports.length, maxIterations) } @@ -4679,10 +4789,15 @@ export class FactoryLoop implements Factory { DISPATCH_LIFECYCLE_LEASE_MS, ) if (!claim.acquired || !claim.lease) { - const reason = isTerminalDispatchLifecycle(claim.lifecycle) + const terminal = isTerminalDispatchLifecycle(claim.lifecycle) + const reason = terminal ? 'dispatch lifecycle is already terminal' : `dispatch lifecycle is owned by ${claim.lifecycle.lease?.owner ?? 'another publisher'}` - throw new Error(`Refusing to dispatch ${decision.issue.key}: ${reason}`) + throw new DispatchLifecycleClaimRefusedError( + decision.issue.key, + terminal ? 'terminal' : 'owned-elsewhere', + `Refusing to dispatch ${decision.issue.key}: ${reason}`, + ) } this.#dispatchLifecycleEpochs.set(claim.key ?? key, claim.lease.epoch) this.#hydrateCostLedger(claim.lifecycle) @@ -6836,7 +6951,10 @@ export class FactoryLoop implements Factory { }) } - async #reapDispatchFailureHandoffsNow(heartbeatPath: string, registryPath: string): Promise { + async #reapDispatchFailureHandoffsNow( + heartbeatPath = this.#loopReapPaths?.heartbeatPath ?? this.#config.loop.heartbeatPath, + registryPath = this.#loopReapPaths?.registryPath ?? this.#config.loop.registryPath, + ): Promise { const handoffs = await this.#state.listFailureHandoffs(this.#workspaceId) if (handoffs.length === 0) { return @@ -19049,6 +19167,55 @@ export function isLiveDispatchStateChangedError(error: unknown): error is LiveDi return error instanceof LiveDispatchStateChangedError } +/** + * Error types that abort a readiness pass rather than skipping one work unit. + * Kept deliberately short; the reasoning for what is in and out of this set + * lives on `FactoryLoop#isPassFatalFailure`. + */ +const PASS_FATAL_DISPATCH_ERRORS: ReadonlyArray Error> = [ + FleetControlPlaneCircuitOpenError, +] + +/** How deep to follow `cause` when classifying a wrapped failure. */ +const PASS_FATAL_CAUSE_DEPTH = 4 + +/** + * How many consecutive *unclassified* per-item failures end the pass. Named + * per-item conditions (a lifecycle claim refusal, a live-state race) never + * count toward it: those legitimately affect many units at once and are + * exactly the benign case #292 asks the loop to survive. + */ +const UNCLASSIFIED_DISPATCH_FAILURE_LIMIT = 5 + +/** + * Failures the loop recognizes as belonging to one work unit. They are always + * skippable and are exempt from the consecutive-failure fuse. + */ +const isClassifiedPerItemDispatchFailure = (error: unknown): boolean => + error instanceof LiveDispatchStateChangedError || + error instanceof DispatchLifecycleClaimRefusedError + +const isPassFatalDispatchError = (error: unknown, depth = 0): boolean => { + if (depth > PASS_FATAL_CAUSE_DEPTH || !(error instanceof Error)) return false + if (isClassifiedPerItemDispatchFailure(error)) return false + if (PASS_FATAL_DISPATCH_ERRORS.some((type) => error instanceof type)) return true + // `contextualError` and the fleet control-plane guard both re-throw wrapped, + // so classification has to follow the cause chain rather than trust the + // outermost type. + return isPassFatalDispatchError((error as { cause?: unknown }).cause, depth + 1) +} + +/** The run-report reason recorded for a work unit the pass could not dispatch. */ +const perItemDispatchSkipReason = (error: unknown): string => { + if (error instanceof LiveDispatchStateChangedError) return 'live state changed during dispatch' + if (error instanceof DispatchLifecycleClaimRefusedError) { + return error.refusal === 'terminal' + ? 'dispatch lifecycle already terminal' + : 'dispatch lifecycle owned by another publisher' + } + return `dispatch failed: ${describeError(error).errorMessage}` +} + const triageEscalationQuestion = (decision: TriageDecision, issue?: { title?: string }): string => { const routedRepos = decision.routes.map((route) => route.repo).filter(Boolean) const subject = issue?.title?.trim() || decision.issue.key From 7e690ab257d9316e0271eb7580c0947645d62d13 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 12:00:27 +0200 Subject: [PATCH 2/4] fix(orchestrator): abort the pass when a per-item roster failure opens the circuit FleetControlPlaneCircuit.probe() records the threshold-crossing failure and rethrows the original transport error, not a FleetControlPlaneCircuitOpenError, so classifying by error type alone skipped the very work unit whose roster request paused dispatch. If that unit was the last ready issue, the pass returned successfully while the circuit was already open and readiness stayed healthy. #isPassFatalFailure now reads the circuit state the way runLoop's catch does. The regression test uses a single ready issue on purpose: with more work behind it the pass would abort one item later on the now-open circuit and hide the gap. Co-Authored-By: Claude Opus 5 Session-Id: a84a310e-4e12-4d54-8cf6-6daa1ac9032e --- src/orchestrator/factory.test.ts | 45 ++++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 12 ++++++--- 2 files changed, 54 insertions(+), 3 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 423a50d..68eee16 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -4576,6 +4576,51 @@ describe('FactoryLoop', () => { expect(fleet.spawns).toEqual([]) }) + // Must-not-fire control. The failure that trips the circuit threshold is + // rethrown by `probe()` as the original transport error, not as a + // FleetControlPlaneCircuitOpenError, so classifying by error type alone + // would skip the very item that paused dispatch and finish the pass green. + it('still aborts the pass when a per-item roster failure opens the circuit', async () => { + class BreakableRosterFleetClient extends LocalLifecycleFleetClient { + failRoster = false + + override async roster() { + if (this.failRoster) { + throw Object.assign(new Error('broker unreachable'), { code: 'ECONNREFUSED' }) + } + return await super.roster() + } + } + + // Deliberately a single ready issue: with more work behind it the pass + // would abort one item later on the now-open circuit, which would hide + // the gap this test exists for. + const mount = new FakeMountClient({ + [blockedPath]: githubIssueFile(59, { labels: ['factory', 'pear'] }), + }) + const fleet = new BreakableRosterFleetClient() + // The pass-wide admission probe succeeds; the broker goes away just + // before this work unit's own admission probe, so the failure that trips + // the threshold surfaces as the raw transport error. + const factory = createFactory(config({ + issueSource: 'github', + batchSize: 4, + fleetHealth: { rosterTimeoutMs: 1_000, failureThreshold: 1, resetTimeoutMs: 60_000 }, + }), { + mount, + fleet, + triage: new FailingTriage((issue) => { + if (issue.key === '59') fleet.failRoster = true + return undefined + }), + githubWriteback: new RecordingGithubWriteback(), + }) + + await expect(factory.runOnce()).rejects.toThrow(/fleet control plane is unavailable/) + expect(factory.status().fleetControlPlane.state).not.toBe('closed') + expect(fleet.spawns).toEqual([]) + }) + // Must-not-fire control. A pass-wide fault that arrives disguised as a // string of per-item faults must still surface as a failed pass rather // than a green report full of skips. diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 572f92a..cb33a89 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -2669,9 +2669,14 @@ export class FactoryLoop implements Factory { * anyway. * - The factory is stopping. Teardown is in progress and dispatching more * agents now leaks them past the shutdown deadline. - * - The fleet control-plane circuit is open. Dispatch is globally paused — - * this is the same condition `#assertFleetControlPlaneAvailable` refuses - * to *start* a pass on, so it must also stop one already in flight. + * - The fleet control-plane circuit is no longer closed. Dispatch is + * globally paused — this is the same condition + * `#assertFleetControlPlaneAvailable` refuses to *start* a pass on, so it + * must also stop one already in flight. This is read as circuit *state*, + * not as an error type, because the failure that trips the threshold is + * rethrown by `FleetControlPlaneCircuit.probe` as the original transport + * error: a type check alone would skip the very item whose roster request + * opened the circuit and let the pass finish reporting healthy. * * Deliberately NOT here: JavaScript builtin error types. Classifying * "programmer faults" such as `TypeError` as fatal is the obvious next rule @@ -2689,6 +2694,7 @@ export class FactoryLoop implements Factory { if (this.#discoverySweepLeaseLost || this.#discoveryOverloadError !== undefined || this.#stopping) { return true } + if (this.#fleetControlPlane.status().state !== 'closed') return true return isPassFatalDispatchError(error) } From 8ca3de7f26ddcb26516ff46ebebb8b74b5d7f9da Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 12:20:14 +0200 Subject: [PATCH 3/4] fix(fleet,orchestrator): name the circuit-open transition and sanitize skip reasons MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on #292. FleetControlPlaneCircuit.probe() now rejects the failure that trips the threshold as FleetControlPlaneCircuitOpenError, keeping the transport error as `cause`. Previously the open transition arrived as an ordinary timeout or `TypeError: fetch failed`, so a caller classifying by error type could not tell "one roster request failed" from "dispatch is now globally paused" — the mirror image of the builtin-type trap this PR argues against. Two existing circuit tests asserted the old contract and now assert the transition plus its cause. IterationReport.skipped[].reason no longer embeds the raw error message. `factory run-once` serializes the report to stdout, so provider text and filesystem paths could leak from a public repo's output; the reason is now a fixed classification plus an allowlisted error class, and the full message stays in the operator log. Classified per-item skips (lifecycle claim refusal, live-state race) now increment dispatchItemsSkippedUndispatchable, so a terminal-lifecycle backlog is visible to counters rather than only to the report. They stay out of counters.errors: an undispatchable unit is a state, not a fault. The fuse counter is renamed unclassifiedFailuresSinceDispatch and its abort message matches, because it is only reset by a completed dispatch and a classified skip neither counts toward it nor clears it. Co-Authored-By: Claude Opus 5 Session-Id: a84a310e-4e12-4d54-8cf6-6daa1ac9032e --- src/fleet/control-plane-circuit.test.ts | 44 +++++++++++++++++++++-- src/fleet/control-plane-circuit.ts | 16 ++++++++- src/orchestrator/factory.test.ts | 13 +++++-- src/orchestrator/factory.ts | 47 +++++++++++++++++-------- 4 files changed, 100 insertions(+), 20 deletions(-) diff --git a/src/fleet/control-plane-circuit.test.ts b/src/fleet/control-plane-circuit.test.ts index 0eb3104..c761917 100644 --- a/src/fleet/control-plane-circuit.test.ts +++ b/src/fleet/control-plane-circuit.test.ts @@ -41,8 +41,14 @@ describe('FleetControlPlaneCircuit', () => { await firstFailure expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 }) + // The threshold-crossing failure rejects as the open transition, not as + // the bare timeout that caused it (factory#292); the timeout stays on + // `cause` so the diagnostic is not lost. const second = circuit.probe(never) - const secondFailure = expect(second).rejects.toMatchObject({ name: 'TimeoutError' }) + const secondFailure = expect(second).rejects.toMatchObject({ + name: 'FleetControlPlaneCircuitOpenError', + cause: expect.objectContaining({ name: 'TimeoutError' }), + }) await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS) await secondFailure expect(circuit.status()).toMatchObject({ state: 'open', consecutiveFailures: 2, retryAtMs: 61_000 }) @@ -87,6 +93,35 @@ describe('FleetControlPlaneCircuit', () => { expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 }) }) + // factory#292: the failure that trips the threshold IS the open transition, + // but it arrives as an ordinary transport error. A caller that classifies by + // error type — a dispatcher deciding whether one work unit's failure should + // abort the whole pass — cannot tell the two apart unless probe() names it. + it('MUST FIRE: the failure that trips the threshold rejects as circuit-open, keeping the cause', async () => { + const transport = Object.assign(new TypeError('fetch failed'), { code: 'ECONNREFUSED' }) + const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 1, resetTimeoutMs: 1_000 }) + + const rejection = await circuit.probe(async () => { throw transport }).catch((error: unknown) => error) + + expect(rejection).toBeInstanceOf(FleetControlPlaneCircuitOpenError) + expect((rejection as { code?: unknown }).code).toBe('FACTORY_FLEET_CONTROL_CIRCUIT_OPEN') + expect((rejection as { cause?: unknown }).cause).toBe(transport) + expect(circuit.status()).toMatchObject({ state: 'open', consecutiveFailures: 1 }) + }) + + // MUST NOT FIRE control for the above: a failure that leaves the circuit + // closed is one unit's problem and must keep its own identity, or a caller + // would treat every transient fault as a global pause. + it('MUST NOT FIRE: a failure below the threshold still rejects with the original error', async () => { + const transport = Object.assign(new TypeError('fetch failed'), { code: 'ECONNREFUSED' }) + const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 2, resetTimeoutMs: 1_000 }) + + const rejection = await circuit.probe(async () => { throw transport }).catch((error: unknown) => error) + + expect(rejection).toBe(transport) + expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 }) + }) + it('coalesces a rejected probe, rejects every waiter, and does not cache the failure', async () => { const sharedError = Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' }) let rejectProbe: ((error: Error) => void) | undefined @@ -217,8 +252,13 @@ describe('FleetControlPlaneCircuit', () => { await firstFailure expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 }) + // As above: the admission probe that opens the circuit reports the + // transition, keeping the timeout as `cause`. const second = guarded.spawn({ name: 'worker-2', capability: 'spawn:codex' }) - const secondFailure = expect(second).rejects.toMatchObject({ name: 'TimeoutError' }) + const secondFailure = expect(second).rejects.toMatchObject({ + name: 'FleetControlPlaneCircuitOpenError', + cause: expect.objectContaining({ name: 'TimeoutError' }), + }) await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS) await secondFailure expect(circuit.status()).toMatchObject({ state: 'open', consecutiveFailures: 2 }) diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts index 6a8d5b5..eb8a633 100644 --- a/src/fleet/control-plane-circuit.ts +++ b/src/fleet/control-plane-circuit.ts @@ -102,7 +102,21 @@ export class FleetControlPlaneCircuit { const probe = withTimeout(roster, this.#timeoutMs) .catch((error: unknown) => { this.recordFailure(error) - throw error + // The failure that trips the threshold IS the open transition, but the + // transport error it arrives as says nothing about that. Callers that + // saw only the original error could not tell "one roster request + // failed" from "dispatch is now globally paused" — factory#292 — + // without re-reading status() after every rejection. Name the + // transition here, the same way the two branches above already do, and + // keep the original as `cause` for diagnostics. + const settled = this.status() + if (settled.state === 'closed') throw error + const opened = new FleetControlPlaneCircuitOpenError( + settled.retryAtMs ?? this.#now(), + settled.state, + ) + ;(opened as Error & { cause?: unknown }).cause = error + throw opened }) .then((result) => { const settledStatus = this.status() diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 68eee16..20f5762 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -4306,7 +4306,7 @@ describe('FactoryLoop', () => { // this test is about is still consumed, so the next pass recovers it. await expect(factory.runOnce()).resolves.toMatchObject({ dispatched: [], - skipped: [{ issue: { key: '58' }, reason: 'dispatch failed: transient triage failure' }], + skipped: [{ issue: { key: '58' }, reason: 'dispatch failed (Error)' }], }) await expect(factory.runOnce()).resolves.toMatchObject({ dispatched: [{ issue: { key: '58' } }], @@ -4532,6 +4532,10 @@ describe('FactoryLoop', () => { 'ar-60-impl-pear', 'ar-60-review-pear', ]) + // Visible as a number, not only as a report line: a terminal-lifecycle + // backlog must not be invisible to an operator watching counters. + expect(factory.status().counters.dispatchItemsSkippedUndispatchable).toBe(1) + expect(factory.status().counters.errors ?? 0).toBe(0) }) it('skips a work unit whose per-item failure is unclassified and dispatches the rest of the pass', async () => { @@ -4552,7 +4556,9 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: blockedPath }, - reason: 'dispatch failed: fetch failed', + // Sanitized: `run-once` prints the report as JSON, so the reason + // carries a classification rather than raw provider text. + reason: 'dispatch failed (TypeError)', }) expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) expect(factory.status().counters.dispatchItemFailuresSkipped).toBe(1) @@ -4637,7 +4643,8 @@ describe('FactoryLoop', () => { githubWriteback: new RecordingGithubWriteback(), }) - await expect(factory.runOnce()).rejects.toThrow(/consecutive unclassified dispatch failures/) + await expect(factory.runOnce()) + .rejects.toThrow(/unclassified dispatch failures without a successful dispatch/) expect(fleet.spawns).toEqual([]) }) }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index cb33a89..ddaa0cd 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -2448,8 +2448,10 @@ export class FactoryLoop implements Factory { reason: entry.reason, }) } - // Backstop for the skip-by-default catch below: see #292. - let consecutiveUnclassifiedFailures = 0 + // Backstop for the skip-by-default catch below: see #292. Reset only by + // a completed dispatch, so the name says "since a dispatch" rather than + // "consecutive" — a benign classified skip in between does not clear it. + let unclassifiedFailuresSinceDispatch = 0 let lastReadyReadProgressAtMs = this.#clock.now() let readyIssueReads = 0 @@ -2561,7 +2563,7 @@ export class FactoryLoop implements Factory { const result = await this.dispatch(decision, { dryRun }) // A completed dispatch — even one that parks or escalates the issue — // proves the pipeline still works, so the fuse below starts over. - consecutiveUnclassifiedFailures = 0 + unclassifiedFailuresSinceDispatch = 0 if (result.agents.length === 0 && !dryRun) { const reason = result.hold?.kind === 'dependency-cycle' ? `dependency cycle detected: ${result.hold.cycle?.join(' -> ') ?? 'unknown cycle'}` @@ -2579,22 +2581,24 @@ export class FactoryLoop implements Factory { // continuing the pass is meaningless — abort the whole sweep. if (this.#isPassFatalFailure(error)) throw error if (!isClassifiedPerItemDispatchFailure(error)) { - consecutiveUnclassifiedFailures += 1 + unclassifiedFailuresSinceDispatch += 1 // A pass-wide fault can arrive disguised as a run of per-item // faults. Skipping every unit would then hand back a green report // that dispatched nothing, which is the same silent wedge #292 is // about, wearing the opposite costume. Fail the pass loudly so // `readinessReconcile.lastError` carries the cause. - if (consecutiveUnclassifiedFailures >= UNCLASSIFIED_DISPATCH_FAILURE_LIMIT) { + if (unclassifiedFailuresSinceDispatch >= UNCLASSIFIED_DISPATCH_FAILURE_LIMIT) { throw contextualError( - `Aborting readiness pass after ${consecutiveUnclassifiedFailures} consecutive unclassified dispatch failures`, + `Aborting readiness pass after ${unclassifiedFailuresSinceDispatch} unclassified dispatch failures without a successful dispatch`, error, ) } this.#increment('dispatchItemFailuresSkipped') + // The raw message is operator-facing only; the run report carries + // the sanitized classification from `perItemDispatchSkipReason`. this.#logger.warn?.('[factory] skipped a work unit whose dispatch failed; continuing the pass', { issue: issueRef(issue).key, - consecutiveUnclassifiedFailures, + unclassifiedFailuresSinceDispatch, error: describeError(error).errorMessage, }) this.#error(error, issueRef(issue)) @@ -2604,6 +2608,11 @@ export class FactoryLoop implements Factory { // agents leak until the next failed iteration. await this.#reapDispatchFailureHandoffsNow() } else { + // Not an error — the unit simply cannot be dispatched right now — + // so this stays out of `counters.errors` and gets its own counter + // instead, or a terminal-lifecycle backlog would be invisible to + // anyone watching only `dispatchItemFailuresSkipped`. + this.#increment('dispatchItemsSkippedUndispatchable') this.#logger.info?.('[factory] skipped a work unit that cannot be dispatched right now', { issue: issueRef(issue).key, error: describeError(error).errorMessage, @@ -2687,7 +2696,7 @@ export class FactoryLoop implements Factory { * * Everything else — a refused lifecycle claim, a terminal lifecycle record, * a transient network fault on one issue — is per-item: record a skip and - * keep going. The consecutive-failure fuse in `#performRunOnce` is the + * keep going. The unclassified-failure fuse in `#performRunOnce` is the * backstop for a pass-wide fault that does not announce itself as one. */ #isPassFatalFailure(error: unknown): boolean { @@ -19186,10 +19195,12 @@ const PASS_FATAL_DISPATCH_ERRORS: ReadonlyArray { return isPassFatalDispatchError((error as { cause?: unknown }).cause, depth + 1) } -/** The run-report reason recorded for a work unit the pass could not dispatch. */ +/** + * The run-report reason recorded for a work unit the pass could not dispatch. + * + * `factory run-once` serializes the whole report to stdout, so this string is + * a public surface: it stays a fixed classification plus an allowlisted error + * class name, never raw provider text or filesystem paths. The full message + * goes to the operator log instead, the same split + * `describeControlPlaneError` makes for circuit state. + */ const perItemDispatchSkipReason = (error: unknown): string => { if (error instanceof LiveDispatchStateChangedError) return 'live state changed during dispatch' if (error instanceof DispatchLifecycleClaimRefusedError) { @@ -19219,7 +19238,7 @@ const perItemDispatchSkipReason = (error: unknown): string => { ? 'dispatch lifecycle already terminal' : 'dispatch lifecycle owned by another publisher' } - return `dispatch failed: ${describeError(error).errorMessage}` + return `dispatch failed (${telemetryErrorClass(error)})` } const triageEscalationQuestion = (decision: TriageDecision, issue?: { title?: string }): string => { From 971b9fa0518fe15de201eb1d91e7543018121cd4 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 12:35:08 +0200 Subject: [PATCH 4/4] fix(orchestrator): exempt dry runs from the fleet-circuit fatal rule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The circuit rule added for the previous review round made an open circuit fatal to every pass, but fleet admission is live-only: #assertFleetControlPlaneAvailable is not called for a dry run and a dry run spawns nothing, so a globally paused control plane is irrelevant to it rather than fatal to it. One live pass that tripped the circuit would otherwise poison every later dry run for the whole reset window, including the boot gate's own `run-once --dry-run` probe — turning a recoverable circuit-open condition into a failed boot. #isPassFatalFailure now takes the effective dryRun and splits its rules: the sweep-scoped conditions (lease lost, Relayfile overload, shutdown) hold for every pass, while the fleet-scoped rule, now named #isFleetControlPlaneHalted, applies only to live passes. The one-entry fatal type table is replaced by a generic cause-chain walk, since the fleet rule is the only thing that used it. Tested as a pair on one open circuit with opposite verdicts: a dry run skips the per-item fault and completes, a live pass still aborts. Co-Authored-By: Claude Opus 5 Session-Id: a84a310e-4e12-4d54-8cf6-6daa1ac9032e --- src/orchestrator/factory.test.ts | 67 +++++++++++++++++++++++++++ src/orchestrator/factory.ts | 79 +++++++++++++++++++------------- 2 files changed, 115 insertions(+), 31 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 20f5762..2ebc419 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -4627,6 +4627,73 @@ describe('FactoryLoop', () => { expect(fleet.spawns).toEqual([]) }) + // The circuit rule is fleet-scoped, so it is live-only. A dry run never + // calls fleet admission and never spawns, so an open circuit is irrelevant + // to it — otherwise one live pass that trips the circuit would poison every + // later dry run, including the boot gate's `run-once --dry-run` probe. + // These two cases are a pair: the same open circuit, opposite verdicts. + const openTheCircuit = async () => { + class BrieflyBrokenRosterFleetClient extends LocalLifecycleFleetClient { + failRoster = false + + override async roster() { + if (this.failRoster) { + throw Object.assign(new Error('broker unreachable'), { code: 'ECONNREFUSED' }) + } + return await super.roster() + } + } + + const fleet = new BrieflyBrokenRosterFleetClient() + const failures = new Map() + const factory = createFactory(config({ + issueSource: 'github', + batchSize: 4, + fleetHealth: { rosterTimeoutMs: 1_000, failureThreshold: 1, resetTimeoutMs: 60_000 }, + }), { + mount: new FakeMountClient({ + [blockedPath]: githubIssueFile(59, { labels: ['factory', 'pear'] }), + [freshPath]: githubIssueFile(60, { labels: ['factory', 'pear'] }), + }), + fleet, + triage: new FailingTriage((issue) => { + if (issue.key === '59') fleet.failRoster = true + return failures.get(issue.key) + }), + githubWriteback: new RecordingGithubWriteback(), + }) + + // A live pass trips the circuit and aborts, as it must. + await expect(factory.runOnce()).rejects.toThrow() + expect(factory.status().fleetControlPlane.state).not.toBe('closed') + // The broker recovers, but the circuit stays open for its reset window — + // which is exactly the window a later pass runs in. + fleet.failRoster = false + failures.set('59', new Error('unrelated per-item fault')) + return { factory, fleet } + } + + it('skips per-item failures in a dry run while the fleet circuit is open', async () => { + const { factory, fleet } = await openTheCircuit() + + const report = await factory.runOnce({ dryRun: true }) + + expect(report.skipped).toContainEqual({ + issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: blockedPath }, + reason: 'dispatch failed (Error)', + }) + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(fleet.spawns).toEqual([]) + expect(factory.status().fleetControlPlane.state).not.toBe('closed') + }) + + it('still treats the same open circuit as fatal on a live pass', async () => { + const { factory, fleet } = await openTheCircuit() + + await expect(factory.runOnce()).rejects.toThrow() + expect(fleet.spawns).toEqual([]) + }) + // Must-not-fire control. A pass-wide fault that arrives disguised as a // string of per-item faults must still surface as a failed pass rather // than a green report full of skips. diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index ddaa0cd..6f0fe2a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -2579,7 +2579,7 @@ export class FactoryLoop implements Factory { // that is about ONE unit costs that unit and nothing else. Only the // conditions named in `#isPassFatalFailure` — the ones where // continuing the pass is meaningless — abort the whole sweep. - if (this.#isPassFatalFailure(error)) throw error + if (this.#isPassFatalFailure(error, dryRun)) throw error if (!isClassifiedPerItemDispatchFailure(error)) { unclassifiedFailuresSinceDispatch += 1 // A pass-wide fault can arrive disguised as a run of per-item @@ -2678,14 +2678,16 @@ export class FactoryLoop implements Factory { * anyway. * - The factory is stopping. Teardown is in progress and dispatching more * agents now leaks them past the shutdown deadline. - * - The fleet control-plane circuit is no longer closed. Dispatch is - * globally paused — this is the same condition - * `#assertFleetControlPlaneAvailable` refuses to *start* a pass on, so it - * must also stop one already in flight. This is read as circuit *state*, - * not as an error type, because the failure that trips the threshold is - * rethrown by `FleetControlPlaneCircuit.probe` as the original transport - * error: a type check alone would skip the very item whose roster request - * opened the circuit and let the pass finish reporting healthy. + * - The fleet control-plane circuit is no longer closed, **on a live pass**. + * Dispatch is globally paused — the same condition + * `#assertFleetControlPlaneAvailable` refuses to *start* a live pass on, + * so it must also stop one already in flight. A dry run is exempt: it + * never calls that admission gate and never spawns, so a paused control + * plane is irrelevant to it rather than fatal to it. Without the + * exemption, one live pass that trips the circuit would poison every + * later dry run — including the boot gate's own `run-once --dry-run` + * probe, turning a recoverable circuit-open condition into a failed boot. + * That is a nastier version of the wedge this whole change removes. * * Deliberately NOT here: JavaScript builtin error types. Classifying * "programmer faults" such as `TypeError` as fatal is the obvious next rule @@ -2699,12 +2701,30 @@ export class FactoryLoop implements Factory { * keep going. The unclassified-failure fuse in `#performRunOnce` is the * backstop for a pass-wide fault that does not announce itself as one. */ - #isPassFatalFailure(error: unknown): boolean { + #isPassFatalFailure(error: unknown, dryRun: boolean): boolean { + // Sweep-scoped: these are about this process's right or ability to run the + // pass at all, so they hold for a dry run exactly as for a live one. if (this.#discoverySweepLeaseLost || this.#discoveryOverloadError !== undefined || this.#stopping) { return true } - if (this.#fleetControlPlane.status().state !== 'closed') return true - return isPassFatalDispatchError(error) + // Fleet-scoped, and therefore live-only. See the doc comment above. + return !dryRun && this.#isFleetControlPlaneHalted(error) + } + + /** + * Whether dispatch is globally paused by the fleet control-plane circuit. + * + * Two reads, because the circuit announces itself two different ways. The + * state read covers `guardedMutation`, which records a mutation's own + * transport failure and rethrows the *original* error rather than the + * circuit-open type — converting it there would be wrong, since the mutation + * may already have reached the broker and callers key spawn-failure handling + * off that original error. The type check covers a rejection raised without + * any state transition, such as an already-open circuit refusing admission. + */ + #isFleetControlPlaneHalted(error: unknown): boolean { + return this.#fleetControlPlane.status().state !== 'closed' || + wrapsErrorOfType(error, FleetControlPlaneCircuitOpenError) } #startDiscoverySweepRenewal(epoch: number): void { @@ -19182,18 +19202,25 @@ export function isLiveDispatchStateChangedError(error: unknown): error is LiveDi return error instanceof LiveDispatchStateChangedError } -/** - * Error types that abort a readiness pass rather than skipping one work unit. - * Kept deliberately short; the reasoning for what is in and out of this set - * lives on `FactoryLoop#isPassFatalFailure`. - */ -const PASS_FATAL_DISPATCH_ERRORS: ReadonlyArray Error> = [ - FleetControlPlaneCircuitOpenError, -] - /** How deep to follow `cause` when classifying a wrapped failure. */ const PASS_FATAL_CAUSE_DEPTH = 4 +/** + * Whether `error`, or anything it wraps, is an instance of `type`. + * `contextualError` and the fleet control-plane guard both rethrow wrapped, so + * classification has to follow the cause chain rather than trust the outermost + * type. + */ +const wrapsErrorOfType = ( + error: unknown, + type: abstract new (...args: never[]) => Error, + depth = 0, +): boolean => { + if (depth > PASS_FATAL_CAUSE_DEPTH || !(error instanceof Error)) return false + if (error instanceof type) return true + return wrapsErrorOfType((error as { cause?: unknown }).cause, type, depth + 1) +} + /** * How many *unclassified* per-item failures without an intervening successful * dispatch end the pass. Named per-item conditions (a lifecycle claim refusal, @@ -19212,16 +19239,6 @@ const isClassifiedPerItemDispatchFailure = (error: unknown): boolean => error instanceof LiveDispatchStateChangedError || error instanceof DispatchLifecycleClaimRefusedError -const isPassFatalDispatchError = (error: unknown, depth = 0): boolean => { - if (depth > PASS_FATAL_CAUSE_DEPTH || !(error instanceof Error)) return false - if (isClassifiedPerItemDispatchFailure(error)) return false - if (PASS_FATAL_DISPATCH_ERRORS.some((type) => error instanceof type)) return true - // `contextualError` and the fleet control-plane guard both re-throw wrapped, - // so classification has to follow the cause chain rather than trust the - // outermost type. - return isPassFatalDispatchError((error as { cause?: unknown }).cause, depth + 1) -} - /** * The run-report reason recorded for a work unit the pass could not dispatch. *