From cf427eb50e71653ccc9018737ac52b91fdcea454 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 00:15:47 +0200 Subject: [PATCH 1/9] fix(orchestrator): reap a lifecycle that took a batch slot and never placed an agent (#303) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A dispatch lifecycle that reached a slot-occupying phase and never had an agent placed — spawn failed, fleet fault, or the process died between promote and spawn — was unreapable. `#scheduleHeldAgentDeadline` and `#sweepHeldAgentDeadlines` both keyed on a `heldSinceAtMs` that is only stamped by a *successful* placement, so no timer was ever armed and no sweep ever collected it. With `batchSize` defaulting to 1, that one row held the only slot forever, every other issue was claimed as `queued`, and the retry re-armed at 1 Hz with no bound — 1477 state GETs in 111s on production, across a ~14 hour dispatch outage that every operator surface reported as healthy. Three changes: 1. Bound a never-placed lifecycle. `DispatchLifecycle.slotHeldSinceAtMs` records when a row took its batch slot; `updatedAtMs` could not serve (lease renewal bumps it every 60s) and `heldSinceAtMs` could not either (it is the clock that never starts). Both halves of the reaper now take the shorter `dispatch.agentlessHoldTimeoutMs` (30m) from that anchor, re-derived against the durable row immediately before teardown so a placement that just succeeded is never raced. The predicate is "no successful placement", not `agents.size === 0`: `recordPlanned` writes the spec before the spawn returns, so a dispatch that died mid-spawn leaves an agent entry and no placement. Such an agent is also excluded from the release, since asking the broker to release a name it never issued fails the cleanup and would re-arm the abandon retry forever. 2. Bound the capacity retry. It backs off 1s → 30s and escalates on every step, naming the issues holding the slots, instead of logging once per key and going silent forever. Only the capacity path backs off: an ownership wait is already bounded by the lease. The wait is not abandoned on a deadline — a real multi-hour run holds the slot honestly — so what is bounded is the retry rate. 3. Make batch occupancy observable. `status().dispatchCapacity` and the heartbeat carry slot occupancy, waiters and the longest wait; `/healthz` carries the redacted counts and lists `dispatchCapacity` as dispatch-gating once a wait passes `dispatch.capacityWaitWarnMs`; `factory diagnose` names a wedged batch instead of reporting green. Also extracts the batch-slot predicate the two state stores had each copy-pasted into `src/state/dispatch-lifecycle-slot.ts`, now that a third reader depends on it. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/cli/diagnose.test.ts | 54 +++++ src/cli/diagnose.ts | 30 +++ src/cli/fleet.ts | 8 + src/config/schema.test.ts | 4 + src/config/schema.ts | 30 +++ src/orchestrator/batch-tracker.ts | 8 + src/orchestrator/factory.test.ts | 209 +++++++++++++++++ src/orchestrator/factory.ts | 296 +++++++++++++++++++++---- src/orchestrator/public-health.test.ts | 74 +++++++ src/orchestrator/public-health.ts | 107 ++++++++- src/ports/state.ts | 9 + src/state/dispatch-lifecycle-slot.ts | 58 +++++ src/state/file-state-store.ts | 27 +-- src/state/in-memory-state-store.ts | 26 +-- src/state/watch-state-document.ts | 1 + src/types.ts | 57 +++++ 16 files changed, 920 insertions(+), 78 deletions(-) create mode 100644 src/state/dispatch-lifecycle-slot.ts diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index e7e97703..a4e19ba1 100644 --- a/src/cli/diagnose.test.ts +++ b/src/cli/diagnose.test.ts @@ -174,6 +174,60 @@ describe('factory diagnose --deployed (#295)', () => { expect(report.verdict).toContain('stalled') }) + // #303: a wedged batch was the one dispatch-gating condition every surface + // reported as healthy. `factory diagnose` has to name it, or the operator is + // back to reading the state document. + it('names a wedged batch when everything else reads healthy', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + health: { + schemaVersion: 1, + ok: true, + status: 'degraded', + stale: false, + loopStatus: 'running', + degradedSubsystems: ['dispatchCapacity'], + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: NOW_MS - 30_000, + lastCompletedAtMs: NOW_MS - 29_000, + }, + eventListener: { state: 'subscribed' }, + dispatchCapacity: { + state: 'stalled', + batchSize: 1, + active: 1, + waiting: 7, + waitWarnMs: 1_800_000, + longestWaitMs: 46_800_000, + agentlessOccupants: 1, + }, + }, + }, + }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string } + expect(report.dispatching).toBe(false) + expect(report.verdict).toContain('batch slot') + expect(report.verdict).toContain('never placed an agent') + expect(report.verdict).toContain('7 issue(s) waiting') + }) + // The deployed container serves the block inside its heartbeat projection. it('reads the health block where the container actually serves it', async () => { const out = buffer() diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index c7c619f2..d7409807 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -263,6 +263,23 @@ function verdictFor(diagnosis: Omit 0 + ? `. ${agentless} occupied slot(s) never placed an agent, so they cannot finish on their own` + : '') + + '. Pass --token to read /evidence for the issues holding the slots.', + } + } if (health.status === 'unknown') { return { dispatching: false, @@ -444,6 +461,19 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st lines.push(` lastCompletedAt : ${formatInstant(readiness.lastCompletedAtMs)}`) lines.push(` lastFailureAt : ${formatInstant(readiness.lastFailureAtMs)}`) } + const capacity = health.dispatchCapacity + if (capacity) { + lines.push(' dispatchCapacity:') + lines.push(` state : ${capacity.state}`) + lines.push(` slots : ${capacity.active}/${capacity.batchSize} occupied`) + lines.push(` waiting : ${capacity.waiting} issue(s)`) + lines.push( + ` longest wait : ${formatDuration(capacity.longestWaitMs)} (warn past ${formatDuration(capacity.waitWarnMs)})`, + ) + if (capacity.agentlessOccupants !== undefined) { + lines.push(` never placed agent : ${capacity.agentlessOccupants} occupied slot(s)`) + } + } lines.push(` eventListener : ${health.eventListener?.state ?? 'unknown'}`) } else if (diagnosis.unreadable) { lines.push(' health block : none — this response carried no Factory health') diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 2bd4c5e9..2e74468a 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -1293,6 +1293,12 @@ async function factoryStatusWithMountHealth( const fleetControlPlane = liveness.ok ? heartbeat?.fleetControlPlane : observableStatus.fleetControlPlane + // Same rule as readinessReconcile (#303): a live daemon owns the batch, and + // a fresh local Factory instance holds no lifecycles, so its empty view must + // not be reported as "the batch is free". + const dispatchCapacity = liveness.ok + ? heartbeat?.dispatchCapacity ?? observableStatus.dispatchCapacity + : observableStatus.dispatchCapacity const eventListener = liveness.ok ? heartbeat?.eventListener ?? { state: 'unknown' as const, @@ -1311,6 +1317,7 @@ async function factoryStatusWithMountHealth( eventListener, readinessReconcile, fleetControlPlane, + ...(dispatchCapacity ? { dispatchCapacity } : {}), } return { ...observableStatus, @@ -1320,6 +1327,7 @@ async function factoryStatusWithMountHealth( eventListener, readinessReconcile, fleetControlPlane, + ...(dispatchCapacity ? { dispatchCapacity } : {}), localMountDegraded: health.degraded, ...(health.reason ? { localMountDegradedReason: health.reason } : {}), ...(health.localDir ? { localMountRoot: health.localDir } : {}), diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index d0e7be62..96c204d5 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -32,6 +32,10 @@ describe('FactoryConfigSchema', () => { expect(parsed.dispatch).toEqual({ errorCooldownMs: 60_000, maxAttempts: 2, + // Much shorter than the placed-agent hold: a slot-occupying lifecycle + // that never placed an agent has nothing that can move it (#303). + agentlessHoldTimeoutMs: 30 * 60_000, + capacityWaitWarnMs: 30 * 60_000, agentHoldTimeoutMs: 4 * 60 * 60_000, }) expect(parsed.fleetHealth).toEqual({ diff --git a/src/config/schema.ts b/src/config/schema.ts index c1276062..db61fc1d 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -84,6 +84,21 @@ const liveSubscriptionSchema = z.object({ export const DEFAULT_AGENT_HOLD_TIMEOUT_MS = 4 * 60 * 60_000 +/** + * How long a lifecycle may occupy a batch slot without ever placing an agent. + * + * Much shorter than `agentHoldTimeoutMs` because the two bound different + * things: that one bounds a team that is plausibly working, this one bounds a + * row that definitionally is not — nothing but a placement can move it, and no + * placement ever happened. It still has to clear the whole promote-to-spawn + * window (clone, worktree prep, fleet spawn, roster adoption) with room to + * spare, or the reaper races a dispatch that was about to succeed (#303). + */ +export const DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS = 30 * 60_000 + +/** Capacity wait past which a full batch stops reading as ordinary backpressure. */ +export const DEFAULT_CAPACITY_WAIT_WARN_MS = 30 * 60_000 + const dispatchSchema = z.object({ errorCooldownMs: z.number().int().min(0).default(60_000), maxAttempts: z.number().int().min(1).max(5).default(2), @@ -93,6 +108,21 @@ const dispatchSchema = z.object({ // agent placement until terminal cleanup. agentHoldTimeoutMs: z.number().int().min(1).max(7 * 24 * 60 * 60_000) .default(DEFAULT_AGENT_HOLD_TIMEOUT_MS), + // A slot-occupying lifecycle with no successful placement has no other + // deadline: `agentHoldTimeoutMs` is anchored on a placement that never + // happened, so before #303 nothing could ever reap it. + agentlessHoldTimeoutMs: z.number().int().min(1).max(7 * 24 * 60 * 60_000) + .default(DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS), + /** + * Wall-clock capacity wait past which dispatch is reported degraded (#303). + * + * A full batch is normal; a batch that has been full for hours is how a + * total dispatch outage looked from every operator surface. Deployments that + * legitimately run multi-hour issues against a small `batchSize` should + * raise this rather than lower `batchSize`'s usefulness. + */ + capacityWaitWarnMs: z.number().int().min(1_000).max(7 * 24 * 60 * 60_000) + .default(DEFAULT_CAPACITY_WAIT_WARN_MS), }).default({}) const fleetHealthSchema = z.object({ diff --git a/src/orchestrator/batch-tracker.ts b/src/orchestrator/batch-tracker.ts index cdf37219..d4b25556 100644 --- a/src/orchestrator/batch-tracker.ts +++ b/src/orchestrator/batch-tracker.ts @@ -27,6 +27,13 @@ export interface InFlightIssue { dispatchClaim?: FactoryDispatchClaimStatus /** Wall-clock anchor set when the first agent placement succeeds. */ heldSinceAtMs?: number + /** + * Wall-clock anchor set when this record took a `batchSize` slot (#303). + * + * The only deadline a lifecycle that never placed an agent has: it holds a + * slot, `heldSinceAtMs` is never stamped, and nothing else will move it. + */ + slotHeldSinceAtMs?: number /** Latest durable phase, used only for operator-facing held-agent status. */ lifecyclePhase?: DispatchLifecyclePhase } @@ -315,6 +322,7 @@ export class BatchTracker { invocationIds: new Set(record.invocationIds), result: record.result ? structuredClone(record.result) : undefined, heldSinceAtMs: record.heldSinceAtMs, + slotHeldSinceAtMs: record.slotHeldSinceAtMs, lifecyclePhase: record.lifecyclePhase, } this.#inFlight.set(key, restored) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index da040dd7..5a01d912 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -10852,6 +10852,215 @@ describe('FactoryLoop', () => { } }) + // #303: a lifecycle that reached a slot-occupying phase and never had a + // live agent placement has no `heldSinceAtMs`, so both halves of the + // held-agent reaper skipped it. With batchSize 1 that one row held the only + // slot forever and every other issue spun at 1 Hz in `queued`. + for (const variant of [ + { label: 'never recorded an agent', wedged: 303, queued: 304, planned: false }, + { label: 'recorded a planned agent that never spawned', wedged: 308, queued: 309, planned: true }, + ]) { + it(`reaps a slot-holding lifecycle that ${variant.label} so a queued issue can dispatch (#303)`, async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-agentless-slot-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const wedgedIssue = parseLinearIssue(issuePath(variant.wedged), issueFile(variant.wedged)) + const seed = createFactory(config({ batchSize: 1 }), { + mount: new FakeMountClient({ [issuePath(variant.wedged)]: issueFile(variant.wedged) }), + fleet: new RemoteLifecycleFleetClient(), + stateStore: state(), + triage: new StaticTriage(), + }) + const wedgedDecision = await seed.triageIssue(wedgedIssue) + await seed.stop() + + // The wedged row: promoted into `dispatching`, then the owner died + // before any placement succeeded. Its issue is deliberately absent from + // the replacement's mount, so nothing the resume driver can do will ever + // move it forward. + const plannedSpec = wedgedDecision.implementers[0]! + const wedged: DispatchLifecycle = { + runId: 'wedged-run', + issue: { ...wedgedDecision.issue }, + decision: wedgedDecision, + dryRun: false, + phase: 'dispatching', + agents: variant.planned + ? [{ name: plannedSpec.name, tracked: { spec: { ...plannedSpec } } }] + : [], + invocationIds: [], + updatedAtMs: 0, + } + const wedgedKey = issueKey(wedgedDecision.issue) + await state().claimDispatchLifecycle('factory-test', wedgedKey, wedged, 'dead-owner', 0, 1) + const seeded = await state().getDispatchLifecycle('factory-test', wedgedKey) + expect(seeded?.phase).toBe('dispatching') + expect(seeded?.heldSinceAtMs).toBeUndefined() + expect(seeded?.agents).toHaveLength(variant.planned ? 1 : 0) + + // A real broker rejects a release for a name it never issued. If the + // reaper asked for one, the cleanup would fail and re-arm forever, + // trading the wedge for a quieter one. + class StrictReleaseFleetClient extends RemoteLifecycleFleetClient { + override async release(name: string, reason?: string): Promise { + if (!this.spawns.some((spawn) => spawn.name === name)) { + throw new Error(`unknown agent ${name}`) + } + await super.release(name, reason) + } + } + const fleet = new StrictReleaseFleetClient() + const logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() } + const factory = createFactory(config({ + batchSize: 1, + // The ordinary held-agent deadline stays far away: only the + // never-placed deadline can explain a release here. + dispatch: { agentHoldTimeoutMs: 4 * 60 * 60_000, agentlessHoldTimeoutMs: 1_000 }, + loop: { heartbeatPath: join(root, 'heartbeat.json'), registryPath: join(root, 'registry.json') }, + }), { + mount: new FakeMountClient({ [issuePath(variant.queued)]: issueFile(variant.queued) }), + fleet, + stateStore: state(), + triage: new StaticTriage(), + logger, + }) + try { + await factory.start({ mode: 'backfill-and-subscribe' }) + + await vi.waitFor(async () => expect(await state().getDispatchLifecycle('factory-test', wedgedKey)) + .toMatchObject({ phase: 'abandoned', releaseReason: 'agentless-slot-past-deadline' }), { timeout: 8_000 }) + expect(logger.warn).toHaveBeenCalledWith( + '[factory] releasing a dispatch lifecycle that never placed an agent', + expect.objectContaining({ + issue: `AR-${variant.wedged}`, + phase: 'dispatching', + holdTimeoutMs: 1_000, + }), + ) + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)) + .toEqual([`ar-${variant.queued}-impl-pear`, `ar-${variant.queued}-review`]), { timeout: 8_000 }) + expect(factory.status().counters.agentlessSlotPastDeadlineReleases).toBe(1) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 30_000) + } + + // #303 must-not-fire control. The window between `promoteDispatchLifecycle` + // and the first `recordSpawn` legitimately has zero agents. Reaping it on + // sight would convert a wedge into a dispatch race. + it('does not reap a lifecycle still inside its promote-to-spawn window (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-agentless-window-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const gate = Promise.withResolvers() + class BlockedSpawnFleetClient extends RemoteLifecycleFleetClient { + override async spawn(input: SpawnInput): Promise { + await gate.promise + return super.spawn(input) + } + } + const fleet = new BlockedSpawnFleetClient() + const mount = new FakeMountClient({ [issuePath(305)]: issueFile(305) }) + const factory = createFactory(config({ + batchSize: 1, + // The ordinary held deadline is 1 s and would be long past by the time + // this assertion runs; the agent-less deadline is the one under test. + dispatch: { agentHoldTimeoutMs: 1_000, agentlessHoldTimeoutMs: 60_000 }, + loop: { heartbeatPath: join(root, 'heartbeat.json'), registryPath: join(root, 'registry.json') }, + }), { mount, fleet, stateStore: state(), triage: new StaticTriage() }) + try { + const decision = await factory.triageIssue(parseLinearIssue(issuePath(305), issueFile(305))) + const key = issueKey(decision.issue) + const dispatched = factory.dispatch(decision) + + // `recordPlanned` writes the spec before the spawn returns, so the row + // carries an agent with no result and still has no `heldSinceAtMs` -- + // the same "never placed" shape the reaper must bound. + await vi.waitFor(async () => { + const pending = await state().getDispatchLifecycle('factory-test', key) + expect(pending?.phase).toBe('dispatching') + expect(pending?.heldSinceAtMs).toBeUndefined() + expect(pending?.agents.map((agent) => agent.tracked.result)).toEqual([undefined]) + }, { timeout: 4_000 }) + // Well past `agentHoldTimeoutMs`, nowhere near `agentlessHoldTimeoutMs`. + await new Promise((resolve) => setTimeout(resolve, 2_500)) + expect(await state().getDispatchLifecycle('factory-test', key)).toMatchObject({ phase: 'dispatching' }) + expect(fleet.releases).toEqual([]) + expect(factory.status().counters.agentlessSlotPastDeadlineReleases).toBeUndefined() + + gate.resolve() + await dispatched + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-305-impl-pear', 'ar-305-review']) + await vi.waitFor(async () => expect(await state().getDispatchLifecycle('factory-test', key)) + .toMatchObject({ phase: 'running' }), { timeout: 4_000 }) + } finally { + gate.resolve() + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 30_000) + + // #303 deliverable 2/3: the capacity wait re-armed at a flat 1 Hz forever and + // logged once per key, so an operator saw a healthy, idle-looking Factory. + it('backs off and keeps reporting a dispatch capacity wait instead of spinning silently (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-capacity-wait-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const mount = new FakeMountClient({ + [issuePath(306)]: issueFile(306), + [issuePath(307)]: issueFile(307), + }) + const fleet = new RemoteLifecycleFleetClient() + const capacityWaits: Array> = [] + const logger = { + debug: vi.fn(), + info: vi.fn(), + error: vi.fn(), + warn: vi.fn((message: string, details?: unknown) => { + if (message === '[factory] durable dispatch is queued for batch capacity; retries remain active') { + capacityWaits.push(details as Record) + } + }), + } + const factory = createFactory(config({ batchSize: 1 }), { + mount, + fleet, + stateStore: state(), + triage: new StaticTriage(), + logger, + }) + try { + const running = await factory.triageIssue(parseLinearIssue(issuePath(306), issueFile(306))) + const waiting = await factory.triageIssue(parseLinearIssue(issuePath(307), issueFile(307))) + await factory.dispatch(running) + await factory.dispatch(waiting) + expect(await state().getDispatchLifecycle('factory-test', issueKey(waiting.issue))) + .toMatchObject({ phase: 'queued' }) + + // Deliverable 2: the wait escalates instead of going silent after one + // log, and each re-arm is longer than the last. + await vi.waitFor(() => expect(capacityWaits.length).toBeGreaterThanOrEqual(3), { timeout: 12_000 }) + expect(capacityWaits.map((entry) => entry.retryMs).slice(0, 3)).toEqual([1_000, 2_000, 4_000]) + expect(capacityWaits.every((entry) => (entry.retryMs as number) <= 30_000)).toBe(true) + expect(capacityWaits[2]!.waitedMs as number).toBeGreaterThan(capacityWaits[0]!.waitedMs as number) + expect(capacityWaits[2]!.occupiedBy).toEqual(['AR-306']) + + // Deliverable 3: batch occupancy is on the operator surface rather than + // only in a log line that fires once. + const capacity = factory.status().dispatchCapacity + expect(capacity).toMatchObject({ batchSize: 1, active: 1, waiting: 1 }) + expect(capacity?.longestWaitMs).toBeGreaterThan(0) + expect(capacity?.occupants).toEqual([ + expect.objectContaining({ issue: 'AR-306', phase: 'running', agents: 2 }), + ]) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 30_000) + it('stop releases each in-flight factory-dispatched agent', async () => { const mount = new FakeMountClient({ [issuePath(60)]: issueFile(60) }) const fleet = new CapturedPidFleetClient([ diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 7bad0e00..3186c357 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -56,6 +56,7 @@ 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 { dispatchPhaseOccupiesSlot } from '../state/dispatch-lifecycle-slot' import { containsExplicitIssueReference, containsIssueKey, factoryBranchBelongsToIssue } from '../issue-key-match' import { normalizeLogger, normalizeLogValue, setSafeErrorStack, stringifyLogValue } from '../logging' import { isInFactoryScope } from '../safety/factory-scope' @@ -86,6 +87,8 @@ import type { FactoryLoopRunOptions, FactoryLoopHeartbeat, FactoryLoopLiveness, + FactoryDispatchCapacityStatus, + FactoryDispatchSlotOccupant, FactoryReadinessReconcileStatus, FactoryDispatchClaimStatus, FactoryInFlightDispatchStatus, @@ -408,9 +411,28 @@ const STOP_TEARDOWN_TIMEOUT_MS = 2_500 const DISPATCH_LIFECYCLE_LEASE_MS = 5 * 60_000 const DISPATCH_LIFECYCLE_RENEW_MS = 60_000 const DISPATCH_LIFECYCLE_RETRY_MS = 1_000 +/** + * Ceiling on the durable capacity-wait re-arm (#303). + * + * The retry was a flat 1 Hz with no bound at all. When the batch was wedged, + * every queued issue re-read the shared state document once a second forever — + * production measured 1477 state GETs in 111 s — and the timers could not even + * keep up, so they coalesced into a continuous spin against the serialized + * store. Waiting for capacity is legitimate and must not be abandoned (a real + * multi-hour run holds the slot honestly), so what is bounded is the *rate*. + */ +const DISPATCH_LIFECYCLE_RETRY_MAX_MS = 30_000 +/** Rate limit for the capacity-wait warning once the backoff has capped. */ +const DISPATCH_LIFECYCLE_CAPACITY_WAIT_LOG_MS = 60_000 const DISPATCH_WRITEBACK_MAX_ATTEMPTS = 3 const DISPATCH_WRITEBACK_RETRY_MS = 250 const HELD_PAST_DEADLINE_RELEASE_REASON = 'held-past-deadline' +/** + * Release reason for a lifecycle that took a batch slot and never placed an + * agent (#303). Deliberately distinct from `held-past-deadline`: that one + * means a team ran and never finished, this one means no team ever existed. + */ +const AGENTLESS_SLOT_PAST_DEADLINE_RELEASE_REASON = 'agentless-slot-past-deadline' const HELD_DEADLINE_OVERDUE_RETRY_MS = 1_000 const STARTUP_AGENT_EXIT_DRAIN_TIMEOUT_MS = 30_000 const RECONCILED_AGENT_EXIT_CONCURRENCY = 4 @@ -688,7 +710,22 @@ export class FactoryLoop implements Factory { readonly #dispatchLifecycleRetryTimers = new Map>() readonly #dispatchLifecycleDrives = new Set>() readonly #abandonedDispatchReasons = new Map() - readonly #dispatchLifecycleCapacityWaitLogged = new Set() + /** + * Live batch-capacity waits, keyed by issue (#303). + * + * Replaces a `Set` of "already logged" keys. That set made the wait a + * one-shot log and nothing else: after the first line, an outage in which + * every issue was stuck behind a wedged slot was indistinguishable from an + * idle Factory on every operator surface. The wait now carries its own start + * instant and attempt count, which is what both the escalating warning and + * `status().dispatchCapacity` are derived from. + */ + readonly #dispatchLifecycleCapacityWaits = new Map() readonly #dispatchLifecycleOwnershipWaitLogged = new Set() readonly #dispatchClaimStatuses = new Map() readonly #localReleaseCheckpoints = new Map>() @@ -1274,6 +1311,7 @@ export class FactoryLoop implements Factory { 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 @@ -4700,6 +4738,7 @@ export class FactoryLoop implements Factory { slackDegradedReason: this.#slackDegradedReason, eventListener: this.#eventListenerStatus(), readinessReconcile: this.#readinessReconcileStatus(), + dispatchCapacity: this.#dispatchCapacityStatus(), heldAgents: batch?.inFlight.flatMap((record) => heldAgentsForRecord( record, nowMs, @@ -5145,7 +5184,7 @@ export class FactoryLoop implements Factory { const timer = this.#dispatchLifecycleRetryTimers.get(key) if (timer) clearTimeout(timer) this.#dispatchLifecycleRetryTimers.delete(key) - this.#dispatchLifecycleCapacityWaitLogged.delete(key) + this.#dispatchLifecycleCapacityWaits.delete(key) clearedKeys.add(key) this.#increment('dispatchLifecycleGithubAliasesCollapsed') } @@ -5204,9 +5243,58 @@ export class FactoryLoop implements Factory { this.#dispatchLifecycleRenewTimer.unref?.() } + /** + * The wall-clock deadline that can free this record's batch slot, if any. + * + * Two clocks, never both. `heldSinceAtMs` is stamped by the first successful + * placement and bounds a team that ran and never reached a terminal state. + * A record that has no `heldSinceAtMs` never had a placement at all, and + * before #303 that made it permanently unreapable: `#scheduleHeldAgentDeadline` + * armed no timer and `#sweepHeldAgentDeadlines` skipped it, so a row that + * reached `dispatching` and lost its process held the only batch slot + * forever. Such a row is definitionally stuck — nothing but a placement can + * move it, and no placement is coming — so it gets the much shorter + * `agentlessHoldTimeoutMs` anchored on when it took the slot. + * + * The anchor is deliberately not `agents.size === 0`: `recordPlanned` writes + * the spec before the spawn returns, so a process that died mid-spawn leaves + * an agent entry with no result and still no placement. + */ + #holdDeadline(record: InFlightIssue): { + kind: 'agents' | 'agentless' + sinceAtMs: number + timeoutMs: number + dueAtMs: number + } | undefined { + if (record.dryRun) return undefined + if (record.heldSinceAtMs !== undefined) { + const timeoutMs = this.#config.dispatch.agentHoldTimeoutMs + return { + kind: 'agents', + sinceAtMs: record.heldSinceAtMs, + timeoutMs, + dueAtMs: record.heldSinceAtMs + timeoutMs, + } + } + // Only a row that is actually holding a slot is worth reaping; a `queued` + // or `waiting-for-human` row costs nothing and may wait indefinitely. + if (record.slotHeldSinceAtMs === undefined || !dispatchPhaseOccupiesSlot(record.lifecyclePhase)) { + return undefined + } + const timeoutMs = this.#config.dispatch.agentlessHoldTimeoutMs + return { + kind: 'agentless', + sinceAtMs: record.slotHeldSinceAtMs, + timeoutMs, + dueAtMs: record.slotHeldSinceAtMs + timeoutMs, + } + } + #scheduleHeldAgentDeadline(record: InFlightIssue): void { - if (this.#stopping || record.dryRun || record.heldSinceAtMs === undefined || record.agents.size === 0) return - const dueAtMs = record.heldSinceAtMs + this.#config.dispatch.agentHoldTimeoutMs + if (this.#stopping) return + const deadline = this.#holdDeadline(record) + if (!deadline) return + const dueAtMs = deadline.dueAtMs if ( this.#heldAgentDeadlineTimer && this.#heldAgentDeadlineDueAtMs !== undefined && @@ -5245,15 +5333,9 @@ export class FactoryLoop implements Factory { async #sweepHeldAgentDeadlines(): Promise { const nowMs = this.#clock.now() - const timeoutMs = this.#config.dispatch.agentHoldTimeoutMs for (const record of [...(await this.#batch()).inFlight]) { - const heldSinceAtMs = record.heldSinceAtMs - if ( - record.dryRun || - heldSinceAtMs === undefined || - record.agents.size === 0 || - nowMs - heldSinceAtMs < timeoutMs - ) continue + const deadline = this.#holdDeadline(record) + if (!deadline || nowMs < deadline.dueAtMs) continue const key = issueKey(record.issue) if (this.#abandonedDispatchReasons.has(key)) continue @@ -5266,26 +5348,46 @@ export class FactoryLoop implements Factory { await this.#finishDurableRelease(record, lifecycle.releaseReason) continue } + // Re-derive against the durable row before tearing anything down. The + // in-memory record can be a beat behind a placement that just + // succeeded in this process or a takeover in another, and the + // never-placed deadline exists precisely to catch rows nothing is + // moving — it must not be what ends a dispatch that just started + // moving (#303 must-not-fire). + const durable = this.#holdDeadline(inFlightRecordFromLifecycle(lifecycle)) + if (!durable || this.#clock.now() < durable.dueAtMs) continue if (!await this.#assertDispatchLifecycleOwner(record)) continue } - const heldForMs = Math.max(0, this.#clock.now() - heldSinceAtMs) + const agentless = deadline.kind === 'agentless' + const heldForMs = Math.max(0, this.#clock.now() - deadline.sinceAtMs) const details = { issue: record.issue.key, heldForMs, - holdTimeoutMs: timeoutMs, + holdTimeoutMs: deadline.timeoutMs, waitingForTerminalState: this.#config.terminalState, - reason: HELD_PAST_DEADLINE_RELEASE_REASON, + reason: agentless ? AGENTLESS_SLOT_PAST_DEADLINE_RELEASE_REASON : HELD_PAST_DEADLINE_RELEASE_REASON, agents: [...record.agents.keys()].sort(), + ...(agentless ? { phase: record.lifecyclePhase } : {}), } - this.#logger.warn?.('[factory] releasing agents held past deadline', details) - await this.#abandonStuckDispatch(record, HELD_PAST_DEADLINE_RELEASE_REASON) + this.#logger.warn?.( + agentless + ? '[factory] releasing a dispatch lifecycle that never placed an agent' + : '[factory] releasing agents held past deadline', + details, + ) + await this.#abandonStuckDispatch(record, details.reason) const lifecycle = this.#usesDurableDispatchLifecycle() ? await this.#state.getDispatchLifecycle(this.#workspaceId, key) : undefined if (!lifecycle || isTerminalDispatchLifecycle(lifecycle)) { - this.#increment('heldPastDeadlineReleases') - this.#logger.warn?.('[factory] released agents held past deadline', details) + this.#increment(agentless ? 'agentlessSlotPastDeadlineReleases' : 'heldPastDeadlineReleases') + this.#logger.warn?.( + agentless + ? '[factory] released a dispatch lifecycle that never placed an agent' + : '[factory] released agents held past deadline', + details, + ) } } } @@ -5724,6 +5826,13 @@ export class FactoryLoop implements Factory { telemetry: { cancellationReason?: FactoryCloudCancellationReasonV1 } = {}, ): Promise { record.lifecyclePhase = phase + // Mirror what the store stamps, so the reaper's never-placed clock is + // readable from the in-memory record between durable reads (#303). The + // store owns the authoritative value and re-applies the babysitter-handoff + // half of the predicate; here `agents.size === 0` for every record this + // anchor is ever consulted for, and a handed-off row has babysitters. + if (dispatchPhaseOccupiesSlot(phase)) record.slotHeldSinceAtMs ??= this.#clock.now() + else record.slotHeldSinceAtMs = undefined if (record.dryRun || !this.#usesDurableDispatchLifecycle()) return true if (isTerminalDispatchPhase(phase)) await this.#drainAgentUsage() const key = issueKey(record.issue) @@ -5820,29 +5929,122 @@ export class FactoryLoop implements Factory { this.#dispatchTerminalWaiters.delete(key) } - #scheduleDispatchLifecycleRetry(record: InFlightIssue): void { + /** + * Re-arm delay for a capacity wait: 1 s doubling to a 30 s ceiling (#303). + * + * Only the capacity path backs off. An ownership wait is already bounded by + * `DISPATCH_LIFECYCLE_LEASE_MS`, and every other failure is a real error + * whose fast retry is the recovery. A capacity wait has no bound at all — + * it ends when some other lifecycle terminates, which may be hours away or, + * before this fix, never. + */ + #capacityRetryDelayMs(attempts: number): number { + return Math.min( + DISPATCH_LIFECYCLE_RETRY_MS * 2 ** Math.max(0, attempts - 1), + DISPATCH_LIFECYCLE_RETRY_MAX_MS, + ) + } + + /** Issue keys currently holding a `batchSize` slot, for operator surfaces. */ + #dispatchSlotOccupants(): FactoryDispatchSlotOccupant[] { + return (this.#batchView?.inFlight ?? []) + .filter((record) => !record.dryRun && dispatchPhaseOccupiesSlot(record.lifecyclePhase)) + .map((record) => ({ + issue: record.issue.key, + ...(record.lifecyclePhase ? { phase: record.lifecyclePhase } : {}), + agents: record.agents.size, + ...(record.heldSinceAtMs !== undefined + ? { heldForMs: Math.max(0, this.#clock.now() - record.heldSinceAtMs) } + : {}), + ...(record.slotHeldSinceAtMs !== undefined + ? { slotHeldForMs: Math.max(0, this.#clock.now() - record.slotHeldSinceAtMs) } + : {}), + })) + .sort((left, right) => left.issue.localeCompare(right.issue)) + } + + /** + * Batch occupancy as an operator-readable fact (#303). + * + * Before this, a full batch was visible only as the *absence* of dispatch: + * `readinessReconcile` stayed green, `consecutiveFailures` stayed 0, and the + * one capacity log had fired hours earlier. Publishing occupancy is what + * turns "nothing is being dispatched" into a question an operator can answer + * without reading the state document. + */ + #dispatchCapacityStatus(): FactoryDispatchCapacityStatus { + const nowMs = this.#clock.now() + const occupants = this.#dispatchSlotOccupants() + const waits = [...this.#dispatchLifecycleCapacityWaits.entries()] + const longestWaitMs = waits.length === 0 + ? undefined + : Math.max(...waits.map(([, wait]) => Math.max(0, nowMs - wait.sinceAtMs))) + return { + batchSize: this.#config.batchSize, + active: occupants.length, + waiting: waits.length, + waitWarnMs: this.#config.dispatch.capacityWaitWarnMs, + ...(longestWaitMs !== undefined ? { longestWaitMs } : {}), + ...(occupants.length > 0 ? { occupants } : {}), + ...(waits.length > 0 + ? { + waitingIssues: waits + .sort(([, left], [, right]) => left.sinceAtMs - right.sinceAtMs) + .map(([key]) => key), + } + : {}), + } + } + + #recordDispatchCapacityWait(record: InFlightIssue, key: string): number { + const nowMs = this.#clock.now() + let wait = this.#dispatchLifecycleCapacityWaits.get(key) + if (!wait) { + wait = { sinceAtMs: nowMs, attempts: 0 } + this.#dispatchLifecycleCapacityWaits.set(key, wait) + this.#increment('dispatchLifecycleCapacityWaits') + } + wait.attempts += 1 + const retryMs = this.#capacityRetryDelayMs(wait.attempts) + const waitedMs = Math.max(0, nowMs - wait.sinceAtMs) + // Escalate on every backoff step, then once a minute after the delay + // caps. The old behaviour logged once per key and went silent forever, + // which is what made a 14-hour dispatch outage look like an idle Factory. + const stepChanged = wait.lastLoggedRetryMs !== retryMs + const overdue = wait.lastLoggedAtMs === undefined || + nowMs - wait.lastLoggedAtMs >= DISPATCH_LIFECYCLE_CAPACITY_WAIT_LOG_MS + if (stepChanged || overdue) { + wait.lastLoggedAtMs = nowMs + wait.lastLoggedRetryMs = retryMs + this.#logger.warn?.('[factory] durable dispatch is queued for batch capacity; retries remain active', { + issue: record.issue.key, + retryMs, + attempts: wait.attempts, + waitedMs, + batchSize: this.#config.batchSize, + occupiedBy: this.#dispatchSlotOccupants().map((occupant) => occupant.issue), + }) + } + return retryMs + } + + #scheduleDispatchLifecycleRetry(record: InFlightIssue, delayMs = DISPATCH_LIFECYCLE_RETRY_MS): void { const key = issueKey(record.issue) if (this.#stopping || this.#dispatchLifecycleRetryTimers.has(key)) return const timer = setTimeout(() => { this.#dispatchLifecycleRetryTimers.delete(key) const drive = this.#driveDispatchLifecycle(key) .then(() => { - this.#dispatchLifecycleCapacityWaitLogged.delete(key) + this.#dispatchLifecycleCapacityWaits.delete(key) this.#dispatchLifecycleOwnershipWaitLogged.delete(key) }) .catch((error) => { + let nextDelayMs = DISPATCH_LIFECYCLE_RETRY_MS if (error instanceof DispatchLifecycleCapacityError) { this.#dispatchLifecycleOwnershipWaitLogged.delete(key) - if (!this.#dispatchLifecycleCapacityWaitLogged.has(key)) { - this.#dispatchLifecycleCapacityWaitLogged.add(key) - this.#increment('dispatchLifecycleCapacityWaits') - this.#logger.warn?.('[factory] durable dispatch is queued for batch capacity; retries remain active', { - issue: record.issue.key, - retryMs: DISPATCH_LIFECYCLE_RETRY_MS, - }) - } + nextDelayMs = this.#recordDispatchCapacityWait(record, key) } else if (error instanceof DispatchLifecycleOwnedElsewhereError) { - this.#dispatchLifecycleCapacityWaitLogged.delete(key) + this.#dispatchLifecycleCapacityWaits.delete(key) if (!this.#dispatchLifecycleOwnershipWaitLogged.has(key)) { this.#dispatchLifecycleOwnershipWaitLogged.add(key) this.#increment('dispatchLifecycleOwnershipWaits') @@ -5855,18 +6057,18 @@ export class FactoryLoop implements Factory { }) } } else { - this.#dispatchLifecycleCapacityWaitLogged.delete(key) + this.#dispatchLifecycleCapacityWaits.delete(key) this.#dispatchLifecycleOwnershipWaitLogged.delete(key) this.#logger.warn?.('[factory] durable dispatch lifecycle retry failed', { issue: record.issue.key, error: describeError(error).errorMessage, }) } - this.#scheduleDispatchLifecycleRetry(record) + this.#scheduleDispatchLifecycleRetry(record, nextDelayMs) }) .finally(() => this.#dispatchLifecycleDrives.delete(drive)) this.#dispatchLifecycleDrives.add(drive) - }, DISPATCH_LIFECYCLE_RETRY_MS) + }, delayMs) this.#dispatchLifecycleRetryTimers.set(key, timer) } @@ -7494,6 +7696,7 @@ export class FactoryLoop implements Factory { registryPath, eventListener: this.#eventListenerStatus(), readinessReconcile: this.#readinessReconcileStatus(), + dispatchCapacity: this.#dispatchCapacityStatus(), fleetControlPlane: this.#fleetControlPlane.status(), } // The deployed container serves `/healthz` straight out of this file and @@ -7943,10 +8146,13 @@ export class FactoryLoop implements Factory { async #teardownFailedDispatchWorktrees( handoffs: RegistryHandoffAgent[], releaseReason = 'dispatch failed', + opts: { skipNeverPlacedAgents?: boolean } = {}, ): Promise { if (!this.#worktrees || !handoffs.some((handoff) => handoff.worktree)) return false const failed = await this.#releaseAndTerminateAgents( - handoffs.map((handoff) => [handoff.name, handoff.tracked]), + handoffs + .filter((handoff) => !opts.skipNeverPlacedAgents || handoff.tracked.result !== undefined) + .map((handoff) => [handoff.name, handoff.tracked]), releaseReason, 'completion', ) @@ -9244,7 +9450,14 @@ export class FactoryLoop implements Factory { this.#scheduleAbandonedDispatchRetry(record, reason) return } - const agents = [...record.agents] + // A never-placed record carries specs, not workers: `recordPlanned` writes + // the spec before the spawn returns, so a dispatch that died mid-spawn + // leaves a name the broker never issued. Releasing one fails, which fails + // the whole cleanup and re-arms the abandon retry forever — turning the + // #303 reap into a second, quieter wedge. Their worktrees are still torn + // down below. + const neverPlaced = reason === AGENTLESS_SLOT_PAST_DEADLINE_RELEASE_REASON + const agents = [...record.agents].filter(([, tracked]) => !neverPlaced || tracked.result !== undefined) for (const [agentName, tracked] of agents) { if (!heldPastDeadline && tracked.spec.role === 'implementer') continue this.#fleet.markAgentTerminal?.( @@ -9266,7 +9479,11 @@ export class FactoryLoop implements Factory { const failed = await this.#releaseAndTerminateAgents(nonWorktreeAgents, agentReleaseReason, 'completion') cleanupComplete = failed.length === 0 } - cleanupComplete = await this.#teardownFailedDispatchWorktrees(worktreeHandoffs, agentReleaseReason) && cleanupComplete + cleanupComplete = await this.#teardownFailedDispatchWorktrees( + worktreeHandoffs, + agentReleaseReason, + { skipNeverPlacedAgents: neverPlaced }, + ) && cleanupComplete } else if (agents.length > 0) { const failed = await this.#releaseAndTerminateAgents(agents, agentReleaseReason, 'completion') cleanupComplete = failed.length === 0 @@ -19739,6 +19956,7 @@ const lifecycleFromInFlightRecord = ( ...(releaseReason ? { releaseReason } : {}), ...(cost ? { cost: structuredClone(cost) } : {}), ...(record.heldSinceAtMs !== undefined ? { heldSinceAtMs: record.heldSinceAtMs } : {}), + ...(record.slotHeldSinceAtMs !== undefined ? { slotHeldSinceAtMs: record.slotHeldSinceAtMs } : {}), updatedAtMs, }) @@ -19757,10 +19975,14 @@ const inFlightRecordFromLifecycle = (lifecycle: DispatchLifecycle): InFlightIssu result: lifecycle.result ? structuredClone(lifecycle.result) : undefined, ...(lifecycle.dispatchClaim ? { dispatchClaim: { ...lifecycle.dispatchClaim } } : {}), heldSinceAtMs: lifecycle.heldSinceAtMs ?? ( - lifecycle.agents.some((agent) => agent.releasedAtMs === undefined) + // A live placement the durable row predates the `heldSinceAtMs` field for. + // `tracked.result` is what distinguishes a placement from a spec that + // `recordPlanned` wrote and no spawn ever answered (#303). + lifecycle.agents.some((agent) => agent.releasedAtMs === undefined && agent.tracked.result !== undefined) ? lifecycle.updatedAtMs : undefined ), + slotHeldSinceAtMs: lifecycle.slotHeldSinceAtMs, lifecyclePhase: lifecycle.phase, }) diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 36645b24..ffe0fe6a 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -32,6 +32,80 @@ function heartbeat(overrides: Partial = {}): FactoryLoopHe } } +describe('dispatch capacity health (#303)', () => { + const capacity = (overrides: Partial> = {}) => heartbeat({ + dispatchCapacity: { + batchSize: 1, + active: 1, + waiting: 3, + waitWarnMs: 30 * 60_000, + longestWaitMs: 6 * 60 * 60_000, + occupants: [{ issue: 'AR-303', phase: 'dispatching', agents: 0, slotHeldForMs: 13 * 60 * 60_000 }], + waitingIssues: ['AR-304', 'AR-305', 'AR-306'], + ...overrides, + }, + }) + + it('reports a long capacity wait as a dispatch-gating degradation', () => { + const health = publicHealthFromHeartbeat(capacity(), { nowMs: BOOT_MS + 1_000 }) + + expect(health.dispatchCapacity).toEqual({ + state: 'stalled', + batchSize: 1, + active: 1, + waiting: 3, + waitWarnMs: 30 * 60_000, + longestWaitMs: 6 * 60 * 60_000, + agentlessOccupants: 1, + }) + expect(health.degradedSubsystems).toContain('dispatchCapacity') + expect(health.status).toBe('degraded') + // Liveness must not move: recycling the container would destroy the + // evidence of the wedge and carry the durable lock into the replacement. + expect(health.ok).toBe(true) + }) + + it('keeps issue keys behind the authenticated surface', () => { + const health = publicHealthFromHeartbeat(capacity(), { nowMs: BOOT_MS + 1_000 }) + + expect(JSON.stringify(health)).not.toContain('AR-30') + }) + + it('treats an ordinary full batch as healthy backpressure', () => { + const health = publicHealthFromHeartbeat( + capacity({ longestWaitMs: 60_000 }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.dispatchCapacity?.state).toBe('waiting') + expect(health.degradedSubsystems).not.toContain('dispatchCapacity') + expect(health.status).toBe('ok') + }) + + it('re-derives the state when a remote record carries an unrecognised one', () => { + const normalized = normalizePublicHealth({ + ...publicHealthFromHeartbeat(capacity(), { nowMs: BOOT_MS + 1_000 }), + dispatchCapacity: { + state: 'catastrophically-fine', + batchSize: 1, + active: 1, + waiting: 3, + waitWarnMs: 30 * 60_000, + longestWaitMs: 6 * 60 * 60_000, + }, + }) + + expect(normalized?.dispatchCapacity?.state).toBe('stalled') + }) + + it('omits the block entirely for an instance that predates it', () => { + const health = publicHealthFromHeartbeat(heartbeat(), { nowMs: BOOT_MS + 1_000 }) + + expect(health.dispatchCapacity).toBeUndefined() + expect(health.degradedSubsystems).not.toContain('dispatchCapacity') + }) +}) + describe('publicHealthFromHeartbeat (#295)', () => { it('carries the failure count and an allowlisted error class', () => { const health = publicHealthFromHeartbeat( diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index 69bd7568..f37a4a1d 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -1,8 +1,11 @@ import { telemetryErrorClassName } from '../observability/error-class.js' import type { FleetControlPlaneStatus } from '../fleet/control-plane-circuit' +import { DEFAULT_CAPACITY_WAIT_WARN_MS } from '../config/schema' import type { + FactoryDispatchCapacityStatus, FactoryEventListenerStatus, FactoryLoopHeartbeat, + FactoryPublicDispatchCapacityHealth, FactoryPublicEventListenerHealth, FactoryPublicFleetControlPlaneHealth, FactoryPublicHealth, @@ -71,7 +74,17 @@ const FLEET_CONTROL_PLANE_STATES: readonly FleetControlPlaneStatus['state'][] = ] /** Subsystems whose degradation stops issues from being dispatched. */ -const DISPATCH_GATING_SUBSYSTEMS = ['readinessReconcile', 'eventListener', 'fleetControlPlane'] as const +const DISPATCH_GATING_SUBSYSTEMS = [ + 'readinessReconcile', + 'eventListener', + 'fleetControlPlane', + // #303. A full batch stops dispatch exactly as hard as a failing sweep, and + // is the only one of the four that fails without anything throwing: nothing + // increments a failure counter, nothing writes `lastError`, and the wait + // logged once and went quiet. It belongs on this list because "why is + // nothing being dispatched" is the question the list exists to answer. + 'dispatchCapacity', +] as const const finiteNumber = (value: unknown): number | undefined => typeof value === 'number' && Number.isFinite(value) ? value : undefined @@ -149,6 +162,41 @@ const boundedText = (value: string): string => // C1 range as escape introducers. value.replace(/[\u0000-\u001F\u007F-\u009F]+/gu, ' ').trim().slice(0, 300) +const DISPATCH_CAPACITY_STATES: readonly FactoryPublicDispatchCapacityHealth['state'][] = [ + 'healthy', + 'waiting', + 'stalled', +] + +/** + * Capacity state from the numbers that produced it. + * + * Used by the writer, and as the fallback when a record arrives carrying an + * unrecognised state string. Falling back to `healthy` there would hide the + * exact condition this block exists to report, and unlike the readiness + * derivations this one needs no clock — `longestWaitMs` is a duration the + * writer already measured. + */ +const deriveDispatchCapacityState = ( + waiting: number, + longestWaitMs: number | undefined, + warnMs: number, +): FactoryPublicDispatchCapacityHealth['state'] => waiting === 0 + ? 'healthy' + : longestWaitMs !== undefined && longestWaitMs > warnMs + ? 'stalled' + : 'waiting' + +const dispatchCapacityState = ( + value: unknown, + waiting: number, + longestWaitMs: number | undefined, + warnMs: number, +): FactoryPublicDispatchCapacityHealth['state'] => + typeof value === 'string' && (DISPATCH_CAPACITY_STATES as readonly string[]).includes(value) + ? value as FactoryPublicDispatchCapacityHealth['state'] + : deriveDispatchCapacityState(waiting, longestWaitMs, warnMs) + const enumValue = (value: unknown, allowed: readonly T[]): T | 'unknown' => typeof value === 'string' && (allowed as readonly string[]).includes(value) ? value as T : 'unknown' @@ -242,6 +290,33 @@ function readinessReconcileHealth( } } +/** + * Batch occupancy, redacted (#303). + * + * Issue keys stay behind the authenticated surface — they carry customer + * project and repository names — so the public record carries counts and + * durations only. `agentlessOccupants` is the wedge signature: a slot held by + * a lifecycle that never placed an agent cannot make progress on its own. + */ +function dispatchCapacityHealth( + status: FactoryDispatchCapacityStatus, +): FactoryPublicDispatchCapacityHealth { + const waiting = counter(status.waiting) + const longestWaitMs = finiteNumber(status.longestWaitMs) + const warnMs = positiveNumber(status.waitWarnMs) ?? DEFAULT_CAPACITY_WAIT_WARN_MS + const agentlessOccupants = (status.occupants ?? []) + .filter((occupant) => counter(occupant.agents) === 0).length + return { + state: deriveDispatchCapacityState(waiting, longestWaitMs, warnMs), + batchSize: counter(status.batchSize), + active: counter(status.active), + waiting, + waitWarnMs: warnMs, + ...optionalDuration('longestWaitMs', longestWaitMs), + ...(agentlessOccupants > 0 ? { agentlessOccupants } : {}), + } +} + function fleetControlPlaneHealth( status: FleetControlPlaneStatus, ): FactoryPublicFleetControlPlaneHealth { @@ -292,6 +367,9 @@ export function publicHealthFromHeartbeat( const fleetControlPlane = heartbeat.fleetControlPlane ? fleetControlPlaneHealth(heartbeat.fleetControlPlane) : undefined + const dispatchCapacity = heartbeat.dispatchCapacity + ? dispatchCapacityHealth(heartbeat.dispatchCapacity) + : undefined const eventListener: FactoryPublicEventListenerHealth | undefined = heartbeat.eventListener // Only the state. `reason` is assembled free text and stays behind the // authenticated surface. @@ -313,6 +391,13 @@ export function publicHealthFromHeartbeat( // from either. Both mean dispatch is not admitting work normally. return fleetControlPlane !== undefined && fleetControlPlane.state !== 'closed' } + if (name === 'dispatchCapacity') { + // `waiting` alone is ordinary backpressure and stays green: a batch is + // supposed to fill up. Only a wait past the configured threshold — which + // a deployment running multi-hour issues should raise rather than + // silence — is a degradation. + return dispatchCapacity !== undefined && dispatchCapacity.state === 'stalled' + } // Review follow-up on #300 (P2, codex): `starting` is what a live daemon // reports before `#startLiveSubscription` installs the subscription. No // listener is registered, and startup can be lengthy, so anything short of @@ -359,6 +444,7 @@ export function publicHealthFromHeartbeat( ...(readinessReconcile ? { readinessReconcile } : {}), ...(eventListener ? { eventListener } : {}), ...(fleetControlPlane ? { fleetControlPlane } : {}), + ...(dispatchCapacity ? { dispatchCapacity } : {}), } } @@ -383,6 +469,7 @@ export function normalizePublicHealth(value: unknown): FactoryPublicHealth | und const readiness = plainRecord(record.readinessReconcile) const listener = plainRecord(record.eventListener) const fleet = plainRecord(record.fleetControlPlane) + const capacity = plainRecord(record.dispatchCapacity) const degradedSubsystems = Array.isArray(record.degradedSubsystems) ? DISPATCH_GATING_SUBSYSTEMS.filter((name) => (record.degradedSubsystems as unknown[]).includes(name)) : [] @@ -430,5 +517,23 @@ export function normalizePublicHealth(value: unknown): FactoryPublicHealth | und }, } : {}), + ...(capacity + ? { + dispatchCapacity: { + state: dispatchCapacityState( + capacity.state, + counter(capacity.waiting), + optionalDuration('longestWaitMs', capacity.longestWaitMs).longestWaitMs, + positiveNumber(capacity.waitWarnMs) ?? DEFAULT_CAPACITY_WAIT_WARN_MS, + ), + batchSize: counter(capacity.batchSize), + active: counter(capacity.active), + waiting: counter(capacity.waiting), + waitWarnMs: positiveNumber(capacity.waitWarnMs) ?? DEFAULT_CAPACITY_WAIT_WARN_MS, + ...optionalDuration('longestWaitMs', capacity.longestWaitMs), + ...optionalCount('agentlessOccupants', capacity.agentlessOccupants), + }, + } + : {}), } } diff --git a/src/ports/state.ts b/src/ports/state.ts index 80c2b130..2d767cfc 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -305,6 +305,15 @@ export type DispatchLifecycle = { lease?: DispatchLifecycleLease /** First successful agent placement for this active team generation. */ heldSinceAtMs?: number + /** + * When this row started occupying a `dispatch.batchSize` slot (#303). + * + * `heldSinceAtMs` only exists once a placement succeeds, and `updatedAtMs` + * is bumped by every lease renewal, so neither can bound a lifecycle that + * took the slot and never got an agent. This is that bound; it is cleared + * whenever the row stops occupying a slot. + */ + slotHeldSinceAtMs?: number updatedAtMs: number } diff --git a/src/state/dispatch-lifecycle-slot.ts b/src/state/dispatch-lifecycle-slot.ts new file mode 100644 index 00000000..90d22c35 --- /dev/null +++ b/src/state/dispatch-lifecycle-slot.ts @@ -0,0 +1,58 @@ +import { githubRepositoriesMatch } from '../github/repo-identity' +import type { DispatchLifecycle, DispatchLifecyclePhase } from '../ports/state' + +/** + * Batch-slot accounting for durable dispatch lifecycles. + * + * `dispatch.batchSize` is enforced by counting the rows in a slot-occupying + * phase, so this predicate is what decides whether another issue can ever be + * promoted out of `queued`. It used to be copy-pasted into both state stores; + * #303 added a second reader (the reaper's never-placed deadline), and three + * copies of a predicate that gates all dispatch is one too many. + */ + +/** Phases whose lifecycle counts against `dispatch.batchSize`. */ +export const dispatchPhaseOccupiesSlot = (phase: DispatchLifecyclePhase | undefined): boolean => + phase !== undefined && + phase !== 'queued' && + phase !== 'waiting-for-human' && + phase !== 'releasing' && + phase !== 'complete' && + phase !== 'abandoned' + +export const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => { + const implementerRepos = [...new Set(lifecycle.decision.implementers.map((spec) => spec.repo))] + if (implementerRepos.length === 0) return false + const babysitterRepos = lifecycle.agents + .filter((agent) => agent.tracked.spec.role === 'babysitter') + .map((agent) => agent.tracked.spec.ownedPullRequest?.repo) + .filter((repo): repo is string => Boolean(repo)) + return implementerRepos.every((repo) => babysitterRepos.some((ownedRepo) => + githubRepositoriesMatch(repo, ownedRepo))) +} + +export const dispatchLifecycleOccupiesSlot = (lifecycle: DispatchLifecycle): boolean => + dispatchPhaseOccupiesSlot(lifecycle.phase) && !dispatchLifecycleHandedOffToBabysitters(lifecycle) + +/** + * Stamp the wall-clock instant this row took its batch slot (#303). + * + * `updatedAtMs` cannot serve as this clock: `renewDispatchLifecycle` bumps it + * every `DISPATCH_LIFECYCLE_RENEW_MS`, so a permanently wedged row looks + * freshly touched forever. `heldSinceAtMs` cannot either — it is only set once + * a placement succeeds, and the whole point of the defect is a row that never + * got one. This is the only anchor a never-placed occupant has, so it is + * carried forward across saves and cleared the moment the row stops occupying + * a slot. + */ +export const stampDispatchLifecycleSlot = ( + lifecycle: DispatchLifecycle, + previous: DispatchLifecycle | undefined, + nowMs: number, +): void => { + if (!dispatchLifecycleOccupiesSlot(lifecycle)) { + delete lifecycle.slotHeldSinceAtMs + return + } + lifecycle.slotHeldSinceAtMs ??= previous?.slotHeldSinceAtMs ?? nowMs +} diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 8e99eddc..ae5d6a78 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -4,7 +4,6 @@ import { dirname, join } from 'node:path' import lockfile from 'proper-lockfile' -import { githubRepositoriesMatch } from '../github/repo-identity' import type { BabysitterGenerationRecord, BabysitterSessionState, @@ -24,6 +23,7 @@ import type { } from '../ports/state' import { InMemoryStateStore, type InMemoryStateStoreOptions } from './in-memory-state-store' import { matchingGithubLifecycleEntry } from './github-lifecycle-identity' +import { dispatchLifecycleOccupiesSlot, stampDispatchLifecycleSlot } from './dispatch-lifecycle-slot' import type { PersistedWorkspaceState, WatchStateDocument, @@ -259,6 +259,10 @@ export class DocumentStateStore extends InMemoryStateStore { : (lifecycle.lease?.epoch ?? 0) + 1 lifecycle.lease = { owner, epoch, leaseUntilMs: nowMs + leaseMs } lifecycle.updatedAtMs = nowMs + // Rows written before #303 carry no slot anchor. Claim is where a + // process first takes responsibility for one, so it is also where the + // never-placed clock starts for a pre-existing wedge. + stampDispatchLifecycleSlot(lifecycle, lifecycle, nowMs) await this.#persist(document) return { key, @@ -310,6 +314,7 @@ export class DocumentStateStore extends InMemoryStateStore { ) return false lifecycle.phase = 'dispatching' lifecycle.updatedAtMs = nowMs + stampDispatchLifecycleSlot(lifecycle, lifecycle, nowMs) await this.#persist(document) return true })) @@ -348,6 +353,7 @@ export class DocumentStateStore extends InMemoryStateStore { const next = cloneLifecycle(lifecycle) next.lease = { ...current.lease } next.updatedAtMs = nowMs + stampDispatchLifecycleSlot(next, current, nowMs) workspace!.dispatchLifecycles[key] = next await this.#persist(document) return true @@ -1395,25 +1401,6 @@ const dispatchLifecycleLeaseMatches = ( const activeDispatchLifecycleCount = (lifecycles: Record, exceptKey?: string): number => Object.entries(lifecycles).filter(([key, lifecycle]) => key !== exceptKey && dispatchLifecycleOccupiesSlot(lifecycle)).length -const dispatchLifecycleOccupiesSlot = (lifecycle: DispatchLifecycle): boolean => - lifecycle.phase !== 'queued' && - lifecycle.phase !== 'waiting-for-human' && - lifecycle.phase !== 'releasing' && - lifecycle.phase !== 'complete' && - lifecycle.phase !== 'abandoned' && - !dispatchLifecycleHandedOffToBabysitters(lifecycle) - -const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => { - const implementerRepos = [...new Set(lifecycle.decision.implementers.map((spec) => spec.repo))] - if (implementerRepos.length === 0) return false - const babysitterRepos = lifecycle.agents - .filter((agent) => agent.tracked.spec.role === 'babysitter') - .map((agent) => agent.tracked.spec.ownedPullRequest?.repo) - .filter((repo): repo is string => Boolean(repo)) - return implementerRepos.every((repo) => babysitterRepos.some((ownedRepo) => - githubRepositoriesMatch(repo, ownedRepo))) -} - const emptyWorkspaceState = (): PersistedWorkspaceState => ({ githubIssueCommentWatches: {}, slackThreadWatches: {}, diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 542ee35f..afec7ebc 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -1,7 +1,6 @@ import { randomUUID } from 'node:crypto' import { BatchTracker } from '../orchestrator/batch-tracker' -import { githubRepositoriesMatch } from '../github/repo-identity' import type { BatchSnapshot, BabysitterGenerationRecord, @@ -24,6 +23,7 @@ import type { ClarificationReply, } from '../ports/state' import { matchingGithubLifecycleEntry } from './github-lifecycle-identity' +import { dispatchLifecycleOccupiesSlot, stampDispatchLifecycleSlot } from './dispatch-lifecycle-slot' type WorkspaceState = { batch: BatchTracker @@ -225,6 +225,9 @@ export class InMemoryStateStore implements StateStore { : (lifecycle.lease?.epoch ?? 0) + 1 lifecycle.lease = { owner, epoch, leaseUntilMs: nowMs + leaseMs } lifecycle.updatedAtMs = nowMs + // See FileStateStore#claimDispatchLifecycle: claim starts the #303 + // never-placed clock for rows that predate the field. + stampDispatchLifecycleSlot(lifecycle, lifecycle, nowMs) return { key, acquired: true, @@ -268,6 +271,7 @@ export class InMemoryStateStore implements StateStore { ) return false lifecycle.phase = 'dispatching' lifecycle.updatedAtMs = nowMs + stampDispatchLifecycleSlot(lifecycle, lifecycle, nowMs) return true } @@ -292,6 +296,7 @@ export class InMemoryStateStore implements StateStore { const next = cloneDispatchLifecycle(lifecycle) next.lease = { ...current.lease } next.updatedAtMs = nowMs + stampDispatchLifecycleSlot(next, current, nowMs) this.#workspace(workspaceId).dispatchLifecycles.set(key, next) return true } @@ -1030,25 +1035,6 @@ const compareConversationMessages = (left: ConversationMessage, right: Conversat const activeDispatchLifecycleCount = (lifecycles: Map, exceptKey?: string): number => [...lifecycles].filter(([key, lifecycle]) => key !== exceptKey && dispatchLifecycleOccupiesSlot(lifecycle)).length -const dispatchLifecycleOccupiesSlot = (lifecycle: DispatchLifecycle): boolean => - lifecycle.phase !== 'queued' && - lifecycle.phase !== 'waiting-for-human' && - lifecycle.phase !== 'releasing' && - lifecycle.phase !== 'complete' && - lifecycle.phase !== 'abandoned' && - !dispatchLifecycleHandedOffToBabysitters(lifecycle) - -const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => { - const implementerRepos = [...new Set(lifecycle.decision.implementers.map((spec) => spec.repo))] - if (implementerRepos.length === 0) return false - const babysitterRepos = lifecycle.agents - .filter((agent) => agent.tracked.spec.role === 'babysitter') - .map((agent) => agent.tracked.spec.ownedPullRequest?.repo) - .filter((repo): repo is string => Boolean(repo)) - return implementerRepos.every((repo) => babysitterRepos.some((ownedRepo) => - githubRepositoriesMatch(repo, ownedRepo))) -} - const cloneWaitingClarification = (record: WaitingClarification): WaitingClarification => structuredClone(record) diff --git a/src/state/watch-state-document.ts b/src/state/watch-state-document.ts index 79c53089..5dfb3f02 100644 --- a/src/state/watch-state-document.ts +++ b/src/state/watch-state-document.ts @@ -409,6 +409,7 @@ const validDispatchLifecycle = (value: unknown): value is DispatchLifecycle => i Array.isArray(value.agents) && value.agents.every(validDispatchLifecycleAgent) && Array.isArray(value.invocationIds) && value.invocationIds.every((id) => typeof id === 'string') && validNumber(value.updatedAtMs) && validOptionalNumber(value.heldSinceAtMs) && + validOptionalNumber(value.slotHeldSinceAtMs) && validOptionalString(value.releaseReason) && (value.lease === undefined || validLifecycleLease(value.lease)) && (value.result === undefined || validDispatchResult(value.result)) && diff --git a/src/types.ts b/src/types.ts index 14dc0b59..d54bd6d6 100644 --- a/src/types.ts +++ b/src/types.ts @@ -146,6 +146,8 @@ export interface FactoryLoopHeartbeat { registryPath?: string eventListener?: FactoryEventListenerStatus readinessReconcile?: FactoryReadinessReconcileStatus + /** Batch-slot admission: a full batch is why dispatch stops without failing (#303). */ + dispatchCapacity?: FactoryDispatchCapacityStatus /** Daemon-owned dispatch admission state; status readers must prefer this over a fresh local Factory instance. */ fleetControlPlane?: FleetControlPlaneStatus /** @@ -217,6 +219,58 @@ export interface FactoryPublicReadinessReconcileHealth { lastErrorClass?: string } +/** + * A lifecycle currently holding one of the `batchSize` slots (#303). + * + * `agents` and `slotHeldForMs` together are what separate ordinary + * backpressure from a wedge: a slot held for hours by a row that never placed + * an agent is the shape that produced a total dispatch outage. + */ +export interface FactoryDispatchSlotOccupant { + issue: string + phase?: DispatchLifecyclePhase + /** Placements recorded on the lifecycle, including planned-but-unspawned. */ + agents: number + /** Since the first successful placement, when there has been one. */ + heldForMs?: number + /** Since the row took the batch slot, whether or not it ever placed an agent. */ + slotHeldForMs?: number +} + +/** + * Batch admission as an operator-readable fact (#303). + * + * `promoteDispatchLifecycle` is a silent predicate: it returns `false`, never + * throws, and the caller swallows the result into a retry. A full batch was + * therefore indistinguishable from an idle Factory on every surface an + * operator could reach. This is that predicate, published. + */ +export interface FactoryDispatchCapacityStatus { + batchSize: number + /** Lifecycles occupying a slot right now. */ + active: number + /** Lifecycles waiting on capacity right now. */ + waiting: number + /** Wall-clock wait past which the wait is treated as dispatch-gating. */ + waitWarnMs: number + longestWaitMs?: number + occupants?: FactoryDispatchSlotOccupant[] + /** Issue keys waiting on capacity, longest wait first. */ + waitingIssues?: string[] +} + +/** Batch occupancy, redacted for the unauthenticated surface (#303). */ +export interface FactoryPublicDispatchCapacityHealth { + state: 'healthy' | 'waiting' | 'stalled' + batchSize: number + active: number + waiting: number + waitWarnMs: number + longestWaitMs?: number + /** Occupied slots that never placed an agent — the wedge signature. */ + agentlessOccupants?: number +} + export interface FactoryPublicEventListenerHealth { state: FactoryEventListenerStatus['state'] } @@ -269,6 +323,7 @@ export interface FactoryPublicHealth { readinessReconcile?: FactoryPublicReadinessReconcileHealth eventListener?: FactoryPublicEventListenerHealth fleetControlPlane?: FactoryPublicFleetControlPlaneHealth + dispatchCapacity?: FactoryPublicDispatchCapacityHealth } export interface FactoryInFlightRegistryAgent { @@ -416,6 +471,8 @@ export interface FactoryStatus { eventListener?: FactoryEventListenerStatus /** Periodic ready-issue backfill health as reported by the live daemon. */ readinessReconcile?: FactoryReadinessReconcileStatus + /** Batch-slot admission, including which lifecycles hold the slots (#303). */ + dispatchCapacity?: FactoryDispatchCapacityStatus /** Agents retained while their issue waits for its configured terminal state. */ heldAgents?: FactoryHeldAgent[] } From ac9b7b19a6d3a2392880bfd9a39ef7a7a18b4620 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 00:35:29 +0200 Subject: [PATCH 2/9] fix(orchestrator): make slot occupancy and release classification agree with admission (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings, all real: 1. `#dispatchSlotOccupants` filtered on phase alone, but admission also excludes a lifecycle whose every implementer repo has been handed to a babysitter. `active`, `occupants` and the capacity log's `occupiedBy` could therefore name slots that were not blocking promotion, and `active` could exceed `batchSize`. Records now ask the same predicate the state stores ask of a durable row, via a shared spec-shaped `dispatchHandedOffToBabysitters`. `#holdDeadline` and the record-side slot stamp use it too, so the orchestrator and the stores cannot disagree. (codex P2) 2. `agentlessOccupants` counted `agents === 0`, but `agents` counts specs: `recordPlanned` writes one before the spawn returns. The planned-before-spawn crash this PR exists for reports `agents: 1` with no placement, so `/healthz` and `factory diagnose` lost the wedge signature for one of the two target failure modes. Occupants now carry an explicit `placedAgents`, with the placement stamp as the fallback for a producer that does not send it. (codex P2, CodeRabbit major) 3. The sweep re-derived the deadline from the durable row and then classified from the stale in-memory one. When a placement landed durably, the release was labelled `agentless-slot-past-deadline`, the wrong counter moved, and — worst — `#abandonStuckDispatch` treated the row as never-placed and excluded its agents from the broker release, leaking live workers. Classification now uses the durable deadline whenever there is one. (CodeRabbit major) Each fix has a regression test verified to fail with the fix reverted. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 88 ++++++++++++++++++++++++++ src/orchestrator/factory.ts | 45 ++++++++++--- src/orchestrator/public-health.test.ts | 25 +++++++- src/orchestrator/public-health.ts | 10 ++- src/state/dispatch-lifecycle-slot.ts | 26 ++++++-- src/types.ts | 9 ++- 6 files changed, 185 insertions(+), 18 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 5a01d912..69237f38 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -10947,6 +10947,87 @@ describe('FactoryLoop', () => { }, 30_000) } + // #303 review (CodeRabbit): the sweep re-reads the durable row before tearing + // anything down, so it must also *classify* from it. Reading the stale + // in-memory record instead relabels a team that did run as never-placed — + // which excludes its agents from the release and leaks live workers. + it('classifies a release from the durable row when the in-flight record lags a placement (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-durable-classification-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const seed = createFactory(config({ batchSize: 1 }), { + mount: new FakeMountClient({ [issuePath(310)]: issueFile(310) }), + fleet: new RemoteLifecycleFleetClient(), + stateStore: state(), + triage: new StaticTriage(), + }) + const decision = await seed.triageIssue(parseLinearIssue(issuePath(310), issueFile(310))) + await seed.stop() + + const plannedSpec = decision.implementers[0]! + const key = issueKey(decision.issue) + const wedged: DispatchLifecycle = { + runId: 'lagging-run', + issue: { ...decision.issue }, + decision, + dryRun: false, + phase: 'dispatching', + agents: [{ name: plannedSpec.name, tracked: { spec: { ...plannedSpec } } }], + invocationIds: [], + updatedAtMs: 0, + } + // Claimed on the real clock so the never-placed deadline is still ahead + // while the durable row is patched below; the lease is already expired. + await state().claimDispatchLifecycle('factory-test', key, wedged, 'dead-owner', Date.now(), 1) + + const fleet = new RemoteLifecycleFleetClient() + const factory = createFactory(config({ + batchSize: 1, + dispatch: { agentHoldTimeoutMs: 1_000, agentlessHoldTimeoutMs: 5_000 }, + loop: { heartbeatPath: join(root, 'heartbeat.json'), registryPath: join(root, 'registry.json') }, + }), { + // AR-310 is unreadable here, so the resume driver keeps failing and + // `BatchTracker#restore` keeps handing back the same stale record. + mount: new FakeMountClient({}), + fleet, + stateStore: state(), + triage: new StaticTriage(), + }) + try { + await factory.start({ mode: 'dispatch-owner' }) + + // Another owner placed the agent and stamped the hold. Only the durable + // row knows; this process still holds the never-placed record. + const document = JSON.parse(await readFile(watchStatePath, 'utf8')) as { + workspaces: Record }> + } + const durable = document.workspaces['factory-test']!.dispatchLifecycles[key]! + durable.heldSinceAtMs = 0 + durable.agents[0]!.tracked.result = { + name: plannedSpec.name, + sessionRef: 'session-310', + node: 'sf-mini', + locality: 'remote', + } + await writeFile(watchStatePath, JSON.stringify(document)) + + await vi.waitFor(async () => expect(await state().getDispatchLifecycle('factory-test', key)) + .toMatchObject({ phase: 'abandoned', releaseReason: 'held-past-deadline' }), { timeout: 15_000 }) + expect(fleet.releases).toContainEqual({ name: plannedSpec.name, reason: 'held-past-deadline' }) + // The release reason and the release itself are the discriminators: read + // from the stale record this row is classified never-placed, which both + // relabels it and filters its agent out of the release entirely. + // `heldPastDeadlineReleases` is deliberately not asserted — the sweep + // only increments it when the abandon completes on that same pass, and + // this one does real agent teardown, so a slower pass finishes through + // `#scheduleAbandonedDispatchRetry` with the same durable outcome. + expect(factory.status().counters.agentlessSlotPastDeadlineReleases).toBeUndefined() + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 40_000) + // #303 must-not-fire control. The window between `promoteDispatchLifecycle` // and the first `recordSpawn` legitimately has zero agents. Reaping it on // sight would convert a wedge into a dispatch race. @@ -21724,6 +21805,13 @@ describe('FactoryLoop PR babysitter', () => { reason: 'superseded-pr-receipt', })) expect(factory.status().counters.supersededBabysittersReleased).toBe(1) + // #303 review (codex): admission stops counting a lifecycle once every + // implementer repo is babysat, so the reported occupancy has to stop + // counting it too. Reporting it would name a slot that is not blocking + // anything — and with batchSize 1, `active: 1` here is a claim the + // store's own predicate contradicts. + expect(factory.status().dispatchCapacity).toMatchObject({ batchSize: 1, active: 0, waiting: 0 }) + expect(factory.status().dispatchCapacity?.occupants).toBeUndefined() } finally { await factory.stop() } diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 3186c357..260aadd6 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -56,7 +56,7 @@ 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 { dispatchPhaseOccupiesSlot } from '../state/dispatch-lifecycle-slot' +import { dispatchHandedOffToBabysitters, dispatchPhaseOccupiesSlot } from '../state/dispatch-lifecycle-slot' import { containsExplicitIssueReference, containsIssueKey, factoryBranchBelongsToIssue } from '../issue-key-match' import { normalizeLogger, normalizeLogValue, setSafeErrorStack, stringifyLogValue } from '../logging' import { isInFactoryScope } from '../safety/factory-scope' @@ -5260,6 +5260,22 @@ export class FactoryLoop implements Factory { * the spec before the spawn returns, so a process that died mid-spawn leaves * an agent entry with no result and still no placement. */ + /** + * Does this in-flight record hold a `batchSize` slot right now? + * + * The same predicate the state stores apply to the durable row, asked of the + * in-memory one. Phase alone is not it: once every implementer repo has been + * handed to a babysitter, admission stops counting the lifecycle, so + * reporting it as an occupant would name slots that are not blocking + * anything (#303 review, codex). + */ + #recordOccupiesSlot(record: InFlightIssue): boolean { + return dispatchPhaseOccupiesSlot(record.lifecyclePhase) && !dispatchHandedOffToBabysitters( + record.decision.implementers, + [...record.agents.values()].map((tracked) => tracked.spec), + ) + } + #holdDeadline(record: InFlightIssue): { kind: 'agents' | 'agentless' sinceAtMs: number @@ -5278,7 +5294,7 @@ export class FactoryLoop implements Factory { } // Only a row that is actually holding a slot is worth reaping; a `queued` // or `waiting-for-human` row costs nothing and may wait indefinitely. - if (record.slotHeldSinceAtMs === undefined || !dispatchPhaseOccupiesSlot(record.lifecyclePhase)) { + if (record.slotHeldSinceAtMs === undefined || !this.#recordOccupiesSlot(record)) { return undefined } const timeoutMs = this.#config.dispatch.agentlessHoldTimeoutMs @@ -5337,6 +5353,12 @@ export class FactoryLoop implements Factory { const deadline = this.#holdDeadline(record) if (!deadline || nowMs < deadline.dueAtMs) continue + // The durable row wins the classification when there is one. The + // in-memory record can lag a placement made in another process, and + // reading the stale one would relabel a team that did run as + // never-placed — which also excludes its agents from the release below, + // leaking live workers (#303 review, CodeRabbit). + let effective = deadline const key = issueKey(record.issue) if (this.#abandonedDispatchReasons.has(key)) continue if (this.#usesDurableDispatchLifecycle()) { @@ -5356,15 +5378,16 @@ export class FactoryLoop implements Factory { // moving (#303 must-not-fire). const durable = this.#holdDeadline(inFlightRecordFromLifecycle(lifecycle)) if (!durable || this.#clock.now() < durable.dueAtMs) continue + effective = durable if (!await this.#assertDispatchLifecycleOwner(record)) continue } - const agentless = deadline.kind === 'agentless' - const heldForMs = Math.max(0, this.#clock.now() - deadline.sinceAtMs) + const agentless = effective.kind === 'agentless' + const heldForMs = Math.max(0, this.#clock.now() - effective.sinceAtMs) const details = { issue: record.issue.key, heldForMs, - holdTimeoutMs: deadline.timeoutMs, + holdTimeoutMs: effective.timeoutMs, waitingForTerminalState: this.#config.terminalState, reason: agentless ? AGENTLESS_SLOT_PAST_DEADLINE_RELEASE_REASON : HELD_PAST_DEADLINE_RELEASE_REASON, agents: [...record.agents.keys()].sort(), @@ -5828,10 +5851,9 @@ export class FactoryLoop implements Factory { record.lifecyclePhase = phase // Mirror what the store stamps, so the reaper's never-placed clock is // readable from the in-memory record between durable reads (#303). The - // store owns the authoritative value and re-applies the babysitter-handoff - // half of the predicate; here `agents.size === 0` for every record this - // anchor is ever consulted for, and a handed-off row has babysitters. - if (dispatchPhaseOccupiesSlot(phase)) record.slotHeldSinceAtMs ??= this.#clock.now() + // store still owns the authoritative value; this uses the same predicate so + // the two cannot disagree. + if (this.#recordOccupiesSlot(record)) record.slotHeldSinceAtMs ??= this.#clock.now() else record.slotHeldSinceAtMs = undefined if (record.dryRun || !this.#usesDurableDispatchLifecycle()) return true if (isTerminalDispatchPhase(phase)) await this.#drainAgentUsage() @@ -5948,11 +5970,14 @@ export class FactoryLoop implements Factory { /** Issue keys currently holding a `batchSize` slot, for operator surfaces. */ #dispatchSlotOccupants(): FactoryDispatchSlotOccupant[] { return (this.#batchView?.inFlight ?? []) - .filter((record) => !record.dryRun && dispatchPhaseOccupiesSlot(record.lifecyclePhase)) + .filter((record) => !record.dryRun && this.#recordOccupiesSlot(record)) .map((record) => ({ issue: record.issue.key, ...(record.lifecyclePhase ? { phase: record.lifecyclePhase } : {}), agents: record.agents.size, + // Specs, not workers: `recordPlanned` writes an entry before the spawn + // returns, so `agents > 0` is not proof of a placement (#303 review). + placedAgents: [...record.agents.values()].filter((tracked) => tracked.result !== undefined).length, ...(record.heldSinceAtMs !== undefined ? { heldForMs: Math.max(0, this.#clock.now() - record.heldSinceAtMs) } : {}), diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index ffe0fe6a..b7640e6b 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -40,7 +40,16 @@ describe('dispatch capacity health (#303)', () => { waiting: 3, waitWarnMs: 30 * 60_000, longestWaitMs: 6 * 60 * 60_000, - occupants: [{ issue: 'AR-303', phase: 'dispatching', agents: 0, slotHeldForMs: 13 * 60 * 60_000 }], + // `recordPlanned` wrote a spec and the spawn never returned, so the row + // reports an agent and no placement — the shape the projection must not + // mistake for a healthy occupant (#303 review). + occupants: [{ + issue: 'AR-303', + phase: 'dispatching', + agents: 1, + placedAgents: 0, + slotHeldForMs: 13 * 60 * 60_000, + }], waitingIssues: ['AR-304', 'AR-305', 'AR-306'], ...overrides, }, @@ -65,6 +74,20 @@ describe('dispatch capacity health (#303)', () => { expect(health.ok).toBe(true) }) + it('falls back to the placement stamp for a producer without placedAgents', () => { + const health = publicHealthFromHeartbeat( + capacity({ + occupants: [ + { issue: 'AR-303', phase: 'dispatching', agents: 1 } as never, + { issue: 'AR-307', phase: 'running', agents: 2, heldForMs: 90_000 } as never, + ], + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.dispatchCapacity?.agentlessOccupants).toBe(1) + }) + it('keeps issue keys behind the authenticated surface', () => { const health = publicHealthFromHeartbeat(capacity(), { nowMs: BOOT_MS + 1_000 }) diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index f37a4a1d..97a7a901 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -304,8 +304,16 @@ function dispatchCapacityHealth( const waiting = counter(status.waiting) const longestWaitMs = finiteNumber(status.longestWaitMs) const warnMs = positiveNumber(status.waitWarnMs) ?? DEFAULT_CAPACITY_WAIT_WARN_MS + // `agents` counts specs, and `recordPlanned` writes one before the spawn + // returns — so a dispatch that died mid-spawn reports `agents: 1` with no + // placement, and counting specs would drop the wedge signature for exactly + // the case #303 exists for. `heldForMs` is stamped only by a successful + // placement, and `placedAgents` says so outright; prefer the explicit count + // and fall back for a producer that does not send it. const agentlessOccupants = (status.occupants ?? []) - .filter((occupant) => counter(occupant.agents) === 0).length + .filter((occupant) => (occupant.placedAgents !== undefined + ? counter(occupant.placedAgents) === 0 + : finiteNumber(occupant.heldForMs) === undefined)).length return { state: deriveDispatchCapacityState(waiting, longestWaitMs, warnMs), batchSize: counter(status.batchSize), diff --git a/src/state/dispatch-lifecycle-slot.ts b/src/state/dispatch-lifecycle-slot.ts index 90d22c35..e83520dc 100644 --- a/src/state/dispatch-lifecycle-slot.ts +++ b/src/state/dispatch-lifecycle-slot.ts @@ -20,17 +20,33 @@ export const dispatchPhaseOccupiesSlot = (phase: DispatchLifecyclePhase | undefi phase !== 'complete' && phase !== 'abandoned' -export const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => { - const implementerRepos = [...new Set(lifecycle.decision.implementers.map((spec) => spec.repo))] +/** + * Spec-shaped so the orchestrator's in-flight records can ask the same + * question the stores ask of a durable row. A handed-off lifecycle stops + * counting against `batchSize`, so anything that reports occupancy — or bounds + * an occupant — has to agree with admission, or it reports slots that are not + * blocking anything (#303 review, codex). + */ +export const dispatchHandedOffToBabysitters = ( + implementers: ReadonlyArray<{ repo: string }>, + agentSpecs: ReadonlyArray<{ role?: string; ownedPullRequest?: { repo: string } }>, +): boolean => { + const implementerRepos = [...new Set(implementers.map((spec) => spec.repo))] if (implementerRepos.length === 0) return false - const babysitterRepos = lifecycle.agents - .filter((agent) => agent.tracked.spec.role === 'babysitter') - .map((agent) => agent.tracked.spec.ownedPullRequest?.repo) + const babysitterRepos = agentSpecs + .filter((spec) => spec.role === 'babysitter') + .map((spec) => spec.ownedPullRequest?.repo) .filter((repo): repo is string => Boolean(repo)) return implementerRepos.every((repo) => babysitterRepos.some((ownedRepo) => githubRepositoriesMatch(repo, ownedRepo))) } +export const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => + dispatchHandedOffToBabysitters( + lifecycle.decision.implementers, + lifecycle.agents.map((agent) => agent.tracked.spec), + ) + export const dispatchLifecycleOccupiesSlot = (lifecycle: DispatchLifecycle): boolean => dispatchPhaseOccupiesSlot(lifecycle.phase) && !dispatchLifecycleHandedOffToBabysitters(lifecycle) diff --git a/src/types.ts b/src/types.ts index d54bd6d6..76933750 100644 --- a/src/types.ts +++ b/src/types.ts @@ -229,8 +229,15 @@ export interface FactoryPublicReadinessReconcileHealth { export interface FactoryDispatchSlotOccupant { issue: string phase?: DispatchLifecyclePhase - /** Placements recorded on the lifecycle, including planned-but-unspawned. */ + /** Entries recorded on the lifecycle, including planned-but-unspawned. */ agents: number + /** + * Entries that actually reached a spawn result. + * + * `agents` counts specs: `BatchTracker#recordPlanned` writes one before the + * spawn returns. Zero here with a slot held is the wedge signature. + */ + placedAgents: number /** Since the first successful placement, when there has been one. */ heldForMs?: number /** Since the row took the batch slot, whether or not it ever placed an agent. */ From 247a2b0351c57943aad9abcd77c9a82816240559 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 00:50:00 +0200 Subject: [PATCH 3/9] fix(diagnostics): make the wedge signature mean wedged, not mid-spawn (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four more review findings, all valid: 1. `agentlessOccupants` counted "has no agent yet", which is every dispatch between promote and its first placement — `recordPlanned` writes the spec before the spawn returns, and a cloud spawn takes minutes. On a single-slot batch the field documented as the wedge signature would have read 1 continuously on a batch that was working. It now counts only occupants that never placed an agent *and* are already past `dispatch.agentlessHoldTimeoutMs`, the deadline that should have reaped them — a condition no healthy dispatch reaches. The deadline ships on the status and health records so the threshold is legible rather than implied. This is the same mistake the defect itself was: "no agent yet" is not "never going to have one", and the answer is the same deadline in both places. (cubic P2) 2. The projection threw on a non-array `occupants` or a null entry. It runs inside the heartbeat writer, where a throw costs the whole diagnostics block — the #300 failure mode. Now validated by construction, like every other field in that module. (cubic P2) 3. `factory diagnose` reported `longestWaitMs` as how long the slots had been occupied. It is the oldest queued issue's wait; the verdict now says so, and reports `active`/`batchSize` for occupancy. (cubic P2) 4. `factoryStatusWithMountHealth` fell back to a fresh local Factory's empty capacity view when a live daemon predates the field — publishing "the batch is free" from an instance that holds no lifecycles, which is the misreport the surrounding comment promises not to make. It now reports nothing there, matching readinessReconcile and fleetControlPlane. (cubic P2) Each has a regression test verified to fail with the fix reverted. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/cli/diagnose.test.ts | 10 ++++-- src/cli/diagnose.ts | 19 ++++++++--- src/cli/fleet.ts | 8 +++-- src/orchestrator/factory.test.ts | 7 +++- src/orchestrator/factory.ts | 1 + src/orchestrator/public-health.test.ts | 36 ++++++++++++++++---- src/orchestrator/public-health.ts | 46 ++++++++++++++++++++------ src/types.ts | 15 ++++++++- 8 files changed, 112 insertions(+), 30 deletions(-) diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index a4e19ba1..ac3f017c 100644 --- a/src/cli/diagnose.test.ts +++ b/src/cli/diagnose.test.ts @@ -211,6 +211,7 @@ describe('factory diagnose --deployed (#295)', () => { active: 1, waiting: 7, waitWarnMs: 1_800_000, + agentlessHoldTimeoutMs: 1_800_000, longestWaitMs: 46_800_000, agentlessOccupants: 1, }, @@ -223,9 +224,12 @@ describe('factory diagnose --deployed (#295)', () => { expect(code).not.toBe(0) const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string } expect(report.dispatching).toBe(false) - expect(report.verdict).toContain('batch slot') - expect(report.verdict).toContain('never placed an agent') - expect(report.verdict).toContain('7 issue(s) waiting') + expect(report.verdict).toContain('7 issue(s) have been waiting for batch capacity') + expect(report.verdict).toContain('1/1 slot(s) occupied') + expect(report.verdict).toContain('never placed an agent and are past the 30m 0s reap deadline') + // `longestWaitMs` is a queue wait, so the verdict must not present it as + // how long the slots have been held (#303 review, cubic). + expect(report.verdict).not.toContain('slot(s) have been occupied for') }) // The deployed container serves the block inside its heartbeat projection. diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index d7409807..a9e9d4d6 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -272,10 +272,16 @@ function verdictFor(diagnosis: Omit 0 - ? `. ${agentless} occupied slot(s) never placed an agent, so they cannot finish on their own` + ? `. ${agentless} occupied slot(s) never placed an agent and are past the ` + + `${formatDuration(capacity.agentlessHoldTimeoutMs)} reap deadline, so they cannot finish ` + + 'on their own' : '') + '. Pass --token to read /evidence for the issues holding the slots.', } @@ -468,10 +474,13 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st lines.push(` slots : ${capacity.active}/${capacity.batchSize} occupied`) lines.push(` waiting : ${capacity.waiting} issue(s)`) lines.push( - ` longest wait : ${formatDuration(capacity.longestWaitMs)} (warn past ${formatDuration(capacity.waitWarnMs)})`, + ` longest queue wait : ${formatDuration(capacity.longestWaitMs)} (warn past ${formatDuration(capacity.waitWarnMs)})`, ) if (capacity.agentlessOccupants !== undefined) { - lines.push(` never placed agent : ${capacity.agentlessOccupants} occupied slot(s)`) + lines.push( + ` unreaped wedges : ${capacity.agentlessOccupants} occupied slot(s) never placed an ` + + `agent and are past the ${formatDuration(capacity.agentlessHoldTimeoutMs)} reap deadline`, + ) } } lines.push(` eventListener : ${health.eventListener?.state ?? 'unknown'}`) diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 2e74468a..3d476688 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -1294,10 +1294,12 @@ async function factoryStatusWithMountHealth( ? heartbeat?.fleetControlPlane : observableStatus.fleetControlPlane // Same rule as readinessReconcile (#303): a live daemon owns the batch, and - // a fresh local Factory instance holds no lifecycles, so its empty view must - // not be reported as "the batch is free". + // a fresh local Factory instance holds no lifecycles. Falling back to that + // instance when a live daemon predates the field would publish its empty + // view as "the batch is free" — the exact misreport this exists to prevent, + // so an older daemon reports nothing here instead (#303 review, cubic). const dispatchCapacity = liveness.ok - ? heartbeat?.dispatchCapacity ?? observableStatus.dispatchCapacity + ? heartbeat?.dispatchCapacity : observableStatus.dispatchCapacity const eventListener = liveness.ok ? heartbeat?.eventListener ?? { diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 69237f38..e841752e 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -11131,7 +11131,12 @@ describe('FactoryLoop', () => { // Deliverable 3: batch occupancy is on the operator surface rather than // only in a log line that fires once. const capacity = factory.status().dispatchCapacity - expect(capacity).toMatchObject({ batchSize: 1, active: 1, waiting: 1 }) + expect(capacity).toMatchObject({ + batchSize: 1, + active: 1, + waiting: 1, + agentlessHoldTimeoutMs: 30 * 60_000, + }) expect(capacity?.longestWaitMs).toBeGreaterThan(0) expect(capacity?.occupants).toEqual([ expect.objectContaining({ issue: 'AR-306', phase: 'running', agents: 2 }), diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 260aadd6..ee7460a5 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -6009,6 +6009,7 @@ export class FactoryLoop implements Factory { active: occupants.length, waiting: waits.length, waitWarnMs: this.#config.dispatch.capacityWaitWarnMs, + agentlessHoldTimeoutMs: this.#config.dispatch.agentlessHoldTimeoutMs, ...(longestWaitMs !== undefined ? { longestWaitMs } : {}), ...(occupants.length > 0 ? { occupants } : {}), ...(waits.length > 0 diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index b7640e6b..082ee4ce 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -39,6 +39,7 @@ describe('dispatch capacity health (#303)', () => { active: 1, waiting: 3, waitWarnMs: 30 * 60_000, + agentlessHoldTimeoutMs: 30 * 60_000, longestWaitMs: 6 * 60 * 60_000, // `recordPlanned` wrote a spec and the spawn never returned, so the row // reports an agent and no placement — the shape the projection must not @@ -64,6 +65,7 @@ describe('dispatch capacity health (#303)', () => { active: 1, waiting: 3, waitWarnMs: 30 * 60_000, + agentlessHoldTimeoutMs: 30 * 60_000, longestWaitMs: 6 * 60 * 60_000, agentlessOccupants: 1, }) @@ -74,18 +76,39 @@ describe('dispatch capacity health (#303)', () => { expect(health.ok).toBe(true) }) - it('falls back to the placement stamp for a producer without placedAgents', () => { + // The wedge signature must not fire on a dispatch that is merely mid-spawn. + // `recordPlanned` writes the spec first, so every healthy dispatch has zero + // placements until its spawn returns — minutes, for a cloud placement. + it('does not count a dispatch still inside its spawn window as a wedge', () => { const health = publicHealthFromHeartbeat( capacity({ - occupants: [ - { issue: 'AR-303', phase: 'dispatching', agents: 1 } as never, - { issue: 'AR-307', phase: 'running', agents: 2, heldForMs: 90_000 } as never, - ], + occupants: [{ + issue: 'AR-307', + phase: 'dispatching', + agents: 1, + placedAgents: 0, + slotHeldForMs: 90_000, + }], }), { nowMs: BOOT_MS + 1_000 }, ) - expect(health.dispatchCapacity?.agentlessOccupants).toBe(1) + expect(health.dispatchCapacity?.agentlessOccupants).toBeUndefined() + // The wait itself is still reported: capacity is the outage signal here, + // and only the "cannot finish on its own" claim is withheld. + expect(health.dispatchCapacity?.state).toBe('stalled') + }) + + it('survives a corrupted occupants collection rather than dropping the block', () => { + for (const occupants of ['not-an-array', null, [null], [42], [{ placedAgents: 0 }]]) { + const health = publicHealthFromHeartbeat( + capacity({ occupants: occupants as never }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.dispatchCapacity).toMatchObject({ state: 'stalled', waiting: 3 }) + expect(health.dispatchCapacity?.agentlessOccupants).toBeUndefined() + } }) it('keeps issue keys behind the authenticated surface', () => { @@ -114,6 +137,7 @@ describe('dispatch capacity health (#303)', () => { active: 1, waiting: 3, waitWarnMs: 30 * 60_000, + agentlessHoldTimeoutMs: 30 * 60_000, longestWaitMs: 6 * 60 * 60_000, }, }) diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index 97a7a901..03e6cbcb 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -1,6 +1,6 @@ import { telemetryErrorClassName } from '../observability/error-class.js' import type { FleetControlPlaneStatus } from '../fleet/control-plane-circuit' -import { DEFAULT_CAPACITY_WAIT_WARN_MS } from '../config/schema' +import { DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS, DEFAULT_CAPACITY_WAIT_WARN_MS } from '../config/schema' import type { FactoryDispatchCapacityStatus, FactoryEventListenerStatus, @@ -290,6 +290,35 @@ function readinessReconcileHealth( } } +/** + * Occupied slots that will not free themselves. + * + * NOT "has no agent yet". `BatchTracker#recordPlanned` writes a spec before + * the spawn returns, so every healthy dispatch is agent-less for as long as + * its placement takes — minutes for a cloud spawn — and on a single-slot batch + * that is nearly always. Counting that would make the wedge signature read 1 + * continuously on a batch that is working, which is worse than not having the + * field (#303 review, cubic). The condition no healthy dispatch reaches is + * *never placed and already past the deadline that should have reaped it*. + * + * Defensive throughout: this runs inside the heartbeat writer, where a throw + * costs the whole diagnostics block, and the record may come from an older or + * corrupted producer (#303 review, cubic). + */ +function countAgentlessOccupants(occupants: unknown, reapMs: number): number { + if (!Array.isArray(occupants)) return 0 + return occupants.filter((entry) => { + const occupant = plainRecord(entry) + if (!occupant) return false + const slotHeldForMs = finiteNumber(occupant.slotHeldForMs) + // A producer that sends neither field cannot answer the question, and + // guessing "wedged" from an absence is how a false alarm gets published. + return finiteNumber(occupant.placedAgents) === 0 && + slotHeldForMs !== undefined && + slotHeldForMs > reapMs + }).length +} + /** * Batch occupancy, redacted (#303). * @@ -304,22 +333,15 @@ function dispatchCapacityHealth( const waiting = counter(status.waiting) const longestWaitMs = finiteNumber(status.longestWaitMs) const warnMs = positiveNumber(status.waitWarnMs) ?? DEFAULT_CAPACITY_WAIT_WARN_MS - // `agents` counts specs, and `recordPlanned` writes one before the spawn - // returns — so a dispatch that died mid-spawn reports `agents: 1` with no - // placement, and counting specs would drop the wedge signature for exactly - // the case #303 exists for. `heldForMs` is stamped only by a successful - // placement, and `placedAgents` says so outright; prefer the explicit count - // and fall back for a producer that does not send it. - const agentlessOccupants = (status.occupants ?? []) - .filter((occupant) => (occupant.placedAgents !== undefined - ? counter(occupant.placedAgents) === 0 - : finiteNumber(occupant.heldForMs) === undefined)).length + const reapMs = positiveNumber(status.agentlessHoldTimeoutMs) ?? DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS + const agentlessOccupants = countAgentlessOccupants(status.occupants, reapMs) return { state: deriveDispatchCapacityState(waiting, longestWaitMs, warnMs), batchSize: counter(status.batchSize), active: counter(status.active), waiting, waitWarnMs: warnMs, + agentlessHoldTimeoutMs: reapMs, ...optionalDuration('longestWaitMs', longestWaitMs), ...(agentlessOccupants > 0 ? { agentlessOccupants } : {}), } @@ -538,6 +560,8 @@ export function normalizePublicHealth(value: unknown): FactoryPublicHealth | und active: counter(capacity.active), waiting: counter(capacity.waiting), waitWarnMs: positiveNumber(capacity.waitWarnMs) ?? DEFAULT_CAPACITY_WAIT_WARN_MS, + agentlessHoldTimeoutMs: positiveNumber(capacity.agentlessHoldTimeoutMs) + ?? DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS, ...optionalDuration('longestWaitMs', capacity.longestWaitMs), ...optionalCount('agentlessOccupants', capacity.agentlessOccupants), }, diff --git a/src/types.ts b/src/types.ts index 76933750..3b93ee9c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -260,6 +260,8 @@ export interface FactoryDispatchCapacityStatus { waiting: number /** Wall-clock wait past which the wait is treated as dispatch-gating. */ waitWarnMs: number + /** Deadline after which a slot that never placed an agent should have been reaped. */ + agentlessHoldTimeoutMs: number longestWaitMs?: number occupants?: FactoryDispatchSlotOccupant[] /** Issue keys waiting on capacity, longest wait first. */ @@ -273,8 +275,19 @@ export interface FactoryPublicDispatchCapacityHealth { active: number waiting: number waitWarnMs: number + agentlessHoldTimeoutMs: number longestWaitMs?: number - /** Occupied slots that never placed an agent — the wedge signature. */ + /** + * Occupied slots that never placed an agent **and** are already past the + * deadline that should have reaped them. + * + * Deliberately not "has no agent yet": `recordPlanned` writes a spec before + * the spawn returns, so every healthy dispatch is agent-less for as long as + * its spawn takes — minutes, for a cloud placement. Counting that would make + * the wedge signature read 1 continuously on a single-slot batch that is + * working perfectly (#303 review, cubic). Past the deadline, no healthy + * dispatch is still here. + */ agentlessOccupants?: number } From 7c39f7eaa9bee4fe886ae72746c1651b9e129b4a Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 01:03:34 +0200 Subject: [PATCH 4/9] fix(orchestrator): restart the capacity backoff when a slot is released (#303) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI surfaced this rather than a reviewer. `package` failed on `keeps a durable queued issue from spawning after restart until the running slot is released`, and the honest reading is that the backoff added in this PR widened that test's failure window rather than merely coinciding with it. Before the backoff, a queued lifecycle had two independent paths to the freed slot: the local completion, which dispatches the next issue directly, and the flat 1 Hz retry. The retry was a second-resolution safety net under the first. Backing it off to 30 s removed that net, so anything slow on the completion path now has far longer to go unnoticed — and for a slot released by *another* process there is no local event at all, so the timer is the only signal. Trading a retry storm for up to 30 s of dispatch latency is not the trade this PR meant to make. The backoff exists to damp retries asking a question whose answer is not changing. A terminal lifecycle changes it, so `#saveDispatchLifecycle` now resets every pending capacity waiter to the base delay and re-arms it when a save lands in a terminal phase. `sinceAtMs` is deliberately left alone: the issue really has been waiting that long, and the escalating warning should keep saying so. The storm stays bounded, because it only ever occurred while nothing was moving. Also gives that pre-existing test an explicit 30 s timeout. Its fixed 2.2 s observation plus a 4 s wait always exceeded vitest's 5 s default; it passed only when the wait resolved early, which is not a property of the code under test. Regression test verified to fail with the reset removed: the waiter's ladder stays at `[8000, 16000]` instead of restarting at `1000`. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 74 +++++++++++++++++++++++++++++++- src/orchestrator/factory.ts | 33 +++++++++++++- 2 files changed, 105 insertions(+), 2 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index e841752e..e4572b00 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -8223,7 +8223,9 @@ describe('FactoryLoop', () => { await first.stop() await rm(root, { recursive: true, force: true }) } - }) + // The fixed 2.2 s observation plus the 4 s wait always exceeded vitest's + // 5 s default, so this only ever passed because the wait resolved early. + }, 30_000) it('enforces one global durable slot across concurrent same-host control-plane processes', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-global-capacity-')) @@ -11028,6 +11030,76 @@ describe('FactoryLoop', () => { } }, 40_000) + // #303 review follow-up: the backoff must not outlive the condition it is + // damping. It exists to stop retries asking a question whose answer is not + // changing; a terminal lifecycle changes it, and a waiter with no local + // event to ride on has only this timer. + it('restarts the capacity backoff when a slot is released (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-capacity-backoff-reset-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const mount = new FakeMountClient({ + [issuePath(311)]: issueFile(311), + [issuePath(312)]: issueFile(312), + [issuePath(313)]: issueFile(313), + '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, + }, { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 311, + url: 'https://github.com/AgentWorkforce/pear/pull/311', + headRef: input.headRef!, + }), + closePullRequest: async () => undefined, + }) + const fleet = new RemoteLifecycleFleetClient() + const waits: Array<{ issue: string; retryMs: number }> = [] + const logger = { + debug: vi.fn(), + info: vi.fn(), + error: vi.fn(), + warn: vi.fn((message: string, details?: Record) => { + if (message === '[factory] durable dispatch is queued for batch capacity; retries remain active') { + waits.push({ issue: details!.issue as string, retryMs: details!.retryMs as number }) + } + }), + } + const factory = createFactory(config({ batchSize: 1 }), { + mount, + fleet, + stateStore: state(), + triage: new StaticTriage(), + logger, + probePrResolver: async () => undefined, + }) + try { + for (const number of [311, 312, 313]) { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issueFile(number)))) + } + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-311-impl-pear', 'ar-311-review']) + + // Let the third issue climb its ladder past the base delay. + await vi.waitFor(() => expect(waits.filter((wait) => wait.issue === 'AR-313') + .some((wait) => wait.retryMs >= 4_000)).toBe(true), { timeout: 15_000 }) + + const beforeExit = waits.filter((wait) => wait.issue === 'AR-313').length + fleet.emitAgentExit('ar-311-impl-pear', 'exited') + + // AR-312 takes the freed slot; AR-313 keeps waiting, but its ladder + // starts over at the base delay rather than staying parked behind the + // one it had already climbed. + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)) + .toContain('ar-312-impl-pear'), { timeout: 15_000 }) + await vi.waitFor(() => expect( + waits.filter((wait) => wait.issue === 'AR-313').slice(beforeExit).map((wait) => wait.retryMs), + ).toContain(1_000), { timeout: 15_000 }) + expect(factory.status().counters.dispatchCapacityBackoffResets).toBeGreaterThanOrEqual(1) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 60_000) + // #303 must-not-fire control. The window between `promoteDispatchLifecycle` // and the first `recordSpawn` legitimately has zero agents. Reaping it on // sight would convert a wedge into a dispatch race. diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index ee7460a5..bd21956f 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -721,6 +721,7 @@ export class FactoryLoop implements Factory { * `status().dispatchCapacity` are derived from. */ readonly #dispatchLifecycleCapacityWaits = new Map Date: Fri, 21 Aug 2026 01:49:30 +0200 Subject: [PATCH 5/9] fix(orchestrator): wake capacity waiters on the occupancy transition, not the terminal phase (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two findings, and the first turned out to be broader than reported. 1. cubic: the backoff reset fired on every terminal save, including a `queued` row abandoned at startup because its source issue went terminal — a transition that frees no slot. Correct, and the same category of error as the defect itself: treating a non-slot event as a slot event. With batchSize 1 and a queue of waiters it re-triggers the thundering herd against the state document that this PR bounded. Gating on "terminal AND the previous phase occupied a slot" is still wrong, and my own positive test caught it: `releasing` does not occupy a slot, so a normal completion frees it one save *before* `complete`, and that gate reset nothing on a real release. A babysitter handoff frees a slot without ever going terminal at all, which the original code missed in the other direction. The event is the occupancy transition, not the phase. The reset now fires exactly when a write takes a row from occupying to not occupying, which is the only thing that can change a waiter's answer. Phase was only ever a proxy for it. 2. cubic: `countAgentlessOccupants` used a strict `>` against the reap deadline while the reaper skips only while `nowMs < dueAtMs` — so at exactly the deadline the reaper reaps and the diagnostic said the slot was fine. A diagnostic that disagrees with the mechanism it reports on is how this outage stayed invisible; now `>=`. Both regression tests verified to fail with their fix reverted, and the occupancy gate is proven in both directions: reverted to unconditional it fails the never-held-a-slot test, and the phase-based gate fails the slot-released test. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 103 +++++++++++++++++++++++++ src/orchestrator/factory.ts | 22 +++++- src/orchestrator/public-health.test.ts | 21 +++++ src/orchestrator/public-health.ts | 6 +- 4 files changed, 147 insertions(+), 5 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index e4572b00..2d3e7b52 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -11100,6 +11100,109 @@ describe('FactoryLoop', () => { } }, 60_000) + // #303 review (cubic): the reset must be gated on the transition actually + // freeing a slot. A `queued` row going terminal — a queued issue abandoned at + // startup because its source went terminal — frees nothing, and waking every + // waiter for it re-creates the retry storm on an event that cannot have + // changed any of their answers. + it('does not restart the capacity backoff when a terminal row never held a slot (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-non-slot-terminal-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const seedMount = new FakeMountClient({ + [issuePath(314)]: issueFile(314), + [issuePath(315)]: issueFile(315), + [issuePath(316)]: issueFile(316), + }) + const seed = createFactory(config({ batchSize: 1 }), { + mount: seedMount, + fleet: new RemoteLifecycleFleetClient(), + stateStore: state(), + triage: new StaticTriage(), + }) + const decisions = new Map() + for (const number of [314, 315, 316]) { + decisions.set(number, await seed.triageIssue(parseLinearIssue(issuePath(number), issueFile(number)))) + } + await seed.stop() + + const lifecycleFor = ( + number: number, + phase: DispatchLifecycle['phase'], + agents: DispatchLifecycle['agents'], + heldSinceAtMs?: number, + ): DispatchLifecycle => ({ + runId: `run-${number}`, + issue: { ...decisions.get(number)!.issue }, + decision: decisions.get(number)!, + dryRun: false, + phase, + agents, + invocationIds: [], + ...(heldSinceAtMs !== undefined ? { heldSinceAtMs } : {}), + updatedAtMs: 0, + }) + + // Seeded in this order so adoption walks holder -> waiter -> abandoned row. + const holderSpec = decisions.get(314)!.implementers[0]! + const holder = lifecycleFor(314, 'running', [{ + name: holderSpec.name, + tracked: { + spec: { ...holderSpec }, + result: { name: holderSpec.name, sessionRef: 'session-314', node: 'sf-mini', locality: 'remote' }, + }, + }], Date.now()) + const nowMs = Date.now() + await state().claimDispatchLifecycle('factory-test', issueKey(holder.issue), holder, 'dead-owner', nowMs, 1) + for (const number of [315, 316]) { + const queued = lifecycleFor(number, 'queued', []) + await state().claimDispatchLifecycle('factory-test', issueKey(queued.issue), queued, 'dead-owner', nowMs, 1) + } + + // AR-316's source went terminal while it sat queued. Its read is delayed so + // AR-315's capacity wait is already registered when AR-316 is abandoned — + // without that, the map is empty and the reset is a no-op eitherway. + let delayTerminalRead = true + class SlowTerminalIssueMount extends FakeMountClient { + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === issuePath(316) && delayTerminalRead) { + delayTerminalRead = false + await new Promise((resolve) => setTimeout(resolve, 1_800)) + } + return super.readFile(path) + } + } + const fleet = new RemoteLifecycleFleetClient() + const factory = createFactory(config({ batchSize: 1 }), { + mount: new SlowTerminalIssueMount({ + [issuePath(314)]: issueFile(314), + [issuePath(315)]: issueFile(315), + [issuePath(316)]: issueFile(316, done), + }), + fleet, + stateStore: state(), + triage: new StaticTriage(), + probePrResolver: async () => undefined, + }) + try { + await factory.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(async () => expect( + await state().getDispatchLifecycle('factory-test', issueKey(decisions.get(316)!.issue)), + ).toMatchObject({ phase: 'abandoned' }), { timeout: 15_000 }) + // AR-314 still holds the only slot, so nothing was freed. + expect(await state().getDispatchLifecycle('factory-test', issueKey(decisions.get(314)!.issue))) + .toMatchObject({ phase: 'running' }) + // Establishes that a waiter really was parked at that moment — without + // this the assertion below passes vacuously. + expect(factory.status().counters.dispatchLifecycleCapacityWaits).toBeGreaterThanOrEqual(1) + expect(factory.status().counters.dispatchCapacityBackoffResets).toBeUndefined() + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 40_000) + // #303 must-not-fire control. The window between `promoteDispatchLifecycle` // and the first `recordSpawn` legitimately has zero agents. Reaping it on // sight would convert a wedge into a dispatch race. diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index bd21956f..ea22df7b 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -56,7 +56,11 @@ 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 { dispatchHandedOffToBabysitters, dispatchPhaseOccupiesSlot } from '../state/dispatch-lifecycle-slot' +import { + dispatchHandedOffToBabysitters, + dispatchLifecycleOccupiesSlot, + dispatchPhaseOccupiesSlot, +} from '../state/dispatch-lifecycle-slot' import { containsExplicitIssueReference, containsIssueKey, factoryBranchBelongsToIssue } from '../issue-key-match' import { normalizeLogger, normalizeLogValue, setSafeErrorStack, stringifyLogValue } from '../logging' import { isInFactoryScope } from '../safety/factory-scope' @@ -5937,9 +5941,19 @@ export class FactoryLoop implements Factory { } if (isTerminalDispatchLifecycle(lifecycle)) { this.#dispatchLifecycleEpochs.delete(key) - // This row just gave up its slot, so every waiter's answer may have - // changed. Local completion dispatches the next issue directly; this - // is what covers a waiter with no local event to ride on. + } + // Wake the capacity waiters exactly when this write gives a slot back — + // occupied before, not occupied after — and never otherwise (#303 + // review, cubic). + // + // Not "when it goes terminal". `releasing` already does not occupy a + // slot, so a normal completion frees it one save *before* `complete`, + // and a babysitter handoff frees it without ever going terminal at all. + // Keying on the terminal save alone therefore both fires for rows that + // freed nothing (a `queued` row abandoned at startup) and misses the + // writes that actually freed something. The occupancy transition is the + // event; the phase is only a proxy for it. + if (previous && dispatchLifecycleOccupiesSlot(previous) && !dispatchLifecycleOccupiesSlot(lifecycle)) { this.#resetDispatchCapacityBackoff() } return true diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 082ee4ce..ecf4612e 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -99,6 +99,27 @@ describe('dispatch capacity health (#303)', () => { expect(health.dispatchCapacity?.state).toBe('stalled') }) + // The reaper skips only while `nowMs < dueAtMs`, so it reaps AT the + // deadline. A strict `>` here would make the diagnostic disagree with the + // mechanism it reports on for that instant. + it('counts a never-placed slot that reached its reap deadline exactly', () => { + const health = publicHealthFromHeartbeat( + capacity({ + occupants: [{ + issue: 'AR-303', + phase: 'dispatching', + agents: 1, + placedAgents: 0, + slotHeldForMs: 30 * 60_000, + }], + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.dispatchCapacity?.agentlessHoldTimeoutMs).toBe(30 * 60_000) + expect(health.dispatchCapacity?.agentlessOccupants).toBe(1) + }) + it('survives a corrupted occupants collection rather than dropping the block', () => { for (const occupants of ['not-an-array', null, [null], [42], [{ placedAgents: 0 }]]) { const health = publicHealthFromHeartbeat( diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index 03e6cbcb..937fad9a 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -313,9 +313,13 @@ function countAgentlessOccupants(occupants: unknown, reapMs: number): number { const slotHeldForMs = finiteNumber(occupant.slotHeldForMs) // A producer that sends neither field cannot answer the question, and // guessing "wedged" from an absence is how a false alarm gets published. + // `>=`, not `>`: the reaper skips only while `nowMs < dueAtMs`, so it + // reaps at exactly the deadline. A diagnostic that disagrees with the + // mechanism it reports on — even on one boundary instant — is the failure + // mode this whole PR exists to close (#303 review, cubic). return finiteNumber(occupant.placedAgents) === 0 && slotHeldForMs !== undefined && - slotHeldForMs > reapMs + slotHeldForMs >= reapMs }).length } From 0634d505624c3939f64a4b76ca8c3d7a4895cdf1 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 02:31:17 +0200 Subject: [PATCH 6/9] fix(orchestrator): arm the slot deadline at dispatch, and stop a released babysitter suppressing occupancy (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit's first pass that was not rate-limited found two Majors, both real. 1. The fresh dispatch path stamps `slotHeldSinceAtMs` at its first `dispatching` save, but only armed the deadline after a placement succeeded. `#fleet.spawn` deliberately carries no mutation timeout, so a first attempt that hangs in an otherwise idle process held the batch slot with no timer that could ever fire — the #303 shape reached through dispatch instead of durable recovery, and a gap in this PR's own fix. The deadline is now armed before the first await. 2. `dispatchHandedOffToBabysitters` ignored release state, so a released babysitter still satisfied the handoff and dropped its lifecycle out of `batchSize` accounting while nothing was watching the PR. Admission over-subscribes and the reaper stops bounding a row it still needs to bound. The predicate now takes agents rather than bare specs so the release stamp is available at all, reading it from the durable row or the tracked agent the way `inFlightRecordFromLifecycle` does. The ignored-release half predates this PR — both state stores had it — but the extraction in cf427eb made it structurally unfixable at the call site by passing specs only, and three new callers now depend on the predicate, so it is fixed here rather than left behind a refactor that made it worse. Adds `src/state/dispatch-lifecycle-slot.test.ts` for the extracted module, covering the handoff, the released-babysitter cases and the slot anchor's carry-forward and clearing. Both fixes verified to fail with the fix reverted. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 54 +++++++++++++ src/orchestrator/factory.ts | 10 ++- src/state/dispatch-lifecycle-slot.test.ts | 94 +++++++++++++++++++++++ src/state/dispatch-lifecycle-slot.ts | 25 ++++-- 4 files changed, 177 insertions(+), 6 deletions(-) create mode 100644 src/state/dispatch-lifecycle-slot.test.ts diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 2d3e7b52..72fcb78a 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -11203,6 +11203,60 @@ describe('FactoryLoop', () => { } }, 40_000) + // #303 review (CodeRabbit): the fresh dispatch path stamps the slot at its + // first `dispatching` save, but only armed the deadline after a placement + // succeeded. `#fleet.spawn` carries no mutation timeout by design, so a first + // attempt that hangs in an otherwise idle process held the slot with no timer + // that could ever fire — the #303 shape reached through dispatch rather than + // durable recovery. + it('reaps a first dispatch whose spawn never returns (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-fresh-dispatch-hang-')) + const state = () => new FileStateStore({ batchSize: 1, watchStatePath: join(root, 'state.json') }) + const gate = Promise.withResolvers() + class HangingSpawnFleetClient extends RemoteLifecycleFleetClient { + override async spawn(input: SpawnInput): Promise { + await gate.promise + return super.spawn(input) + } + } + const fleet = new HangingSpawnFleetClient() + const logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() } + const factory = createFactory(config({ + batchSize: 1, + dispatch: { agentHoldTimeoutMs: 4 * 60 * 60_000, agentlessHoldTimeoutMs: 1_000 }, + loop: { heartbeatPath: join(root, 'heartbeat.json'), registryPath: join(root, 'registry.json') }, + }), { + mount: new FakeMountClient({ [issuePath(317)]: issueFile(317) }), + fleet, + stateStore: state(), + triage: new StaticTriage(), + logger, + }) + try { + const decision = await factory.triageIssue(parseLinearIssue(issuePath(317), issueFile(317))) + const key = issueKey(decision.issue) + // Nothing else is in flight, so no other record's deadline can sweep + // this one in by side effect. + const dispatched = factory.dispatch(decision).catch(() => undefined) + + await vi.waitFor(async () => expect(await state().getDispatchLifecycle('factory-test', key)) + .toMatchObject({ phase: 'dispatching' }), { timeout: 6_000 }) + await vi.waitFor(async () => expect(await state().getDispatchLifecycle('factory-test', key)) + .toMatchObject({ releaseReason: 'agentless-slot-past-deadline' }), { timeout: 12_000 }) + expect(logger.warn).toHaveBeenCalledWith( + '[factory] releasing a dispatch lifecycle that never placed an agent', + expect.objectContaining({ issue: 'AR-317', holdTimeoutMs: 1_000 }), + ) + + gate.resolve() + await dispatched + } finally { + gate.resolve() + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 40_000) + // #303 must-not-fire control. The window between `promoteDispatchLifecycle` // and the first `recordSpawn` legitimately has zero agents. Reaping it on // sight would convert a wedge into a dispatch race. diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index ea22df7b..01575dbd 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -4541,6 +4541,14 @@ export class FactoryLoop implements Factory { if (!await this.#saveDispatchLifecycle(record, 'dispatching')) { throw new Error(`Dispatch lifecycle ownership lost immediately before spawning ${dispatchDecision.issue.key}`) } + // That save stamped `slotHeldSinceAtMs`: the row occupies a batch slot from + // here on. Arm its deadline before the first await rather than after a + // placement succeeds — `#fleet.spawn` deliberately carries no mutation + // timeout, so a first attempt that hangs in an otherwise idle process would + // otherwise hold the slot with no timer that can ever fire. That is the + // exact shape of #303, reached through the fresh dispatch path instead of + // the durable one (#303 review, CodeRabbit). + this.#scheduleHeldAgentDeadline(record) if (!dryRun) await this.#ensureGithubAgentQuestionWatch(record, liveIssue) const spawnedForReaperHandoff: RegistryHandoffAgent[] = [] @@ -5277,7 +5285,7 @@ export class FactoryLoop implements Factory { #recordOccupiesSlot(record: InFlightIssue): boolean { return dispatchPhaseOccupiesSlot(record.lifecyclePhase) && !dispatchHandedOffToBabysitters( record.decision.implementers, - [...record.agents.values()].map((tracked) => tracked.spec), + [...record.agents.values()].map((tracked) => ({ releasedAtMs: tracked.releasedAtMs, spec: tracked.spec })), ) } diff --git a/src/state/dispatch-lifecycle-slot.test.ts b/src/state/dispatch-lifecycle-slot.test.ts new file mode 100644 index 00000000..b46a75c4 --- /dev/null +++ b/src/state/dispatch-lifecycle-slot.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from 'vitest' + +import { dispatchLifecycleOccupiesSlot, stampDispatchLifecycleSlot } from './dispatch-lifecycle-slot' +import type { DispatchLifecycle, DispatchLifecycleAgent } from '../ports/state' + +const issue = { uuid: 'uuid-1', key: 'AR-1', path: '/linear/issues/AR-1__uuid-1.json' } + +const implementer = { + name: 'ar-1-impl-pear', + role: 'implementer' as const, + capability: 'spawn:codex' as const, + repo: 'AgentWorkforce/pear', + task: 'implement', +} + +const babysitter = (releasedAtMs?: number): DispatchLifecycleAgent => ({ + name: 'ar-1-babysit', + ...(releasedAtMs !== undefined ? { releasedAtMs } : {}), + tracked: { + spec: { + name: 'ar-1-babysit', + role: 'babysitter' as const, + capability: 'spawn:codex' as const, + repo: 'AgentWorkforce/pear', + task: 'babysit', + ownedPullRequest: { repo: 'AgentWorkforce/pear', number: 7, path: '/github/pr/7' }, + }, + result: { name: 'ar-1-babysit' }, + }, +}) + +const lifecycle = (overrides: Partial = {}): DispatchLifecycle => ({ + runId: 'run-1', + issue, + decision: { + issue, + implementers: [implementer], + reviewer: { name: 'ar-1-review', role: 'reviewer', capability: 'spawn:codex', repo: 'AgentWorkforce/pear', task: 'review' }, + routes: [], + confidence: 'high', + rationale: 'test', + } as unknown as DispatchLifecycle['decision'], + dryRun: false, + phase: 'running', + agents: [], + invocationIds: [], + updatedAtMs: 0, + ...overrides, +}) + +describe('dispatch batch-slot accounting (#303)', () => { + it('stops counting a lifecycle whose PR is babysat', () => { + expect(dispatchLifecycleOccupiesSlot(lifecycle({ agents: [babysitter()] }))).toBe(false) + }) + + // A released babysitter is not babysitting. Letting one satisfy the handoff + // drops the lifecycle out of `batchSize` accounting while nothing is watching + // its PR — admission over-subscribes and the reaper stops bounding a row it + // still needs to bound. + it('keeps counting a lifecycle whose only babysitter has been released', () => { + expect(dispatchLifecycleOccupiesSlot(lifecycle({ agents: [babysitter(1_000)] }))).toBe(true) + }) + + it('reads the release stamp from the tracked agent too', () => { + const released = babysitter() + released.tracked = { ...released.tracked, releasedAtMs: 1_000 } + + expect(dispatchLifecycleOccupiesSlot(lifecycle({ agents: [released] }))).toBe(true) + }) + + it('resumes the handoff once a live babysitter replaces the released one', () => { + const replacement = babysitter() + replacement.name = 'ar-1-babysit-2' + + expect(dispatchLifecycleOccupiesSlot(lifecycle({ agents: [babysitter(1_000), replacement] }))).toBe(false) + }) + + it('clears the slot anchor when a row stops occupying a slot', () => { + const held = lifecycle({ slotHeldSinceAtMs: 500 }) + stampDispatchLifecycleSlot(held, held, 900) + expect(held.slotHeldSinceAtMs).toBe(500) + + const releasing = lifecycle({ phase: 'releasing', slotHeldSinceAtMs: 500 }) + stampDispatchLifecycleSlot(releasing, releasing, 900) + expect(releasing.slotHeldSinceAtMs).toBeUndefined() + }) + + it('carries the anchor forward from the stored row rather than restamping', () => { + const next = lifecycle() + stampDispatchLifecycleSlot(next, lifecycle({ slotHeldSinceAtMs: 500 }), 9_000) + + expect(next.slotHeldSinceAtMs).toBe(500) + }) +}) diff --git a/src/state/dispatch-lifecycle-slot.ts b/src/state/dispatch-lifecycle-slot.ts index e83520dc..64fbfdca 100644 --- a/src/state/dispatch-lifecycle-slot.ts +++ b/src/state/dispatch-lifecycle-slot.ts @@ -29,13 +29,22 @@ export const dispatchPhaseOccupiesSlot = (phase: DispatchLifecyclePhase | undefi */ export const dispatchHandedOffToBabysitters = ( implementers: ReadonlyArray<{ repo: string }>, - agentSpecs: ReadonlyArray<{ role?: string; ownedPullRequest?: { repo: string } }>, + agents: ReadonlyArray<{ + releasedAtMs?: number + spec: { role?: string; ownedPullRequest?: { repo: string } } + }>, ): boolean => { const implementerRepos = [...new Set(implementers.map((spec) => spec.repo))] if (implementerRepos.length === 0) return false - const babysitterRepos = agentSpecs - .filter((spec) => spec.role === 'babysitter') - .map((spec) => spec.ownedPullRequest?.repo) + const babysitterRepos = agents + // A released babysitter is not babysitting anything. Letting one satisfy + // the handoff would drop the lifecycle out of `batchSize` accounting while + // nothing is actually watching its PR, so admission would over-subscribe + // and the reaper would stop bounding a row it still needs to bound (#303 + // review, CodeRabbit). Release state is the reason this takes agents + // rather than bare specs. + .filter((agent) => agent.releasedAtMs === undefined && agent.spec.role === 'babysitter') + .map((agent) => agent.spec.ownedPullRequest?.repo) .filter((repo): repo is string => Boolean(repo)) return implementerRepos.every((repo) => babysitterRepos.some((ownedRepo) => githubRepositoriesMatch(repo, ownedRepo))) @@ -44,7 +53,13 @@ export const dispatchHandedOffToBabysitters = ( export const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => dispatchHandedOffToBabysitters( lifecycle.decision.implementers, - lifecycle.agents.map((agent) => agent.tracked.spec), + lifecycle.agents.map((agent) => ({ + // The durable row carries the stamp in either place depending on which + // writer last touched it; `inFlightRecordFromLifecycle` reads it the + // same way. + releasedAtMs: agent.releasedAtMs ?? agent.tracked.releasedAtMs, + spec: agent.tracked.spec, + })), ) export const dispatchLifecycleOccupiesSlot = (lifecycle: DispatchLifecycle): boolean => From 640298e277d58220612deaf2d99bc7a3ced06851 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 02:50:33 +0200 Subject: [PATCH 7/9] fix(orchestrator): release a placement that lands after its dispatch was reaped (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Arming the never-placed deadline before the first spawn await (0634d50) made a new interleaving reachable: the reaper can now fence, release and terminalize a lifecycle while `#fleet.spawn` is still in flight. When that spawn finally returned, `#spawnAgent` carried on — stamping `heldSinceAtMs`, recording the placement, and saving — onto a record the reaper had already finished with. The durable save failed on the dropped epoch, so the terminal row was never resurrected, but the worker itself was live on the fleet with nothing left to release it. That trades a wedged batch slot for a leaked agent, which is not a trade worth making. `#spawnAgent` now revalidates ownership after the spawn await — pending abandon reason, dropped epoch, or a terminal durable row — and hands a late placement straight to teardown instead of recording it. Deliberately not routed through `#abandonStuckDispatch`: that record is already terminal and its batch entry is gone, so the only thing still needing attention is the worker. Extends the hung-spawn regression test to release the gate afterwards and assert the late placement is released, that no spawn result is persisted onto the terminal row, and that `heldSinceAtMs` stays unset. Verified to fail with the guard removed. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 17 +++++++++ src/orchestrator/factory.ts | 61 ++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 72fcb78a..a08bf6a6 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -11248,8 +11248,25 @@ describe('FactoryLoop', () => { expect.objectContaining({ issue: 'AR-317', holdTimeoutMs: 1_000 }), ) + // The spawn now returns, long after the reaper released the lifecycle. + // That placement belongs to nothing, so it must be torn down rather than + // recorded onto a record the reaper has finished with (#303 review, + // cubic) — otherwise arming the deadline earlier trades a wedged slot + // for a leaked worker. gate.resolve() await dispatched + await vi.waitFor(() => expect(fleet.releases).toContainEqual({ + name: 'ar-317-impl-pear', + reason: 'dispatch-released-before-placement', + }), { timeout: 8_000 }) + expect(factory.status().counters.lateSpawnPlacementsReleased).toBeGreaterThanOrEqual(1) + // The terminal row is not resurrected by the late placement. It still + // carries the planned spec `recordPlanned` wrote before the spawn — that + // is a record of intent — but no spawn result was ever persisted onto it. + const terminal = await state().getDispatchLifecycle('factory-test', key) + expect(terminal).toMatchObject({ releaseReason: 'agentless-slot-past-deadline' }) + expect(terminal?.agents.map((agent) => agent.tracked.result)).toEqual([undefined]) + expect(terminal?.heldSinceAtMs).toBeUndefined() } finally { gate.resolve() await factory.stop() diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 01575dbd..b63560b4 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -8427,6 +8427,17 @@ export class FactoryLoop implements Factory { : 'agent_spawn_failed' as const, }) } + // The never-placed deadline can fire while this spawn is in flight — that + // is the whole point of arming it before the first await, and it makes a + // late `spawn` result newly reachable (#303 review, cubic). By now the + // reaper may have fenced, released and terminalized the lifecycle, so this + // placement belongs to nothing: recording it would attach a live worker to + // a record the reaper has finished with, and nothing downstream would ever + // release it. Hand it straight to teardown instead. + if (!await this.#dispatchLifecycleStillOwned(record)) { + await this.#releaseOrphanedLatePlacement(record, spec, result) + throw new Error(`Dispatch lifecycle for ${record.issue.key} was released while ${spec.name} was still spawning`) + } record.heldSinceAtMs ??= this.#clock.now() batch.recordSpawn(record, spec, invocationId, result) if (!await this.#saveDispatchLifecycle(record, 'dispatching')) { @@ -8438,6 +8449,56 @@ export class FactoryLoop implements Factory { return { name: result.name } } + /** + * Is this process still the owner of a lifecycle that is not already done? + * + * Cheap local checks first — a pending abandon reason, or a dropped epoch, + * both of which the reaper sets before anything durable is re-read — then the + * durable row, which is authoritative when another owner terminalized it. + */ + async #dispatchLifecycleStillOwned(record: InFlightIssue): Promise { + const key = issueKey(record.issue) + if (this.#abandonedDispatchReasons.has(key)) return false + if (!this.#usesDurableDispatchLifecycle()) return true + if (!this.#dispatchLifecycleEpochs.has(key)) return false + const lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + return Boolean(lifecycle) && !isTerminalDispatchLifecycle(lifecycle!) + } + + /** + * Tear down a placement that landed after its lifecycle was already released. + * + * Deliberately not routed through `#abandonStuckDispatch`: that record is + * terminal and its batch entry is gone, so there is nothing left to abandon. + * The only thing that still exists is a live worker on the fleet. + */ + async #releaseOrphanedLatePlacement( + record: InFlightIssue, + spec: AgentSpec, + result: SpawnResult, + ): Promise { + const name = result.name ?? spec.name + this.#increment('lateSpawnPlacementsReleased') + this.#logger.warn?.('[factory] releasing an agent that finished spawning after its dispatch was released', { + issue: record.issue.key, + agent: name, + role: spec.role, + }) + this.#fleet.markAgentTerminal?.(name, 'dispatch-released-before-placement') + try { + await this.#fleet.release(name, 'dispatch-released-before-placement') + } catch (error) { + // The reaper handoff owns anything this could not clean up; failing here + // would only replace a released worker with an unreleased one. + this.#increment('lateSpawnPlacementReleaseFailures') + this.#logger.warn?.('[factory] failed to release a late placement; leaving it to the orphan reaper', { + issue: record.issue.key, + agent: name, + error: describeError(error).errorMessage, + }) + } + } + async #handleAgentExit(name: string, reason?: string): Promise { if (this.#stopping) { return From 0f8fda38f1a91ac93c612d2711a7d0755df1ae76 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 03:04:33 +0200 Subject: [PATCH 8/9] fix(orchestrator): classify a late-placement release instead of counting it as an unexplained fault (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The late-placement guard in 640298e threw a plain `Error`, so `isClassifiedPerItemDispatchFailure` returned false and every occurrence incremented `unclassifiedFailuresSinceDispatch`. Five in a row with no successful dispatch between them and the whole readiness pass aborts (#292's fuse, `UNCLASSIFIED_DISPATCH_FAILURE_LIMIT`). That is not a remote possibility. The error fires exactly when the never-placed deadline terminalizes a lifecycle whose spawn is still in flight — a slow-spawn condition, which is the precise condition the deadline exists for. A degraded fleet produces the race repeatedly, so the fix for a wedged batch slot would have become an aborting sweep: `readinessReconcile.lastError` red and dispatch stopped, which from outside is the outage it was meant to end. Two individually-correct changes meeting badly — the same seam that produced #303. Adds a narrow, named `LatePlacementReleasedError` carrying the issue key and agent name, classifies it, and gives it its own run-report reason (`dispatch released while its agent was still spawning`) rather than falling through to `dispatch failed (...)`. The predicate is not widened or loosened: the fuse still catches a pass-wide fault wearing per-item clothes. Visibility is unchanged — `lateSpawnPlacementsReleased` and `lateSpawnPlacementReleaseFailures` already carry the condition without polluting `counters.errors`. `mayHaveSpawnedBeforeFailing` deliberately still returns true for it. `#reapDispatchFailureHandoffsNow` is documented and written to be idempotent, and this path never reaches `batch.recordSpawn`, so there is no handoff for the late agent to reap; the existing regression test asserts exactly one release, which would catch a duplicate. Test pair, both verified: - MUST-FIRE: seven consecutive late-placement releases with no successful dispatch between them return a report instead of aborting. Reverted, it fails with `Aborting readiness pass after 5 unclassified dispatch failures without a successful dispatch: Dispatch lifecycle for AR-81 was released while ar-81-impl-pear was still spawning`. - MUST-NOT-FIRE: five genuinely unclassified failures still abort. It is a control, so it passes before and after; widening the predicate to `error instanceof Error` makes it fail with `promise resolved ... instead of rejecting`, which is what proves it still guards the fuse. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 39 ++++++++++++++++++++++++++++++-- src/orchestrator/factory.ts | 34 +++++++++++++++++++++++++++- 2 files changed, 70 insertions(+), 3 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index a08bf6a6..fda31729 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -28,7 +28,7 @@ import { type TriageEngine, type WorkflowRunnerInput, } from '../index' -import { changeEventPath } from './factory' +import { LatePlacementReleasedError, changeEventPath } from './factory' import type { AgentWorktree, AgentWorktreeCleanupInspection, AgentWorktreeManager, AgentWorktreeRepository, ChangeEvent, EventPage, GithubConnectionRead, GithubConnectionWrite, GithubIssueStatus, GithubPublishPullRequestInput, GithubWriteback, LinearWriteback, PreviewReference, PreviewStartInput, ProviderSyncStatus, RosterEntry, SlackWriteback, SpawnInput, SpawnResult } from '../ports' import { FakeFleetClient, FakeMountClient, withDeadline } from '../testing' import type { CloseProbePrInput, GithubMergeGatePort, GithubMergeGateVerdict, GithubMergeInput, LinearIssue, VerificationGate, VerificationGateInput, VerificationVerdict } from '../index' @@ -5449,9 +5449,44 @@ describe('FactoryLoop', () => { expect(fleet.spawns).toEqual([]) }) + // #303 review (factory-lead). The never-placed deadline releases a dispatch + // whose spawn is still in flight, and it does so precisely under slow + // spawns — the condition it exists for — so a degraded fleet produces the + // race repeatedly, not once. Left unclassified, five in a row would trip + // the #292 fuse and abort the whole readiness pass: a bounded slot traded + // for a stopped sweep, which from outside is the same outage again. + it('does not abort the pass when late-placement releases repeat (#303)', async () => { + const numbers = [81, 82, 83, 84, 85, 86, 87] + const paths = numbers.map((number) => githubIssuePath('AgentWorkforce', 'pear', number)) + const mount = new FakeMountClient(Object.fromEntries( + paths.map((path, index) => [path, githubIssueFile(numbers[index]!, { labels: ['factory', 'pear'] })]), + )) + const fleet = new LocalLifecycleFleetClient() + const factory = createFactory(config({ issueSource: 'github', batchSize: 5 }), { + mount, + fleet, + triage: new FailingTriage(() => new LatePlacementReleasedError('AR-81', 'ar-81-impl-pear')), + githubWriteback: new RecordingGithubWriteback(), + }) + + // Seven consecutive, well past UNCLASSIFIED_DISPATCH_FAILURE_LIMIT, with + // no successful dispatch in between to reset the counter. + const report = await factory.runOnce() + + expect(report.skipped).toHaveLength(numbers.length) + expect(report.skipped[0]).toMatchObject({ + reason: 'dispatch released while its agent was still spawning', + }) + expect(factory.status().counters.dispatchItemsSkippedUndispatchable).toBe(numbers.length) + // Stays out of `counters.errors` and out of the unexplained-fault bucket. + expect(factory.status().counters.dispatchItemFailuresSkipped).toBeUndefined() + 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. + // than a green report full of skips. This is the #292 fuse, and the test + // above must not have disabled it (#303 review, factory-lead). 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( diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index b63560b4..bd88ce31 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -8436,7 +8436,7 @@ export class FactoryLoop implements Factory { // release it. Hand it straight to teardown instead. if (!await this.#dispatchLifecycleStillOwned(record)) { await this.#releaseOrphanedLatePlacement(record, spec, result) - throw new Error(`Dispatch lifecycle for ${record.issue.key} was released while ${spec.name} was still spawning`) + throw new LatePlacementReleasedError(record.issue.key, result.name ?? spec.name) } record.heldSinceAtMs ??= this.#clock.now() batch.recordSpawn(record, spec, invocationId, result) @@ -20154,6 +20154,31 @@ const triageEscalationReason = (decision: TriageDecision): string | undefined => * turns it into a distinct exit code — can recognize it by type rather than by * matching on `error.name`, and so tests can construct a genuine instance. */ +/** + * A placement that finished spawning after its dispatch had been released. + * + * The never-placed deadline (#303) can terminalize a lifecycle while + * `#fleet.spawn` is still in flight; the worker is released and the dispatch + * unwinds. That is a known, named, self-healing race — the issue returns to the + * queue and is re-dispatched — so it must be classified rather than counted as + * an unexplained fault. It fires precisely under slow spawns, which is the + * condition the deadline exists for, so a degraded fleet produces it + * repeatedly; left unclassified, five in a row would trip + * `UNCLASSIFIED_DISPATCH_FAILURE_LIMIT` and abort the whole readiness pass, + * turning a bounded slot into a stopped sweep (#303 review, factory-lead). + */ +export class LatePlacementReleasedError extends Error { + readonly issueKey: string + readonly agentName: string + + constructor(issueKey: string, agentName: string) { + super(`Dispatch lifecycle for ${issueKey} was released while ${agentName} was still spawning`) + this.name = 'LatePlacementReleasedError' + this.issueKey = issueKey + this.agentName = agentName + } +} + export class LiveDispatchStateChangedError extends Error { readonly issueKey: string @@ -20205,6 +20230,12 @@ const UNCLASSIFIED_DISPATCH_FAILURE_LIMIT = 5 const isClassifiedPerItemDispatchFailure = (error: unknown): boolean => error instanceof LiveDispatchStateChangedError || error instanceof DispatchLifecycleClaimRefusedError || + // #303: the never-placed deadline released this dispatch while its spawn was + // still in flight. Named, expected and self-healing — the issue goes back to + // the queue — and it recurs under exactly the slow-spawn conditions the + // deadline exists for, so leaving it unclassified would let a degraded fleet + // trip the pass-abort fuse. Its own counters keep it visible. + error instanceof LatePlacementReleasedError || // Relayfile shedding one operation is a state of the dependency, not an // unexplained fault, and it has its own fuse — see #297 and // DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT. @@ -20240,6 +20271,7 @@ const perItemDispatchSkipReason = (error: unknown): string => { const overload = relayfileOverload(error) if (overload) return `relayfile overloaded (${relayfileOverloadReasonLabel(overload.reason)})` if (error instanceof LiveDispatchStateChangedError) return 'live state changed during dispatch' + if (error instanceof LatePlacementReleasedError) return 'dispatch released while its agent was still spawning' if (error instanceof DispatchLifecycleClaimRefusedError) { return error.refusal === 'terminal' ? 'dispatch lifecycle already terminal' From 24b19d41b6e0c0d12d8d4cb0d21ee3d7292d7ea7 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 03:18:56 +0200 Subject: [PATCH 9/9] fix(orchestrator): treat ownership as the lease this process holds (#303 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two cubic findings on the late-placement guard. 1. `#dispatchLifecycleStillOwned` accepted any nonterminal row with a cached epoch, so a lifecycle another owner had reclaimed still read as ours. The placement was then recorded, `saveDispatchLifecycle` refused it on the epoch, and `#spawnAgent` threw the generic `Dispatch lifecycle ownership lost after spawning ...` — a plain Error that leaks the worker and feeds the unclassified-failure fuse. The same two defects the previous two commits fixed, reached through takeover instead of the deadline. The check now mirrors exactly what `saveDispatchLifecycle` will accept — owner, epoch and an unexpired lease — so a placement is recorded only when the write that follows can actually land, and anything else goes to orphan cleanup as a classified `LatePlacementReleasedError`. 2. `LatePlacementReleasedError` had been inserted between `LiveDispatchStateChangedError`'s JSDoc and its declaration, so the doc described the wrong export. Moved below it. Regression test drives a real takeover: a spawn blocked on a gate, another owner claiming the row on a future clock once this process's lease has lapsed, then the spawn returning. Verified to fail with the lease comparison removed — `expected [Function] to throw error matching /was released while .* was still spawning/ but got 'Dispatch lifecycle ownership lost after spawning ar-318-impl-pear'`. Co-Authored-By: Claude Opus 5 Session-Id: 773e58dc-110c-4d0e-97d5-a36a658636db --- src/orchestrator/factory.test.ts | 61 ++++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 37 ++++++++++++------- 2 files changed, 86 insertions(+), 12 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index fda31729..55ef3aaa 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -11309,6 +11309,67 @@ describe('FactoryLoop', () => { } }, 40_000) + // #303 review (cubic): a live durable row is not the same as *our* row. + // Another owner can reclaim an expired lease and leave it nonterminal, and a + // cached epoch would still claim ownership — so the placement gets recorded, + // the save then fails on the epoch, and the worker leaks out through the + // generic ownership-lost path instead of orphan cleanup. + it('releases a late placement when another owner reclaimed the lifecycle (#303)', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-late-takeover-')) + const watchStatePath = join(root, 'state.json') + const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) + const gate = Promise.withResolvers() + class HangingSpawnFleetClient extends RemoteLifecycleFleetClient { + override async spawn(input: SpawnInput): Promise { + await gate.promise + return super.spawn(input) + } + } + const fleet = new HangingSpawnFleetClient() + const factory = createFactory(config({ + batchSize: 1, + // Both deadlines far away: the takeover is the only thing that can end + // this dispatch, so nothing else can explain the release. + dispatch: { agentHoldTimeoutMs: 4 * 60 * 60_000, agentlessHoldTimeoutMs: 60 * 60_000 }, + loop: { heartbeatPath: join(root, 'heartbeat.json'), registryPath: join(root, 'registry.json') }, + }), { + mount: new FakeMountClient({ [issuePath(318)]: issueFile(318) }), + fleet, + stateStore: state(), + triage: new StaticTriage(), + }) + try { + const decision = await factory.triageIssue(parseLinearIssue(issuePath(318), issueFile(318))) + const key = issueKey(decision.issue) + const dispatched = factory.dispatch(decision) + + const claimed = await vi.waitFor(async () => { + const lifecycle = await state().getDispatchLifecycle('factory-test', key) + expect(lifecycle).toMatchObject({ phase: 'dispatching' }) + return lifecycle! + }, { timeout: 6_000 }) + + // Another owner reclaims the row once this process's lease has lapsed. + // Claiming on a future clock is what lets it past the live-lease guard. + const takeover = await state().claimDispatchLifecycle( + 'factory-test', key, claimed, 'other-owner', Date.now() + 10 * 60_000, 10 * 60_000, + ) + expect(takeover.acquired).toBe(true) + + gate.resolve() + await expect(dispatched).rejects.toThrow(/was released while .* was still spawning/) + await vi.waitFor(() => expect(fleet.releases).toContainEqual({ + name: 'ar-318-impl-pear', + reason: 'dispatch-released-before-placement', + }), { timeout: 8_000 }) + expect(factory.status().counters.lateSpawnPlacementsReleased).toBeGreaterThanOrEqual(1) + } finally { + gate.resolve() + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 40_000) + // #303 must-not-fire control. The window between `promoteDispatchLifecycle` // and the first `recordSpawn` legitimately has zero agents. Reaping it on // sight would convert a wedge into a dispatch race. diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index bd88ce31..837b16c0 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -8460,9 +8460,22 @@ export class FactoryLoop implements Factory { const key = issueKey(record.issue) if (this.#abandonedDispatchReasons.has(key)) return false if (!this.#usesDurableDispatchLifecycle()) return true - if (!this.#dispatchLifecycleEpochs.has(key)) return false + const epoch = this.#dispatchLifecycleEpochs.get(key) + if (epoch === undefined) return false const lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, key) - return Boolean(lifecycle) && !isTerminalDispatchLifecycle(lifecycle!) + if (!lifecycle || isTerminalDispatchLifecycle(lifecycle)) return false + // A live row is not the same as *our* row. Another owner can reclaim an + // expired lease and leave it nonterminal, and the cached epoch here would + // still say we hold it (#303 review, cubic). This mirrors exactly what + // `saveDispatchLifecycle` will accept — owner, epoch and an unexpired + // lease — so a placement is recorded only when the write that follows can + // actually land. Otherwise the save fails and the worker leaks through the + // generic ownership-lost path instead of orphan cleanup. + const lease = lifecycle.lease + return lease !== undefined && + lease.owner === this.#dispatchLifecycleOwner && + lease.epoch === epoch && + lease.leaseUntilMs > this.#clock.now() } /** @@ -20154,6 +20167,16 @@ const triageEscalationReason = (decision: TriageDecision): string | undefined => * turns it into a distinct exit code — can recognize it by type rather than by * matching on `error.name`, and so tests can construct a genuine instance. */ +export class LiveDispatchStateChangedError extends Error { + readonly issueKey: string + + constructor(issueKey: string) { + super(`Live state changed before writeback for ${issueKey}`) + this.name = 'LiveDispatchStateChangedError' + this.issueKey = issueKey + } +} + /** * A placement that finished spawning after its dispatch had been released. * @@ -20179,16 +20202,6 @@ export class LatePlacementReleasedError extends Error { } } -export class LiveDispatchStateChangedError extends Error { - readonly issueKey: string - - constructor(issueKey: string) { - super(`Live state changed before writeback for ${issueKey}`) - this.name = 'LiveDispatchStateChangedError' - this.issueKey = issueKey - } -} - /** Whether a thrown value is a {@link LiveDispatchStateChangedError}. */ export function isLiveDispatchStateChangedError(error: unknown): error is LiveDispatchStateChangedError { return error instanceof LiveDispatchStateChangedError