diff --git a/docs/deployed-diagnostics.md b/docs/deployed-diagnostics.md index d0566a64..6d491a32 100644 --- a/docs/deployed-diagnostics.md +++ b/docs/deployed-diagnostics.md @@ -65,7 +65,8 @@ logic of its own by design: the boundary lives in one place, in this repo, with "intervalMs": 60000, "lastStartedAtMs": 1787224595805, // 11:16:35.805Z "lastCompletedAtMs": 1787224535802, // 11:15:35.802Z — 60s EARLIER - "inFlightMs": 4560000, // now − lastStarted: this pass has run 76 minutes + "inFlightSinceMs": 1787224595805, // when the oldest sweep still running began + "inFlightMs": 4560000, // this pass has run 76 minutes "missedPasses": 76, "lastErrorClass": "TimeoutError" }, @@ -78,16 +79,26 @@ logic of its own by design: the boundary lives in one place, in this repo, with - **`consecutiveFailures` / `lastErrorClass`** — the failing case. During the outage this read 7 then 8 while `/healthz` said `ok: true` and published nothing but the string `degraded`. -- **`lastStartedAtMs` vs `lastCompletedAtMs`** — the *silent* case. A sweep that hangs takes neither - the success nor the failure path, so no state is written and every settled field keeps reading - green. `lastStarted > lastCompleted` is the only evidence that a pass is in flight, and `inFlightMs` - says for how long. +- **`inFlightSinceMs`, or `lastStartedAtMs` vs `lastCompletedAtMs`** — the *silent* case. A sweep that + hangs takes neither the success nor the failure path, so no state is written and every settled field + keeps reading green. `inFlightSinceMs` is the daemon saying outright when the oldest sweep still + running began; `inFlightMs` is its age. Where it is absent — a heartbeat written by a build before + #296 — fall back to `lastStarted > lastCompleted`, which infers the same thing from timestamp order. + Prefer the published field: once a sweep has passed its deadline (below) the wait records a failure + while the sweep underneath it keeps running, and order alone then reports nothing in flight. - **`fleetControlPlane`** — an `open` circuit fails every spawn and resume fast, so it gates dispatch as hard as a failing sweep. `closed` is the healthy value. - **`state: "stalled"`** — derived, not written: an in-flight pass older than ten sweep intervals. A cold container legitimately spends minutes in its first pass (#36 measured 61 minutes while the Relayfile mirror hydrated), so check `lastCompletedAtMs`: absent means "first pass since boot, still hydrating"; present and hours old means "was fine, then wedged". +- **How long a stall can last** — a sweep is bounded at `liveSubscription.reconcileTimeoutMs`, + 90 minutes by default (#296). On expiry the *wait* fails, so `consecutiveFailures` starts rising + and the loop schedules the next pass; the sweep itself is not cancelled, because it holds a durable + discovery lease, so `inFlightSinceMs` keeps ageing until it really finishes. A `stalled` state that + never turns into a rising `consecutiveFailures` therefore means the process is not running the loop + at all, which is a restart, not a wait. The deadline sits above #36's 61-minute measurement on + purpose: setting it below realistic cold-mirror hydration would turn a slow boot into a crash loop. ### Why `ok` stays `true` while `status` goes amber diff --git a/src/config/schema.ts b/src/config/schema.ts index b65bc0f5..c1276062 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -40,6 +40,26 @@ const subscriptionSchema = z.object({ assignees: z.array(z.string()).default([]), }).default({}) +/** + * Deadline for one readiness reconcile sweep (#296). + * + * This is a wedge backstop, not a latency target. An unbounded sweep stops the + * reconcile loop permanently and silently, because the timer re-arms only when + * the sweep settles — so the deadline exists to guarantee that it settles. + * + * DO NOT lower this to a small multiple of `reconcileIntervalMs`. Container + * disk is ephemeral, so the Relayfile mirror rehydrates on every boot, and #36 + * measured a real cold-mirror reconcile at 3,665,173 ms (61 minutes) in + * production. A deadline under realistic worst-case hydration converts a slow + * boot into a crash loop, which is worse than the hang it would be preventing. + * 90 minutes leaves roughly 47% headroom over that measurement. + * + * A stall is *reported* far sooner than it is killed — see + * `READINESS_RECONCILE_STALL_INTERVALS` — so operators do not wait 90 minutes + * to learn that a pass is stuck. + */ +export const DEFAULT_READINESS_RECONCILE_TIMEOUT_MS = 90 * 60_000 + const liveSubscriptionSchema = z.object({ transport: z.enum(['subscribe-and-poll', 'subscribe', 'poll']).default('subscribe-and-poll'), pollIntervalMs: z.number().int().min(50).default(5_000), @@ -47,6 +67,19 @@ const liveSubscriptionSchema = z.object({ replaySkewMarginMs: z.number().int().min(0).default(60_000), /** Independent source-of-truth sweep; live event watermarks remain a latency optimization. */ reconcileIntervalMs: z.number().int().min(50).default(60_000), + /** Bounds one sweep so a hung dependency call cannot stop the loop forever. */ + reconcileTimeoutMs: z.number().int().min(50).max(6 * 60 * 60_000) + .default(DEFAULT_READINESS_RECONCILE_TIMEOUT_MS), +}).superRefine((value, ctx) => { + // A deadline below the interval kills every pass that takes longer than one + // tick, which is most of them on a cold mirror. + if (value.reconcileTimeoutMs < value.reconcileIntervalMs) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['reconcileTimeoutMs'], + message: `reconcileTimeoutMs (${value.reconcileTimeoutMs}) must be at least reconcileIntervalMs (${value.reconcileIntervalMs})`, + }) + } }).default({}) export const DEFAULT_AGENT_HOLD_TIMEOUT_MS = 4 * 60 * 60_000 diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 8edb7837..da040dd7 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -16,7 +16,9 @@ import { parseGithubFactoryIssue, parseLinearIssue, readFactoryInFlightRegistry, + publicHealthFromHeartbeat, readFactoryLoopHeartbeat, + readinessReconcileInFlightMs, reapFactoryOrphansOnce, type FactoryConfig, type FactoryEventPayload, @@ -27,7 +29,7 @@ import { type WorkflowRunnerInput, } from '../index' import { changeEventPath } from './factory' -import type { AgentWorktree, AgentWorktreeCleanupInspection, AgentWorktreeManager, AgentWorktreeRepository, ChangeEvent, EventPage, GithubConnectionRead, GithubConnectionWrite, GithubIssueStatus, GithubPublishPullRequestInput, GithubWriteback, LinearWriteback, PreviewReference, PreviewStartInput, ProviderSyncStatus, SlackWriteback, SpawnInput, SpawnResult } from '../ports' +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' import { BatchTracker, issueKey } from './batch-tracker' @@ -12033,6 +12035,638 @@ describe('FactoryLoop', () => { }) }) + // #296: a `runOnce()` that never settles used to stop this loop permanently + // and silently. `#scheduleReadinessReconcile` re-arms only from + // `sweep.finally(...)`, and both state-writing paths run on settle, so a hang + // took neither: the subsystem reported `healthy` with zero failures while + // dispatching nothing, and only a process restart recovered it. + describe('bounded readiness reconciliation', () => { + class HangingDiscoveryStateStore extends InMemoryStateStore { + hangClaims = false + readonly hangStarted: Promise + readonly #releases: Array<() => void> = [] + #signalHangStarted: () => void = () => undefined + + constructor() { + super({ batchSize: 2 }) + this.hangStarted = new Promise((resolve) => { this.#signalHangStarted = resolve }) + } + + /** Frees every hung claim so teardown never inherits the wedge. */ + release(): void { + this.hangClaims = false + this.releaseParked() + } + + /** Frees the claims parked right now, leaving later ones to hang. */ + releaseParked(): void { + while (this.#releases.length > 0) this.#releases.pop()?.() + } + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + if (this.hangClaims) { + this.#signalHangStarted() + // Never settles — the shape of the un-timed-out dependency call that + // wedged production for 104 minutes. + await new Promise((resolve) => { this.#releases.push(resolve) }) + } + return await super.claimDiscoverySweep(workspaceId, owner, nowMs, leaseMs) + } + } + + it('rejects a never-settling sweep on its deadline and schedules the next pass', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 300 }, + }) + try { + stateStore.hangClaims = true + await stateStore.hangStarted + const sweepsWhileHung = factory.status().counters.readinessReconcileSweeps ?? 0 + + // The deadline is the only thing that can end this pass, and ending it + // is what re-arms the timer. Without it both numbers stay frozen and + // the subsystem keeps reporting healthy. + await vi.waitFor(() => { + const status = factory.status() + expect(status.readinessReconcile?.consecutiveFailures ?? 0).toBeGreaterThanOrEqual(1) + expect(status.readinessReconcile?.lastError) + .toMatch(/readiness reconcile sweep exceeded its 300ms deadline/) + expect(status.counters.readinessReconcileSweeps ?? 0).toBeGreaterThan(sweepsWhileHung) + }, { timeout: 3_000 }) + + // Repeated deadline expiry must carry the failure count past the same + // threshold a repeatedly-throwing sweep crosses. The published `state` + // is `stalled` rather than `degraded` because + // `derivedReadinessReconcileState` (#295/#300) outranks the failure + // ladder with the more specific fact — the count is what proves the + // failure path was reached, and it ships either way. + await vi.waitFor(() => { + const readiness = factory.status().readinessReconcile + expect(readiness?.consecutiveFailures ?? 0) + .toBeGreaterThanOrEqual(readiness?.failureThreshold ?? 3) + expect(readiness?.state).not.toBe('healthy') + }, { timeout: 5_000 }) + } finally { + stateStore.release() + // Drain the abandoned pass before teardown. It is still running by + // design — the deadline rejects the wait, not the sweep — and this + // coalesces onto it rather than leaking it into the next test. + await factory.runOnce().catch(() => undefined) + await factory.stop() + } + }) + + it('reports a pass still in flight past the stall threshold as stalled rather than healthy', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + // A deadline far past this test's horizon: `stalled` has to come from + // the in-flight duration alone, not as a side effect of the kill. + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }) + try { + stateStore.hangClaims = true + await stateStore.hangStarted + + await vi.waitFor(() => expect(factory.status().readinessReconcile).toMatchObject({ + state: 'stalled', + // Nothing has failed yet: this state is derived from the pass in + // flight, not from the last write of a settled one. + consecutiveFailures: 0, + lastStartedAtMs: expect.any(Number), + }), { timeout: 3_000 }) + } finally { + stateStore.release() + // Drain the abandoned pass before teardown. It is still running by + // design — the deadline rejects the wait, not the sweep — and this + // coalesces onto it rather than leaking it into the next test. + await factory.runOnce().catch(() => undefined) + await factory.stop() + } + }) + + // PR #301 review, Codex P1 + cubic P1. The deadline bounds the *wait*, not + // the sweep, so the abandoned `runOnce()` is still live. Shutdown must not + // complete underneath it: `stop()` disposes the fleet and releases dispatch + // lifecycle leases, and `#isPassFatalFailure` only fences a stopping sweep + // once something in it throws — a sweep whose dependency recovers cleanly + // sails past that check and dispatches through torn-down ports. + it('does not complete shutdown while a sweep abandoned by the deadline is still running', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + }) + let stopped = false + let stopping: Promise | undefined + try { + stateStore.hangClaims = true + await stateStore.hangStarted + await vi.waitFor( + () => expect(factory.status().readinessReconcile?.consecutiveFailures ?? 0).toBeGreaterThanOrEqual(1), + { timeout: 3_000 }, + ) + + stopping = factory.stop().then(() => { stopped = true }) + await new Promise((resolve) => setTimeout(resolve, 400)) + expect(stopped).toBe(false) + + stateStore.release() + await stopping + expect(stopped).toBe(true) + } finally { + stateStore.release() + await stopping + if (!stopped) await factory.stop() + } + }) + + // PR #301 review, cubic P2. Clearing the in-flight timestamp when the + // *wait* ends hid the fact that the underlying pass was still stuck, so a + // genuinely wedged Factory fell back to reporting `retrying`. + it('keeps reporting stalled while the sweep the deadline abandoned is still stuck', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + // Deadline (1000ms) far below the stall threshold (READINESS_RECONCILE_ + // STALL_INTERVALS = 10 x 400ms), so no single *wait* can last long enough + // to be reported stalled on its own. Reaching `stalled` here is only + // possible by carrying the abandoned sweep's own start time. + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 1_000 }, + }) + try { + stateStore.hangClaims = true + await stateStore.hangStarted + + await vi.waitFor( + () => expect(factory.status().readinessReconcile?.state).toBe('stalled'), + { timeout: 9_000 }, + ) + } finally { + stateStore.release() + await factory.runOnce().catch(() => undefined) + await factory.stop() + } + }) + + // PR #301 review round two, cubic P2. Every pass that coalesces onto the + // same hung `runOnce()` abandons its own wrapper, and keeping only the + // latest one advanced the stall age by two intervals every two intervals. + // At the tightest legal setting — deadline equal to the interval — the age + // can then never reach the stall threshold at all, so a permanently stuck + // Factory never reports `stalled`, skipping the early warning this exists + // to add. + it('ages the stall from the earliest abandoned sweep when the deadline equals the interval', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 400 }, + }) + try { + stateStore.hangClaims = true + await stateStore.hangStarted + + // All of these waits abandon the *same* stuck sweep, so its age is the + // honest one: measured from when the first pass began. + await vi.waitFor( + () => expect(factory.status().readinessReconcile?.state).toBe('stalled'), + { timeout: 9_000 }, + ) + } finally { + stateStore.release() + await factory.runOnce().catch(() => undefined) + await factory.stop() + } + }) + + // PR #301 review round three, cubic P2. Every expiry registered another + // record even though they all await the SAME hung `runOnce()`, so the + // retention that fixed the stall age grew without bound in exactly the + // scenario this PR is about: a loop that stays wedged indefinitely. + // Registering per underlying sweep instead of per wait is observable — + // the abandoned-sweep completion is reported once, not once per expiry. + it('registers one record per abandoned sweep however many deadlines expire on it', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const warn = vi.fn() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn, error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + }) + try { + stateStore.hangClaims = true + await stateStore.hangStarted + + // Several waits give up on the one stuck sweep. + await vi.waitFor( + () => expect(factory.status().counters.readinessReconcileDeadlineExceeded ?? 0) + .toBeGreaterThanOrEqual(3), + { timeout: 5_000 }, + ) + + stateStore.release() + await factory.runOnce().catch(() => undefined) + await vi.waitFor(() => expect(warn.mock.calls.filter( + ([message]) => message === '[factory] abandoned readiness sweep completed after its deadline', + )).toHaveLength(1), { timeout: 3_000 }) + } finally { + stateStore.release() + await factory.stop() + } + }) + + // PR #301 review round four, cubic P1. `runOnce()` does not coalesce a + // mismatched `dryRun`; it waits BEHIND that sweep. So at expiry the global + // in-flight handle can name an unrelated sweep the readiness pass is queued + // behind rather than the work it will actually do — and once that unrelated + // sweep settles, the readiness pass starts its own, untracked, after + // shutdown believed it had drained everything. + it('drains the readiness sweep queued behind a mismatched dry-run sweep', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + }) + let stopped = false + let stopping: Promise | undefined + let dryRun: Promise | undefined + try { + stateStore.hangClaims = true + // Occupies `#runOnceInFlight` with a sweep the readiness pass cannot + // coalesce onto, so the readiness pass queues behind it. + dryRun = factory.runOnce({ dryRun: true }).catch(() => undefined) + await stateStore.hangStarted + await vi.waitFor( + () => expect(factory.status().readinessReconcile?.consecutiveFailures ?? 0).toBeGreaterThanOrEqual(1), + { timeout: 3_000 }, + ) + + // Let the unrelated sweep finish. The readiness pass now runs its own + // sweep, which hangs in turn — and that is the work shutdown owes. + stateStore.releaseParked() + await dryRun + + stopping = factory.stop().then(() => { stopped = true }) + await new Promise((resolve) => setTimeout(resolve, 500)) + expect(stopped).toBe(false) + + stateStore.release() + await stopping + expect(stopped).toBe(true) + } finally { + stateStore.release() + await dryRun + await stopping + if (!stopped) await factory.stop() + } + }) + + // The seam between #296 and #295/#300. Both changes describe the same + // subsystem: this one knows what is still running, that one publishes the + // health an operator reads out-of-process. They have to be ONE view. + // The hard case is after a deadline expiry, when the wait has written a + // settle timestamp while its `runOnce()` is still stuck — timestamp order + // alone then says "nothing in flight", which is precisely the blindness + // #295 exists to cure. `inFlightSinceMs` is what makes them agree. + it('publishes an abandoned sweep to the deployed health projection as stalled', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new HangingDiscoveryStateStore() + const root = await mkdtemp(join(tmpdir(), 'factory-abandoned-health-')) + const heartbeatPath = join(root, 'heartbeat.json') + const factory = createFactory(config({ + issueSource: 'github', + loop: { heartbeatPath, registryPath: join(root, 'registry.json') }, + }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 100, reconcileTimeoutMs: 200 }, + }) + try { + stateStore.hangClaims = true + await stateStore.hangStarted + // Past the deadline, so the failure path has already written + // `lastFailureAtMs` while the sweep underneath it is still stuck. + await vi.waitFor( + () => expect(factory.status().readinessReconcile?.consecutiveFailures ?? 0) + .toBeGreaterThanOrEqual(1), + { timeout: 3_000 }, + ) + + const status = factory.status().readinessReconcile + expect(status?.inFlightSinceMs).toEqual(expect.any(Number)) + // Timestamp order on its own has already lost the sweep... + expect(readinessReconcileInFlightMs({ + lastStartedAtMs: status?.lastStartedAtMs, + lastCompletedAtMs: status?.lastCompletedAtMs, + lastFailureAtMs: status?.lastFailureAtMs, + }, Date.now())).toBeUndefined() + // ...and with the published start it has not. + expect(readinessReconcileInFlightMs(status ?? {}, Date.now())).toBeGreaterThan(0) + + // The out-of-process reader — `factory diagnose --deployed` reads + // exactly this file — must land on the same conclusion the daemon has. + await vi.waitFor(async () => { + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + expect(heartbeat?.readinessReconcile?.inFlightSinceMs).toEqual(expect.any(Number)) + const health = publicHealthFromHeartbeat(heartbeat!, Date.now() + 10 * 100 * 10) + expect(health.readinessReconcile?.state).toBe('stalled') + }, { timeout: 5_000 }) + } finally { + stateStore.release() + await factory.runOnce().catch(() => undefined) + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + // #299 taught the fleet client to notice a broker rebind, reconnect and + // retry the read. That recovery costs an extra round trip inside a call + // this sweep is already waiting on, so the two changes have to coexist: + // the deadline must not kill a rebind it is recovering from, and must + // still fire when the reconnect cannot succeed. #299's retry is one-shot + // and read-only by construction ("repeated failures cannot turn into a + // reconnect loop"), so it cannot outrun a deadline sized for #36. + describe('against the #299 broker-rebind recovery', () => { + class RebindingFleetClient extends FakeFleetClient { + rebindDelayMs = 0 + rebindsRemaining = 0 + failRoster = false + rosterCalls = 0 + + override async roster(): Promise { + this.rosterCalls += 1 + if (this.failRoster) throw new Error('no running broker') + if (this.rebindsRemaining > 0) { + this.rebindsRemaining -= 1 + // Stands in for detect-rebind, reconnect, reissue the read. + await new Promise((resolve) => setTimeout(resolve, this.rebindDelayMs)) + } + return await super.roster() + } + } + + it('lets a sweep that recovered from a rebind finish instead of killing it', async () => { + const path = githubIssueCompactPath('AgentWorkforce', 'pear', 299) + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const fleet = new RebindingFleetClient() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 4_000 }, + }) + try { + // A reconnect round trip an order of magnitude longer than the whole + // reconcile interval, still far inside the deadline. + fleet.rebindDelayMs = 600 + fleet.rebindsRemaining = 1 + mount.files.set(path, { + content: githubIssueFile(299, { + title: '[factory-e2e] Sweep that recovered from a broker rebind', + labels: ['factory', 'pear'], + }), + }) + + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-299-impl-pear', + 'ar-299-review-pear', + ]), { timeout: 8_000 }) + // The recovered pass was never abandoned mid-recovery. + expect(factory.status().counters.readinessReconcileDeadlineExceeded ?? 0).toBe(0) + expect(factory.status().counters.readinessReconcileErrors ?? 0).toBe(0) + } finally { + await factory.stop() + } + }) + + it('still fails the pass and re-arms when the broker cannot be reached at all', async () => { + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const fleet = new RebindingFleetClient() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 4_000 }, + }) + try { + fleet.failRoster = true + // An unreachable broker RAISES rather than hangs, so the pre-existing + // failure path carries it. The deadline is the backstop for a hang, + // not a substitute for an error, and must stay out of the way here. + await vi.waitFor(() => { + const readiness = factory.status().readinessReconcile + expect(readiness?.consecutiveFailures ?? 0) + .toBeGreaterThanOrEqual(readiness?.failureThreshold ?? 3) + expect(readiness?.state).not.toBe('healthy') + }, { timeout: 8_000 }) + expect(factory.status().counters.readinessReconcileDeadlineExceeded ?? 0).toBe(0) + // And the loop kept sweeping rather than stopping on the failures. + expect(factory.status().counters.readinessReconcileSweeps ?? 0).toBeGreaterThanOrEqual(3) + } finally { + await factory.stop() + } + }, 15_000) + }) + + // MUST-NOT-FIRE. #36 measured a real cold-mirror reconcile at 3,665,173 ms + // (61 minutes) in production: container disk is ephemeral, so the Relayfile + // mirror rehydrates on every boot. A default below that converts a slow + // boot into a crash loop, which is strictly worse than the bug fixed here. + it('defaults the sweep deadline above the measured worst-case cold-mirror hydration', () => { + expect(config().liveSubscription.reconcileTimeoutMs).toBeGreaterThan(3_665_173) + }) + + it('does not kill a slow-but-completing sweep that runs for many intervals', async () => { + const path = githubIssueCompactPath('AgentWorkforce', 'pear', 296) + + class SlowDiscoveryStateStore extends InMemoryStateStore { + slowClaimMs = 0 + appliedSlowClaim = false + onSlowClaim: () => void = () => undefined + + constructor() { + super({ batchSize: 2 }) + } + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + if (this.slowClaimMs > 0 && !this.appliedSlowClaim) { + this.appliedSlowClaim = true + // Publish the work inside the slow pass, so only a sweep that + // survived its own hydration can be the one that dispatches. + this.onSlowClaim() + await new Promise((resolve) => setTimeout(resolve, this.slowClaimMs)) + } + return await super.claimDiscoverySweep(workspaceId, owner, nowMs, leaseMs) + } + } + + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const fleet = new FakeFleetClient() + const stateStore = new SlowDiscoveryStateStore() + stateStore.onSlowClaim = () => { + mount.files.set(path, { + content: githubIssueFile(296, { + title: '[factory-e2e] Slow but completing readiness sweep', + labels: ['factory', 'pear'], + }), + }) + } + const info = vi.fn() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info, warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 4_000 }, + }) + try { + // 600ms of hydration against a 50ms interval — twelve intervals deep, + // far past the stall-report threshold — under a 4s deadline. + stateStore.slowClaimMs = 600 + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-296-impl-pear', + 'ar-296-review-pear', + ]), { timeout: 8_000 }) + + // Nothing was abandoned: no sweep ever took the failure path... + expect(factory.status().counters.readinessReconcileErrors ?? 0).toBe(0) + expect(factory.status().readinessReconcile?.lastError).toBeUndefined() + // ...and the long pass reached the *completion* path, carrying a + // duration that a naive `n * reconcileIntervalMs` deadline would have + // killed. Read from the sweep's own log rather than `lastDurationMs`, + // which the next 50ms pass immediately overwrites. + await vi.waitFor(() => { + const completions = info.mock.calls.filter( + ([message]) => message === '[factory] periodic readiness reconciliation completed', + ) + expect(completions.some( + ([, meta]) => ((meta as { durationMs?: number } | undefined)?.durationMs ?? 0) >= 600, + )).toBe(true) + }, { timeout: 3_000 }) + } finally { + await factory.stop() + } + }) + }) + it('derives a repo-scoped subscription and startup backfill from the simple hoopsheet config', async () => { class ScopedStartupMount extends CountingEventsMount { readonly listTreePrefixes: string[] = [] diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 1cb2dc1a..7bad0e00 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -2,7 +2,7 @@ import { randomUUID } from 'node:crypto' import { mkdir, readFile, writeFile } from 'node:fs/promises' import { dirname, isAbsolute, resolve } from 'node:path' -import { FactoryConfigSchema, type FactoryConfig } from '../config/schema' +import { DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, FactoryConfigSchema, type FactoryConfig } from '../config/schema' import { linearByStatePath, linearByIdPath, linearByUuidPath } from '../constants/linear' import { stateResolutionFromIds, type FactoryStateResolution } from '../linear/state-resolver' import { GithubMergeGate, closeProbePr, type GhRunner, type GithubMergeGate as GithubMergeGatePort } from '../github' @@ -563,6 +563,20 @@ class DispatchLifecycleClaimRefusedError extends Error { } } +/** + * The deadline that makes a hung sweep reachable by the existing recovery path. + * Its message is fully internal (one integer), so it is safe to persist into + * the operator-facing `readinessReconcile.lastError`. + */ +class ReadinessReconcileTimeoutError extends Error { + readonly code = 'FACTORY_READINESS_RECONCILE_TIMEOUT' + + constructor(readonly timeoutMs: number) { + super(`readiness reconcile sweep exceeded its ${timeoutMs}ms deadline`) + this.name = 'ReadinessReconcileTimeoutError' + } +} + const realClock: Clock = { now: () => Date.now(), sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), @@ -713,6 +727,34 @@ export class FactoryLoop implements Factory { #readinessReconcileTimer?: ReturnType #readinessReconcileInFlight?: Promise #readinessReconcileIntervalMs = 60_000 + #readinessReconcileTimeoutMs = DEFAULT_READINESS_RECONCILE_TIMEOUT_MS + // Set for exactly as long as a sweep is running. `state` is derived from + // this, so an in-flight pass can no longer masquerade as the last settled one. + #readinessReconcileInFlightSinceMs?: number + /** + * The work a deadline gave up waiting on. The deadline bounds the wait, not + * the sweep, so this is still live: shutdown has to drain it, and `state` has + * to keep counting from when it actually started. + * + * Two fields rather than a collection, because every live abandoned wait + * converges on the same sweep. They are all `runOnce()` calls with the same + * `dryRun`, so whatever they are queued behind, the first one out starts the + * sweep and the rest coalesce onto it — they settle together. So the newest + * wait is a sufficient drain target, and the earliest start is the honest + * age. Both matter (#301 review): keeping only the newest start advanced the + * age by two intervals every two intervals, so at `reconcileTimeoutMs === + * reconcileIntervalMs` it never reached three and `stalled` was never + * reported; keeping one record per wait grew without bound in exactly the + * never-settling case this change exists for. + * + * The wait, deliberately, and never `#runOnceInFlight`: a mismatched-`dryRun` + * sweep is waited BEHIND rather than coalesced onto, so that handle can name + * an unrelated sweep, and the readiness pass would then start its own work + * after shutdown believed it had drained everything. The wait covers the + * queueing and the sweep it eventually runs, in every branch. + */ + #readinessReconcileAbandonedWait?: Promise + #readinessReconcileAbandonedSinceMs?: number #readinessReconcileConsecutiveFailures = 0 #readinessReconcileLastDurationMs?: number #readinessReconcileLastStartedAtMs?: number @@ -1240,6 +1282,14 @@ export class FactoryLoop implements Factory { if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) this.#previewSweepTimer = undefined await this.#readinessReconcileInFlight + // #301 review: the deadline ends the *wait*, so `#readinessReconcileInFlight` + // can settle with its `runOnce()` still live. Shutdown releases dispatch + // lifecycle leases and disposes ports below, and `#isPassFatalFailure` only + // fences a stopping sweep once something in it throws — so a sweep whose + // dependency recovers cleanly would otherwise dispatch through torn-down + // state. Draining here restores exactly the pre-deadline shutdown contract: + // stop() outlives the sweep it started. + await this.#readinessReconcileAbandonedWait await this.#previewSweepInFlight this.#stoppingHeartbeatRefreshActive = await this.#stopLiveHeartbeat('stopping') try { @@ -1403,6 +1453,9 @@ export class FactoryLoop implements Factory { const options = this.#liveOptions(overrides) this.#liveTransport = options.transport this.#readinessReconcileIntervalMs = options.reconcileIntervalMs + // `start()` overrides skip the schema's cross-field check, so re-apply its + // floor here: a deadline under one interval would kill every pass. + this.#readinessReconcileTimeoutMs = Math.max(options.reconcileTimeoutMs, options.reconcileIntervalMs) this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -1561,6 +1614,7 @@ export class FactoryLoop implements Factory { eventLimit: overrides.eventLimit ?? this.#config.liveSubscription.eventLimit, replaySkewMarginMs: overrides.replaySkewMarginMs ?? this.#config.liveSubscription.replaySkewMarginMs, reconcileIntervalMs: overrides.reconcileIntervalMs ?? this.#config.liveSubscription.reconcileIntervalMs, + reconcileTimeoutMs: overrides.reconcileTimeoutMs ?? this.#config.liveSubscription.reconcileTimeoutMs, } } @@ -1610,15 +1664,78 @@ export class FactoryLoop implements Factory { this.#readinessReconcileTimer.unref?.() } + /** + * Runs one sweep under a deadline (#296). + * + * The sweep itself cannot be cancelled — `runOnce()` owns a durable discovery + * lease and abandoning it mid-flight is not safe — so expiry rejects *this* + * wait and leaves the underlying pass to finish on its own. That is enough: + * the rejection is what reaches the failure path, which re-arms the timer. + * A later reconcile pass coalesces onto the still-running `runOnce()` and + * fails on its own deadline too, so a persistent hang keeps counting up to + * `degraded` instead of going quiet. + */ + async #runOnceWithReadinessDeadline(): Promise { + const timeoutMs = this.#readinessReconcileTimeoutMs + const startedAtMs = this.#clock.now() + const sweep = this.runOnce() + let timer: ReturnType | undefined + try { + return await new Promise((resolve, reject) => { + timer = setTimeout(() => { + this.#increment('readinessReconcileDeadlineExceeded') + if (this.#readinessReconcileAbandonedSinceMs === undefined) { + // `state` ages from the FIRST wait that gave up on this work, not + // from whenever the latest one began. + this.#readinessReconcileAbandonedSinceMs = startedAtMs + // The abandoned pass is still running against the live control + // plane. Report where it lands, so an operator can tell a + // dependency that recovered late from one that never answered. + // Attached once, so a wedge is reported once and not per expiry. + void sweep.then( + (report) => this.#logger.warn?.('[factory] abandoned readiness sweep completed after its deadline', { + timeoutMs, + overrunMs: this.#elapsedSince(startedAtMs) - timeoutMs, + dispatched: report.dispatched.length, + }), + (error: unknown) => this.#logger.warn?.('[factory] abandoned readiness sweep failed after its deadline', { + timeoutMs, + overrunMs: this.#elapsedSince(startedAtMs) - timeoutMs, + error: describeError(error).errorMessage, + }), + ).catch(() => undefined) + } + // Newest wait wins as the drain target: it settles no earlier than + // the ones before it, and clearing on it clears them all. + const wait: Promise = sweep.catch(() => undefined).then(() => { + if (this.#readinessReconcileAbandonedWait !== wait) return + this.#readinessReconcileAbandonedWait = undefined + this.#readinessReconcileAbandonedSinceMs = undefined + }) + this.#readinessReconcileAbandonedWait = wait + reject(new ReadinessReconcileTimeoutError(timeoutMs)) + }, timeoutMs) + timer.unref?.() + // Attaching handlers here is also what keeps a late rejection from the + // abandoned pass from surfacing as an unhandled rejection. + sweep.then(resolve, reject) + }) + } finally { + if (timer) clearTimeout(timer) + } + } + async #reconcileReadyIssues(): Promise { const startedAtMs = this.#clock.now() this.#readinessReconcileLastStartedAtMs = startedAtMs + this.#readinessReconcileInFlightSinceMs = startedAtMs this.#increment('readinessReconcileSweeps') this.#logger.info?.('[factory] periodic readiness reconciliation started', { intervalMs: this.#readinessReconcileIntervalMs, + timeoutMs: this.#readinessReconcileTimeoutMs, }) try { - const report = await this.runOnce() + const report = await this.#runOnceWithReadinessDeadline() this.#readinessReconcileConsecutiveFailures = 0 this.#readinessReconcileLastDurationMs = this.#elapsedSince(startedAtMs) this.#readinessReconcileLastCompletedAtMs = this.#clock.now() @@ -1660,6 +1777,10 @@ export class FactoryLoop implements Factory { consecutiveFailures: this.#readinessReconcileConsecutiveFailures, degraded: this.#readinessReconcileConsecutiveFailures >= READINESS_RECONCILE_FAILURE_THRESHOLD, }) + } finally { + // Cleared before the heartbeat write below, so a slow-but-successful pass + // does not stamp its own tail as `stalled`. + this.#readinessReconcileInFlightSinceMs = undefined } await this.#refreshLiveHeartbeat() } @@ -4609,8 +4730,20 @@ export class FactoryLoop implements Factory { #readinessReconcileStatus(): FactoryReadinessReconcileStatus { const consecutiveFailures = this.#readinessReconcileConsecutiveFailures + // #296 owns the numerator here, #295/#300 own the derivation. The earliest + // sweep still running — after a deadline expiry that is the abandoned one, + // not the current wait, or every expiry would restart the clock and a + // permanently stuck pass would read as merely `retrying`. + const inFlightSinceMs = Math.min( + this.#readinessReconcileInFlightSinceMs ?? Number.POSITIVE_INFINITY, + this.#readinessReconcileAbandonedSinceMs ?? Number.POSITIVE_INFINITY, + ) const settled: FactoryReadinessReconcileStatus['state'] = this.#startMode !== 'live' ? 'not-running' + // Failure-count ladder only. Precedence against `stalled` belongs to + // `derivedReadinessReconcileState`, which outranks everything here: a + // stall is the more specific fact, and `consecutiveFailures` ships + // alongside so nothing an alarm keyed on `degraded` needs is lost. : consecutiveFailures >= READINESS_RECONCILE_FAILURE_THRESHOLD ? 'degraded' : consecutiveFailures > 0 @@ -4623,6 +4756,7 @@ export class FactoryLoop implements Factory { // state from it rather than trusting the last write. const timestamps = { intervalMs: this.#readinessReconcileIntervalMs, + ...(Number.isFinite(inFlightSinceMs) ? { inFlightSinceMs } : {}), ...(this.#readinessReconcileLastStartedAtMs !== undefined ? { lastStartedAtMs: this.#readinessReconcileLastStartedAtMs } : {}), @@ -4656,6 +4790,7 @@ export class FactoryLoop implements Factory { consecutiveFailures, failureThreshold: READINESS_RECONCILE_FAILURE_THRESHOLD, intervalMs: this.#readinessReconcileIntervalMs, + ...(Number.isFinite(inFlightSinceMs) ? { inFlightSinceMs } : {}), ...(inFlightMs !== undefined ? { inFlightMs } : {}), ...(this.#readinessReconcileLastDurationMs !== undefined ? { lastDurationMs: this.#readinessReconcileLastDurationMs } diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index fcba744f..69bd7568 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -163,10 +163,17 @@ const enumValue = (value: unknown, allowed: readonly T[]): T | export function readinessReconcileInFlightMs( status: Pick< FactoryReadinessReconcileStatus, - 'lastStartedAtMs' | 'lastCompletedAtMs' | 'lastFailureAtMs' + 'inFlightSinceMs' | 'lastStartedAtMs' | 'lastCompletedAtMs' | 'lastFailureAtMs' >, nowMs: number, ): number | undefined { + // A daemon that publishes `inFlightSinceMs` knows what is still running and + // does not need the inference below. It is strictly better: a sweep whose + // wait ended on the #296 deadline writes a settle timestamp while its + // `runOnce()` keeps running, so timestamp order alone would call that stuck + // pass "not in flight" — the exact blindness this projection exists to cure. + const inFlightSinceMs = timestamp(status.inFlightSinceMs) + if (inFlightSinceMs !== undefined) return Math.max(0, nowMs - inFlightSinceMs) const startedAtMs = timestamp(status.lastStartedAtMs) if (startedAtMs === undefined) return undefined const settledAtMs = Math.max( @@ -188,7 +195,7 @@ export function readinessReconcileInFlightMs( export function derivedReadinessReconcileState( status: Pick< FactoryReadinessReconcileStatus, - 'state' | 'intervalMs' | 'lastStartedAtMs' | 'lastCompletedAtMs' | 'lastFailureAtMs' + 'state' | 'intervalMs' | 'inFlightSinceMs' | 'lastStartedAtMs' | 'lastCompletedAtMs' | 'lastFailureAtMs' >, nowMs: number, ): FactoryPublicSubsystemState | 'unknown' { diff --git a/src/types.ts b/src/types.ts index a35523c2..14dc0b59 100644 --- a/src/types.ts +++ b/src/types.ts @@ -108,6 +108,12 @@ export interface FactoryLiveSubscriptionOptions { replaySkewMarginMs: number /** Periodic source-of-truth readiness reconciliation, independent of event cursors/watermarks. */ reconcileIntervalMs: number + /** + * Deadline for one reconcile sweep. Expiry rejects the sweep, which is what + * routes a hang into the failure path that re-arms the loop (#296). Must be + * sized above realistic worst-case mirror hydration, not to the interval. + */ + reconcileTimeoutMs: number } /** @@ -172,6 +178,17 @@ export interface FactoryReadinessReconcileStatus { intervalMs?: number lastDurationMs?: number lastStartedAtMs?: number + /** + * When the oldest sweep still running actually began, published by a daemon + * that knows rather than inferred from timestamp order (#296). + * + * `lastStartedAtMs` is the start of the last *wait*. A wait that ends on its + * deadline writes a settle timestamp while its `runOnce()` keeps running, so + * order alone reports "nothing in flight" while work is still stuck. Readers + * prefer this and fall back to the order inference, which is all a heartbeat + * from a daemon that does not publish it can offer. + */ + inFlightSinceMs?: number lastCompletedAtMs?: number lastFailureAtMs?: number /** Age of a pass that started and has neither completed nor failed. */