diff --git a/src/fleet/control-plane-circuit.test.ts b/src/fleet/control-plane-circuit.test.ts index 0eb31043..c7619174 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 6a8d5b5e..eb8a633c 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 73bd2c18..2ebc4193 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 (Error)' }], + }) await expect(factory.runOnce()).resolves.toMatchObject({ dispatched: [{ issue: { key: '58' } }], }) @@ -4448,6 +4455,267 @@ 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', + ]) + // 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 () => { + 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 }, + // 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) + }) + + // 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. 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([]) + }) + + // 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. + 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(/unclassified dispatch failures without a successful dispatch/) + 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 c2dca8dd..6f0fe2ae 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,10 @@ export class FactoryLoop implements Factory { reason: entry.reason, }) } + // 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 @@ -2531,17 +2560,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. + unclassifiedFailuresSinceDispatch = 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 +2574,52 @@ 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, dryRun)) throw error + if (!isClassifiedPerItemDispatchFailure(error)) { + 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 (unclassifiedFailuresSinceDispatch >= UNCLASSIFIED_DISPATCH_FAILURE_LIMIT) { + throw contextualError( + `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, + unclassifiedFailuresSinceDispatch, + 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 { + // 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, + }) + } + recordSkip({ issue: issueRef(issue), reason: perItemDispatchSkipReason(error) }) + continue } finally { if (recoveredIdentity) this.#reconciledGithubInProgress.delete(recoveredIdentity) } @@ -2584,6 +2652,81 @@ 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 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 + * 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 unclassified-failure fuse in `#performRunOnce` is the + * backstop for a pass-wide fault that does not announce itself as one. + */ + #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 + } + // 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 { this.#discoverySweepRenewTimer = setInterval(() => { if (this.#discoverySweepRenewalInFlight || this.#discoverySweepLeaseLost) return @@ -3555,6 +3698,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 +3755,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 +4824,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 +6986,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 +19202,62 @@ export function isLiveDispatchStateChangedError(error: unknown): error is LiveDi return error instanceof LiveDispatchStateChangedError } +/** 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, + * a live-state race) never count toward it and never reset it: those + * legitimately affect many units at once and are exactly the benign case #292 + * asks the loop to survive, so they are neither evidence of a pass-wide fault + * nor evidence against one. + */ +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 + +/** + * 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) { + return error.refusal === 'terminal' + ? 'dispatch lifecycle already terminal' + : 'dispatch lifecycle owned by another publisher' + } + return `dispatch failed (${telemetryErrorClass(error)})` +} + 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