diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index e7e9770..ac3f017 100644 --- a/src/cli/diagnose.test.ts +++ b/src/cli/diagnose.test.ts @@ -174,6 +174,64 @@ 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, + agentlessHoldTimeoutMs: 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('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. 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 c7c619f..a9e9d4d 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -263,6 +263,29 @@ function verdictFor(diagnosis: Omit 0 + ? `. ${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.', + } + } if (health.status === 'unknown') { return { dispatching: false, @@ -444,6 +467,22 @@ 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 queue wait : ${formatDuration(capacity.longestWaitMs)} (warn past ${formatDuration(capacity.waitWarnMs)})`, + ) + if (capacity.agentlessOccupants !== undefined) { + 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'}`) } 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 2bd4c5e..3d47668 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -1293,6 +1293,14 @@ 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. 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 const eventListener = liveness.ok ? heartbeat?.eventListener ?? { state: 'unknown' as const, @@ -1311,6 +1319,7 @@ async function factoryStatusWithMountHealth( eventListener, readinessReconcile, fleetControlPlane, + ...(dispatchCapacity ? { dispatchCapacity } : {}), } return { ...observableStatus, @@ -1320,6 +1329,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 d0e7be6..96c204d 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 c127606..db61fc1 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 cdf3721..d4b2555 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 da040dd..55ef3aa 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( @@ -8223,7 +8258,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-')) @@ -10852,6 +10889,606 @@ 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 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 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 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 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 }), + ) + + // 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() + await rm(root, { recursive: true, force: true }) + } + }, 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. + 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, + agentlessHoldTimeoutMs: 30 * 60_000, + }) + 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([ @@ -21515,6 +22152,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 7bad0e0..837b16c 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -56,6 +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, + 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' @@ -86,6 +91,8 @@ import type { FactoryLoopRunOptions, FactoryLoopHeartbeat, FactoryLoopLiveness, + FactoryDispatchCapacityStatus, + FactoryDispatchSlotOccupant, FactoryReadinessReconcileStatus, FactoryDispatchClaimStatus, FactoryInFlightDispatchStatus, @@ -408,9 +415,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 +714,23 @@ 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 +1316,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 @@ -4498,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[] = [] @@ -4700,6 +4751,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 +5197,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 +5256,74 @@ 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. + */ + /** + * 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) => ({ releasedAtMs: tracked.releasedAtMs, spec: tracked.spec })), + ) + } + + #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 || !this.#recordOccupiesSlot(record)) { + 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,16 +5362,16 @@ 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 + + // 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()) { @@ -5266,26 +5383,47 @@ 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 + effective = durable if (!await this.#assertDispatchLifecycleOwner(record)) continue } - const heldForMs = Math.max(0, this.#clock.now() - heldSinceAtMs) + const agentless = effective.kind === 'agentless' + const heldForMs = Math.max(0, this.#clock.now() - effective.sinceAtMs) const details = { issue: record.issue.key, heldForMs, - holdTimeoutMs: timeoutMs, + holdTimeoutMs: effective.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 +5862,12 @@ 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 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() const key = issueKey(record.issue) @@ -5806,6 +5950,20 @@ export class FactoryLoop implements Factory { if (isTerminalDispatchLifecycle(lifecycle)) { this.#dispatchLifecycleEpochs.delete(key) } + // 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 }) } @@ -5820,29 +5978,152 @@ 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 && 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) } + : {}), + ...(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, + agentlessHoldTimeoutMs: this.#config.dispatch.agentlessHoldTimeoutMs, + ...(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 = { record, sinceAtMs: nowMs, attempts: 0 } + this.#dispatchLifecycleCapacityWaits.set(key, wait) + this.#increment('dispatchLifecycleCapacityWaits') + } + wait.record = record + 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 + } + + /** + * Put every capacity waiter back on the fast path, because a slot just freed. + * + * The backoff exists to stop a storm of retries asking a question whose + * answer is not changing. When a lifecycle reaches a terminal phase the + * answer *has* changed, so parking a waiter behind a 30 s timer would trade + * the storm for latency — and for a slot released by another process, that + * timer is the only signal this one gets (#303 review follow-up). + * + * The wait's `sinceAtMs` is deliberately untouched: the issue really has + * been waiting that long, and the escalating warning should keep saying so. + */ + #resetDispatchCapacityBackoff(): void { + if (this.#stopping || this.#dispatchLifecycleCapacityWaits.size === 0) return + for (const [key, wait] of this.#dispatchLifecycleCapacityWaits) { + wait.attempts = 0 + const timer = this.#dispatchLifecycleRetryTimers.get(key) + if (!timer) continue + clearTimeout(timer) + this.#dispatchLifecycleRetryTimers.delete(key) + this.#scheduleDispatchLifecycleRetry(wait.record) + } + this.#increment('dispatchCapacityBackoffResets') + } + + #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 +6136,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 +7775,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 +8225,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', ) @@ -8142,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 LatePlacementReleasedError(record.issue.key, result.name ?? spec.name) + } record.heldSinceAtMs ??= this.#clock.now() batch.recordSpawn(record, spec, invocationId, result) if (!await this.#saveDispatchLifecycle(record, 'dispatching')) { @@ -8153,6 +8449,69 @@ 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 + const epoch = this.#dispatchLifecycleEpochs.get(key) + if (epoch === undefined) return false + const lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + 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() + } + + /** + * 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 @@ -9244,7 +9603,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 +9632,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 +20109,7 @@ const lifecycleFromInFlightRecord = ( ...(releaseReason ? { releaseReason } : {}), ...(cost ? { cost: structuredClone(cost) } : {}), ...(record.heldSinceAtMs !== undefined ? { heldSinceAtMs: record.heldSinceAtMs } : {}), + ...(record.slotHeldSinceAtMs !== undefined ? { slotHeldSinceAtMs: record.slotHeldSinceAtMs } : {}), updatedAtMs, }) @@ -19757,10 +20128,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, }) @@ -19802,6 +20177,31 @@ export class LiveDispatchStateChangedError extends Error { } } +/** + * 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 + } +} + /** Whether a thrown value is a {@link LiveDispatchStateChangedError}. */ export function isLiveDispatchStateChangedError(error: unknown): error is LiveDispatchStateChangedError { return error instanceof LiveDispatchStateChangedError @@ -19843,6 +20243,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. @@ -19878,6 +20284,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' diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 36645b2..ecf4612 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -32,6 +32,148 @@ 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, + 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 + // 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, + }, + }) + + 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, + agentlessHoldTimeoutMs: 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) + }) + + // 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-307', + phase: 'dispatching', + agents: 1, + placedAgents: 0, + slotHeldForMs: 90_000, + }], + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + 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') + }) + + // 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( + 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', () => { + 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, + agentlessHoldTimeoutMs: 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 69bd756..937fad9 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_AGENTLESS_HOLD_TIMEOUT_MS, 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,67 @@ 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. + // `>=`, 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 + }).length +} + +/** + * 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 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 } : {}), + } +} + function fleetControlPlaneHealth( status: FleetControlPlaneStatus, ): FactoryPublicFleetControlPlaneHealth { @@ -292,6 +401,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 +425,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 +478,7 @@ export function publicHealthFromHeartbeat( ...(readinessReconcile ? { readinessReconcile } : {}), ...(eventListener ? { eventListener } : {}), ...(fleetControlPlane ? { fleetControlPlane } : {}), + ...(dispatchCapacity ? { dispatchCapacity } : {}), } } @@ -383,6 +503,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 +551,25 @@ 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, + agentlessHoldTimeoutMs: positiveNumber(capacity.agentlessHoldTimeoutMs) + ?? DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS, + ...optionalDuration('longestWaitMs', capacity.longestWaitMs), + ...optionalCount('agentlessOccupants', capacity.agentlessOccupants), + }, + } + : {}), } } diff --git a/src/ports/state.ts b/src/ports/state.ts index 80c2b13..2d767cf 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.test.ts b/src/state/dispatch-lifecycle-slot.test.ts new file mode 100644 index 0000000..b46a75c --- /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 new file mode 100644 index 0000000..64fbfdc --- /dev/null +++ b/src/state/dispatch-lifecycle-slot.ts @@ -0,0 +1,89 @@ +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' + +/** + * 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 }>, + 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 = 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))) +} + +export const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): boolean => + dispatchHandedOffToBabysitters( + lifecycle.decision.implementers, + 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 => + 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 8e99edd..ae5d6a7 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 542ee35..afec7eb 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 79c5308..5dfb3f0 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 14dc0b5..3b93ee9 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,78 @@ 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 + /** 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. */ + 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 + /** 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. */ + 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 + agentlessHoldTimeoutMs: number + longestWaitMs?: number + /** + * 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 +} + export interface FactoryPublicEventListenerHealth { state: FactoryEventListenerStatus['state'] } @@ -269,6 +343,7 @@ export interface FactoryPublicHealth { readinessReconcile?: FactoryPublicReadinessReconcileHealth eventListener?: FactoryPublicEventListenerHealth fleetControlPlane?: FactoryPublicFleetControlPlaneHealth + dispatchCapacity?: FactoryPublicDispatchCapacityHealth } export interface FactoryInFlightRegistryAgent { @@ -416,6 +491,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[] }