diff --git a/docs/deployed-diagnostics.md b/docs/deployed-diagnostics.md index be1aa7c4..f4521bd3 100644 --- a/docs/deployed-diagnostics.md +++ b/docs/deployed-diagnostics.md @@ -69,11 +69,14 @@ logic of its own by design: the boundary lives in one place, in this repo, with "inFlightMs": 4560000, // this pass has run 76 minutes "missedPasses": 76, "lastErrorClass": "TimeoutError", - // The last COMPLETED sweep's arithmetic (#355). Absent until one completes. + // The last ENUMERATING sweep's arithmetic (#355). Absent until one enumerates. "candidates": 7, // work units it pulled and evaluated "dispatched": 0, // work units it dispatched "skipped": 7, // work units it saw and declined - "skipReasons": { "dispatch-terminal": 7 } + "skipReasons": { "dispatch-terminal": 7 }, + // When THOSE counts were measured. Not lastCompletedAtMs, which also + // advances on a deferred pass that enumerated nothing. + "lastEnumeratedAtMs": 1787224535802 }, "eventListener": { "state": "subscribed" }, "fleetControlPlane": { "state": "closed", "consecutiveFailures": 0, "failureThreshold": 3 } @@ -99,29 +102,61 @@ logic of its own by design: the boundary lives in one place, in this repo, with - `candidates > 0` — the sweep **saw** those issues and **rejected** them. The bug is in eligibility evaluation, and `skipReasons` names which gate. - `candidates == 0` — the sweep **never pulled** them. The bug is upstream, in discovery/ingestion. - - **the three fields absent entirely** — this daemon has not *completed* a sweep (or predates #355). - That is not a zero, and must not be read as one: it says nothing about either half. Check - `lastCompletedAtMs` and `inFlightMs`. - - They describe the last sweep that settled **successfully**, the same tense as `lastDurationMs`; - `lastCompletedAtMs` dates them. A pass that failed leaves them untouched rather than zeroing them. - -- **`discoveryDeferred: "sweep-in-flight"`** — the sweep returned immediately because another process - held the discovery lease, so it enumerated nothing. Without this, that pass is indistinguishable - from one that queried the provider and legitimately found no ready work: both publish - `candidates: 0`. + - **the three fields absent entirely** — this daemon has not completed a sweep that **enumerated** + (or predates #355). That is not a zero, and must not be read as one: it says nothing about either + half. Check `discoveryDeferred` first, then `lastCompletedAtMs`, `lastFailureAtMs`, and `inFlightMs` + to distinguish a completed deferral, a failure, work still running, and a pre-counter daemon. + + They describe the last sweep that settled successfully **and enumerated**. A pass that failed — + or that deferred — leaves them untouched rather than zeroing them. + +- **`lastEnumeratedAtMs`** — when those counts were measured, and the field to check before acting on + them. It is **not** `lastCompletedAtMs`: that one advances on every settled pass including a + deferred one, so on a daemon contending for the discovery lease the counts would otherwise sit + beside an ever-fresh completion stamp with no way to tell a measurement one interval old from one + four days old. Equal to `lastCompletedAtMs` on a daemon sweeping normally; where they differ, the + gap is exactly how stale the counts are. + +- **`discoveryDeferred: "sweep-in-flight"`** — the **most recent** pass returned immediately because + another process held the discovery lease, so it enumerated nothing. It is tracked apart from the + three counts, which describe the last sweep that actually enumerated: + + - **with the counts** — those numbers are from an *earlier* pass, not the one `lastCompletedAtMs` + dates. A deferred pass records only this marker; its zeroes measure nothing and must not + overwrite a real sweep's numbers, which under a persistently-held lease would erase them. + - **alone, with no counts** — nothing has enumerated successfully yet on this daemon. The most + recent pass deferred because another process held the lease; an earlier startup attempt may + instead have failed before it could publish an outcome. + + `lastCompletedAtMs` *does* move for a deferred pass. That is deliberate: the stall derivation above + reads it against `lastStartedAtMs`, so freezing it would report a daemon that is correctly + deferring to another owner as hung after ten intervals. - **`skipReasons`** — `skipped` split by a closed vocabulary (`FACTORY_SWEEP_SKIP_REASON_CODES`); zero-count codes are omitted, so an absent key is a zero, and - the counts always sum to `skipped`. `dispatch-terminal` and `dispatch-retry-limit` are the two that - never clear on their own — a work unit in either needs a human. `dispatch-backoff`, - `already-tracked` and `queued-or-escalated` resolve by themselves. `out-of-scope` and `not-ready` - mean the gate is working as configured and the issue does not match it — check the deployed - `safety` config against the issue rather than the daemon. + the counts always sum to `skipped`. The full vocabulary, grouped by what to do about it: + + | code | | + |---|---| + | `dispatch-terminal`, `dispatch-retry-limit` | **needs a human** — permanently declined, never clears on its own | + | `dispatch-backoff`, `dispatch-in-flight`, `already-tracked`, `queued-or-escalated` | transient; resolves by itself | + | `out-of-scope`, `not-ready`, `not-dispatchable` | the gate is working as configured and the issue does not match it — check the deployed `safety` config against the issue, not the daemon | + | `parked-dependency`, `dependency-cycle` | parked on other work; a cycle needs a human to break it | + | `read-failed`, `dispatch-failed` | per-item failures the sweep **absorbed and continued past** (#292/#297) — see below | + | `other` | a code this reader's vocabulary does not know, from a producer on another version | + + `read-failed` and `dispatch-failed` count work units an otherwise-**successful** pass gave up on + individually. Do **not** reach for `lastErrorClass` to explain them: that field describes a pass + that *failed as a whole*, and the success path clears it, so it is absent in exactly this scenario. + The per-item messages go to the container log (`[factory] relayfile shed a ready-issue read…`, + `[factory] skipped a work unit whose dispatch failed…`); the count here is what tells you to go + looking. A rising `read-failed` alongside `state: healthy` is the #297 shedding signature. Counts only, by construction: issue keys, paths and titles carry customer project and repository names and never cross onto this surface. The keys are rebuilt from the reader's own copy of the - vocabulary, so a record from another version cannot publish an arbitrary string as one. + vocabulary — anything unrecognised is counted under `other` rather than dropped, so the parts keep + summing to `skipped` — which is also why a record from another version cannot publish an arbitrary + string as a key. - **`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. diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index ac3f017c..590053a7 100644 --- a/src/cli/diagnose.test.ts +++ b/src/cli/diagnose.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest' +import { formatSweepOutcome } from './diagnose' import { runFleetCli } from './fleet' const BASE = 'https://factory.example.com' @@ -65,6 +66,57 @@ const healthy = { }, } +describe('formatSweepOutcome (#359)', () => { + it('attributes a count-free deferral only to the latest pass', () => { + const outcome = formatSweepOutcome({ + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + discoveryDeferred: 'sweep-in-flight', + }) + + expect(outcome).toBe( + 'nothing has enumerated successfully yet — the most recent pass deferred ' + + 'to another process holding the discovery lease', + ) + expect(outcome).not.toContain('every pass') + }) + + it('does not present an older daemon\'s unstamped deferred zeroes as a measurement', () => { + const outcome = formatSweepOutcome({ + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + candidates: 0, + dispatched: 0, + skipped: 0, + discoveryDeferred: 'sweep-in-flight', + }) + + expect(outcome).toBe( + 'not attributable (legacy deferred report has counts without an enumeration timestamp; ' + + 'the most recent pass deferred to another process holding the discovery lease)', + ) + expect(outcome).not.toContain('candidate(s)') + }) + + it('renders a rejected count snapshot as unknown rather than never enumerated', () => { + const outcome = formatSweepOutcome({ + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + discoveryDeferred: 'sweep-in-flight', + enumerationCountsInvalid: true, + }) + + expect(outcome).toBe( + 'not attributable (the report supplied an incomplete or invalid count snapshot; ' + + 'whether an earlier pass enumerated is unknown)', + ) + expect(outcome).not.toContain('nothing has enumerated successfully yet') + }) +}) + describe('factory diagnose --deployed (#295)', () => { it('reports a healthy deployed instance and exits zero without any credential', async () => { const seen: string[] = [] @@ -765,4 +817,3 @@ describe('factory diagnose --deployed (#295)', () => { } }) }) - diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index 2a1ed36a..ff12cc5a 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -327,28 +327,53 @@ function verdictFor(diagnosis: Omit { const watchStatePath = join(root, 'state.json') const state = () => new FileStateStore({ batchSize: 1, watchStatePath }) const gate = Promise.withResolvers() + const spawnStarted = Promise.withResolvers() class HangingSpawnFleetClient extends RemoteLifecycleFleetClient { override async spawn(input: SpawnInput): Promise { + spawnStarted.resolve() await gate.promise return super.spawn(input) } @@ -11980,16 +11982,17 @@ describe('FactoryLoop', () => { const key = dispatchIssueIdentity(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 }) + // `phase: dispatching` is also persisted before the spawn starts. Wait + // for the fleet call itself so this test exercises a genuinely late + // placement rather than racing an earlier ownership fence (#359 CI). + await spawnStarted.promise + const claimed = await state().getDispatchLifecycle('factory-test', key) + expect(claimed).toMatchObject({ phase: 'dispatching' }) // 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, + 'factory-test', key, claimed!, 'other-owner', Date.now() + 10 * 60_000, 10 * 60_000, ) expect(takeover.acquired).toBe(true) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 191d4fcd..caaeed2f 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -848,14 +848,14 @@ export class FactoryLoop implements Factory { #readinessReconcileLastError?: string #readinessReconcileLastErrorClass?: string /** - * The last *completed* sweep's arithmetic (#355). + * The last *enumerating* sweep's arithmetic (#355). * - * Held as one record rather than four fields so it can only ever be replaced + * Held as one record rather than three fields so it can only ever be replaced * whole: publishing a `dispatched` from one pass beside a `candidates` from * another would be worse than publishing neither, since the whole use of * these numbers is comparing them to each other. * - * `undefined` until a sweep completes, and never initialised to zeroes — + * `undefined` until a sweep enumerates, and never initialised to zeroes — * "this daemon has not finished a sweep" and "a sweep finished and found * nothing" are the two readings #355 has to tell apart. */ @@ -864,8 +864,29 @@ export class FactoryLoop implements Factory { dispatched: number skipped: number skipReasons: Partial> - discoveryDeferred?: 'sweep-in-flight' + /** + * When the pass these counts describe finished enumerating (#359 review). + * + * Retaining counts across a deferral without this left them with no time + * coordinate: `lastCompletedAtMs` moves on every deferred pass, so the + * payload paired arbitrarily old counts with a fresh completion stamp and + * a reader could not tell a measurement one interval old from one four + * days old. It belongs to this record, and is replaced with it. + */ + enumeratedAtMs: number } + /** + * Whether the MOST RECENT pass deferred, tracked apart from the counts above + * (#358 review, CodeRabbit — Major, and right). + * + * A deferred pass enumerates nothing and settles in milliseconds, so folding + * it into the snapshot overwrote the last real sweep's numbers with zeroes. + * On a container where another process holds the lease for any length of time + * — the #347/#349 condition — every pass would publish `candidates: 0` and the + * last actual enumeration would be unrecoverable, destroying the measurement + * this whole change exists to provide. + */ + #readinessReconcileLastSweepDeferred?: 'sweep-in-flight' readonly #liveEventQueue: ChangeEvent[] = [] #liveEventDrainScheduled = false #liveEventDrainActive = false @@ -1635,9 +1656,11 @@ export class FactoryLoop implements Factory { // cold container it is the first — and for the next interval, only — // sweep whose counts exist. Leaving it unrecorded would make a daemon // that has completed a full pass still read as "never ran" (#355). - this.#recordReadinessSweepOutcome(await this.runOnce()) + const report = await this.runOnce() this.#readinessReconcileLastDurationMs = this.#elapsedSince(backfillStartedAtMs) - this.#readinessReconcileLastCompletedAtMs = this.#clock.now() + const completedAtMs = this.#clock.now() + this.#readinessReconcileLastCompletedAtMs = completedAtMs + this.#recordReadinessSweepOutcome(report, completedAtMs) } catch (error) { this.#readinessReconcileLastDurationMs = this.#elapsedSince(backfillStartedAtMs) this.#readinessReconcileLastFailureAtMs = this.#clock.now() @@ -1875,20 +1898,26 @@ export class FactoryLoop implements Factory { const report = await this.#runOnceWithReadinessDeadline() this.#readinessReconcileConsecutiveFailures = 0 this.#readinessReconcileLastDurationMs = this.#elapsedSince(startedAtMs) - this.#readinessReconcileLastCompletedAtMs = this.#clock.now() + const completedAtMs = this.#clock.now() + this.#readinessReconcileLastCompletedAtMs = completedAtMs this.#readinessReconcileLastError = undefined this.#readinessReconcileLastErrorClass = undefined // The three integers below have gone to stdout since this loop existed, // and stdout does not reach the deployed container's operator (#355). // Publishing them is what lets a reader tell a sweep that saw eligible // work and rejected it from one that never pulled it at all. - this.#recordReadinessSweepOutcome(report) + this.#recordReadinessSweepOutcome(report, completedAtMs) this.#logger.info?.('[factory] periodic readiness reconciliation completed', { durationMs: this.#readinessReconcileLastDurationMs, candidates: report.pulled.length, dispatched: report.dispatched.length, skipped: report.skipped.length, - skipReasons: this.#readinessReconcileLastSweep?.skipReasons, + // THIS pass's breakdown, never the retained snapshot (#359 review, + // codex P2). Logging the retained one beside a deferred pass's zeroes + // produced a line that contradicted its own arithmetic — + // `skipped: 0` next to a non-empty breakdown — and this log is what a + // local operator reads. + skipReasons: factorySweepSkipReasonCounts(report.skipped), discoveryDeferred: report.discoveryDeferred, }) } catch (error) { @@ -1911,6 +1940,10 @@ export class FactoryLoop implements Factory { this.#readinessReconcileLastDurationMs = this.#elapsedSince(startedAtMs) this.#readinessReconcileLastFailureAtMs = this.#clock.now() this.#readinessReconcileLastError = errorMessage + // This failure is now the latest settled pass. A deferral marker left by + // an older pass would falsely describe this one as lease contention when + // the timestamps/error below prove that it acquired the lease and failed. + this.#readinessReconcileLastSweepDeferred = undefined // The class, unlike the message, is publishable: #295 puts it on the // unauthenticated health surface through the same allowlist. this.#readinessReconcileLastErrorClass = telemetryErrorClass(error) @@ -5033,16 +5066,32 @@ export class FactoryLoop implements Factory { * * Only successful passes reach here: a pass that threw has no report, and * inventing zeroes for it would publish "found nothing" for a sweep that - * never got to look. The previous pass's numbers stay put instead, dated by - * `lastCompletedAtMs`, which is the honest reading. + * never got to look. The previous enumerating pass's numbers stay put + * instead, dated by `lastEnumeratedAtMs`, which is the honest reading. + * + * A deferred pass gets the same treatment for the same reason. It settles + * successfully, and `lastCompletedAtMs` moves — deliberately, because the + * #295/#296 stall derivation reads that timestamp against `lastStartedAtMs`, + * and freezing it would report a functioning daemon as a hung one after ten + * intervals of deferring correctly to another owner. But it enumerated + * nothing, so its zeroes are not a measurement of anything and must not + * replace one. Only the marker is recorded. */ - #recordReadinessSweepOutcome(report: IterationReport): void { + #recordReadinessSweepOutcome(report: IterationReport, completedAtMs: number): void { + if (report.discoveryDeferred) { + this.#readinessReconcileLastSweepDeferred = report.discoveryDeferred + return + } + this.#readinessReconcileLastSweepDeferred = undefined this.#readinessReconcileLastSweep = { candidates: report.pulled.length, dispatched: report.dispatched.length, skipped: report.skipped.length, skipReasons: factorySweepSkipReasonCounts(report.skipped), - ...(report.discoveryDeferred ? { discoveryDeferred: report.discoveryDeferred } : {}), + // The caller's completion stamp, not a fresh clock read: on a pass that + // enumerated, `lastEnumeratedAtMs` and `lastCompletedAtMs` describe the + // same instant and must not drift apart by a tick. + enumeratedAtMs: completedAtMs, } } @@ -5133,11 +5182,14 @@ export class FactoryLoop implements Factory { ...(Object.keys(this.#readinessReconcileLastSweep.skipReasons).length > 0 ? { skipReasons: { ...this.#readinessReconcileLastSweep.skipReasons } } : {}), - ...(this.#readinessReconcileLastSweep.discoveryDeferred - ? { discoveryDeferred: this.#readinessReconcileLastSweep.discoveryDeferred } - : {}), + lastEnumeratedAtMs: this.#readinessReconcileLastSweep.enumeratedAtMs, } : {}), + // Independent of the trio: a daemon whose FIRST pass deferred has no + // counts to publish and still needs to say why. + ...(this.#readinessReconcileLastSweepDeferred + ? { discoveryDeferred: this.#readinessReconcileLastSweepDeferred } + : {}), ...(this.#readinessReconcileLastError ? { lastError: this.#readinessReconcileLastError } : {}), ...(this.#readinessReconcileLastErrorClass ? { lastErrorClass: this.#readinessReconcileLastErrorClass } diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 26c4221e..78272975 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -779,6 +779,7 @@ describe('sweep counters on the public surface (#355)', () => { // "candidates minus dispatched" arithmetic that the missing field makes // wrong, so the group travels whole or not at all. it('drops a partial trio rather than publishing a misleading fragment', () => { + expect(swept({ candidates: 4 })).toMatchObject({ enumerationCountsInvalid: true }) expect(Object.hasOwn(swept({ candidates: 4 }) ?? {}, 'candidates')).toBe(false) expect(Object.hasOwn(swept({ candidates: 4, dispatched: 1 }) ?? {}, 'candidates')).toBe(false) expect(swept({ candidates: 4, dispatched: 1, skipped: 3 })).toMatchObject({ @@ -788,9 +789,42 @@ describe('sweep counters on the public surface (#355)', () => { }) }) + it('distinguishes a rejected deferred count snapshot from a genuine count-free deferral', () => { + const rejected = swept({ + candidates: 4, + dispatched: 'invalid' as unknown as number, + discoveryDeferred: 'sweep-in-flight', + }) + expect(rejected).toMatchObject({ + discoveryDeferred: 'sweep-in-flight', + enumerationCountsInvalid: true, + }) + expect(Object.hasOwn(rejected ?? {}, 'candidates')).toBe(false) + + const normalizedAgain = normalizePublicHealth({ + schemaVersion: 1, + ok: true, + status: 'ok', + stale: false, + loopStatus: 'running', + degradedSubsystems: [], + readinessReconcile: rejected, + }) + expect(normalizedAgain?.readinessReconcile).toMatchObject({ + discoveryDeferred: 'sweep-in-flight', + enumerationCountsInvalid: true, + }) + }) + it('names the deferred sweep, so a zero from a held lease is not read as an empty provider', () => { expect(swept({ candidates: 0, dispatched: 0, skipped: 0, discoveryDeferred: 'sweep-in-flight' })) .toMatchObject({ candidates: 0, discoveryDeferred: 'sweep-in-flight' }) + // Independent of the trio (#358 review). A daemon whose first pass deferred + // has no counts to publish and still has to say why, so dropping the marker + // with the counts would leave the only surface silent about it. + const noCounts = swept({ discoveryDeferred: 'sweep-in-flight' }) + expect(noCounts?.discoveryDeferred).toBe('sweep-in-flight') + expect(Object.hasOwn(noCounts ?? {}, 'candidates')).toBe(false) // Only the one value the vocabulary has. expect(swept({ candidates: 0, diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index 7942db3d..68bb6c39 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -203,11 +203,11 @@ const skipReasonCounts = ( } /** - * The last completed sweep's arithmetic, published (#355). + * The last enumerating sweep's arithmetic, published (#355). * * Deliberately NOT `counter()`: that coerces an absent field to `0`, which - * would make a daemon that has never completed a sweep indistinguishable from - * one that completed a sweep and found nothing. Those are the two halves of + * would make a daemon that has never enumerated a sweep indistinguishable from + * one that enumerated a sweep and found nothing. Those are the two halves of * the split this block exists to make, so the three fields travel together — * all present, or none — and a zero is published as a zero. */ @@ -223,21 +223,39 @@ const sweepOutcome = ( skipped?: unknown skipReasons?: unknown discoveryDeferred?: unknown + lastEnumeratedAtMs?: unknown + enumerationCountsInvalid?: unknown }, ): Partial> => { + // Independent of the trio (#358 review, CodeRabbit): the counts describe the + // last sweep that ENUMERATED, and this describes the most recent pass. A + // daemon whose first pass deferred has no counts and still has to say why, + // and one that deferred after a real sweep publishes both — which is the + // pairing that tells a reader the numbers are from an earlier pass. + const deferred = status.discoveryDeferred === 'sweep-in-flight' + ? { discoveryDeferred: 'sweep-in-flight' as const } + : {} const candidates = optionalCount('candidates', status.candidates) const dispatched = optionalCount('dispatched', status.dispatched) const skipped = optionalCount('skipped', status.skipped) + const suppliedCounts = status.enumerationCountsInvalid === true || + status.candidates !== undefined || + status.dispatched !== undefined || + status.skipped !== undefined // A record carrying only some of the three is a producer we do not // understand; publishing the fragment would invite exactly the arithmetic // ("candidates minus dispatched") that the missing field makes wrong. if (candidates.candidates === undefined || dispatched.dispatched === undefined || skipped.skipped === undefined) { - return {} + return { + ...deferred, + ...(suppliedCounts ? { enumerationCountsInvalid: true as const } : {}), + } } const skipReasons = skipReasonCounts(status.skipReasons) return { @@ -245,9 +263,11 @@ const sweepOutcome = ( ...dispatched, ...skipped, ...(skipReasons ? { skipReasons } : {}), - ...(status.discoveryDeferred === 'sweep-in-flight' - ? { discoveryDeferred: 'sweep-in-flight' as const } - : {}), + // Part of the same atomic snapshot as the counts: it is what dates them, + // and without it retained counts have no freshness a reader can recover + // (#359 review). + ...optionalTimestamp('lastEnumeratedAtMs', status.lastEnumeratedAtMs), + ...deferred, } } diff --git a/src/orchestrator/sweep-counters.test.ts b/src/orchestrator/sweep-counters.test.ts index f8816409..9c5ba677 100644 --- a/src/orchestrator/sweep-counters.test.ts +++ b/src/orchestrator/sweep-counters.test.ts @@ -184,6 +184,9 @@ describe('readiness sweep counters (#355)', () => { expect(spawns).toEqual(['ar-901-impl-pear', 'ar-901-review', 'ar-902-impl-pear', 'ar-902-review']) expect(status).toMatchObject({ candidates: 2, dispatched: 2, skipped: 0 }) expect(published(status)).toMatchObject({ candidates: 2, dispatched: 2, skipped: 0 }) + // On a pass that enumerated, the two stamps describe the same instant. + expect(status.lastEnumeratedAtMs).toBe(status.lastCompletedAtMs) + expect(published(status).lastEnumeratedAtMs).toBe(status.lastCompletedAtMs) }) it('publishes zero — not undefined — for a sweep that completed and found nothing', async () => { @@ -239,6 +242,7 @@ describe('readiness sweep counters (#355)', () => { const status = factory.status().readinessReconcile expect(status?.state).toBe('not-running') expect(status && Object.hasOwn(status, 'candidates')).toBe(false) + expect(status && Object.hasOwn(status, 'lastEnumeratedAtMs')).toBe(false) // The projection must not invent a zero for it either — that would make // an instance that has never swept indistinguishable from one that swept // and found nothing, which is the ambiguity #355 was stuck on. @@ -323,12 +327,13 @@ describe('readiness sweep counters (#355)', () => { } const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const stateStore = new LeaseHeldElsewhereStateStore({ batchSize: 4 }) const factory = createFactory( config({ loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') } }), { mount: new FakeMountClient({ [issuePath(941)]: issueFile(941) }), fleet: new FakeFleetClient(), - stateStore: new LeaseHeldElsewhereStateStore({ batchSize: 4 }), + stateStore, triage: new StaticTriage(), logger: {}, }, @@ -339,23 +344,275 @@ describe('readiness sweep counters (#355)', () => { liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 600_000 }, }) const status = factory.status().readinessReconcile + // Nothing has enumerated, so there are NO counts to publish — and the + // marker still has to say why. A `0` here would claim a sweep queried the + // provider and found nothing, which is the opposite diagnosis. + expect(status).toMatchObject({ state: 'healthy', discoveryDeferred: 'sweep-in-flight' }) + expect(status && Object.hasOwn(status, 'candidates')).toBe(false) + const readiness = published(status!) + expect(readiness.discoveryDeferred).toBe('sweep-in-flight') + expect(Object.hasOwn(readiness, 'candidates')).toBe(false) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('a deferred pass does not overwrite the last enumerating sweep it followed', async () => { + // #358 review, CodeRabbit (Major). A deferred pass settles healthy in + // milliseconds having read nothing. Folding its zeroes into the snapshot + // erased the last real measurement — and where another process holds the + // lease for any length of time, EVERY pass would publish `candidates: 0` + // and the numbers this change exists to provide would be unrecoverable. + class DeferrableStateStore extends InMemoryStateStore { + deferClaims = false + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const claim = await super.claimDiscoverySweep( + workspaceId, + this.deferClaims ? 'another-process' : owner, + nowMs, + leaseMs, + ) + return this.deferClaims ? { ...claim, acquired: false, lease: undefined } : claim + } + } + + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const mount = new FakeMountClient({ [issuePath(961)]: issueFile(961) }) + const stateStore = new DeferrableStateStore({ batchSize: 4 }) + const factory = createFactory( + config({ loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') } }), + { mount, fleet: new FakeFleetClient(), stateStore, triage: new StaticTriage(), logger: {} }, + ) + try { + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + // A real enumeration first. + await vi.waitFor(() => expect(factory.status().readinessReconcile).toMatchObject({ + candidates: 1, + dispatched: 1, + skipped: 0, + }), { timeout: 5_000 }) + + stateStore.deferClaims = true + + // Then deferred passes, for long enough that several land. + await vi.waitFor(() => { + const deferred = factory.status().readinessReconcile + expect(deferred?.discoveryDeferred).toBe('sweep-in-flight') + expect(deferred?.lastEnumeratedAtMs).toBeDefined() + expect(deferred?.lastCompletedAtMs).toBeGreaterThan(deferred?.lastEnumeratedAtMs ?? 0) + }, { timeout: 5_000 }) + + const status = factory.status().readinessReconcile + // Capture the enumerating baseline only after deferral is observable. A + // 50ms pass can land between the first wait and flipping `deferClaims`, + // so a pre-toggle completion timestamp is inherently racy. + const completedAtMs = status?.lastEnumeratedAtMs + expect(completedAtMs).toBeDefined() + // The measurement survives, and the marker says it is from an earlier pass. expect(status).toMatchObject({ state: 'healthy', - candidates: 0, - dispatched: 0, + candidates: 1, + dispatched: 1, skipped: 0, discoveryDeferred: 'sweep-in-flight', }) + // `lastCompletedAtMs` DOES move for a deferred pass, deliberately: the + // #295/#296 stall derivation reads it against `lastStartedAtMs`, so + // freezing it would report a daemon that is correctly deferring as hung + // after ten intervals. + expect(status?.lastCompletedAtMs).toBeGreaterThan(completedAtMs ?? 0) + // ...which is exactly why the counts need their own stamp (#359 review). + // It stays pinned to the pass that enumerated, so the gap between the + // two IS the staleness of the retained measurement — without it, old + // counts sat beside an ever-fresh completion time and a reader could not + // tell a measurement one interval old from one four days old. + expect(status?.lastEnumeratedAtMs).toBe(completedAtMs) + expect(status?.lastEnumeratedAtMs).toBeLessThan(status?.lastCompletedAtMs ?? 0) expect(published(status!)).toMatchObject({ - candidates: 0, + candidates: 1, + dispatched: 1, discoveryDeferred: 'sweep-in-flight', + lastEnumeratedAtMs: completedAtMs, }) + + // And it holds across FURTHER deferrals rather than creeping forward. + const pinned = status?.lastEnumeratedAtMs + await vi.waitFor(() => { + expect(factory.status().readinessReconcile?.lastCompletedAtMs) + .toBeGreaterThan(status?.lastCompletedAtMs ?? 0) + }, { timeout: 5_000 }) + expect(factory.status().readinessReconcile?.lastEnumeratedAtMs).toBe(pinned) + expect(factory.status().readinessReconcile?.candidates).toBe(1) } finally { await factory.stop() await rm(root, { recursive: true, force: true }) } }) + it('logs THIS pass\'s breakdown on a deferred completion, never the retained one', async () => { + // #359 review, codex P2. The completion log is what a local operator reads, + // and it drew `skipReasons` from the retained snapshot while drawing the + // counts from the current report. On a deferred pass that printed + // `skipped: 0` beside a non-empty breakdown — a line contradicting its own + // arithmetic, on the surface meant to explain the arithmetic. + class DeferrableStateStore extends InMemoryStateStore { + deferClaims = false + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const claim = await super.claimDiscoverySweep( + workspaceId, + this.deferClaims ? 'another-process' : owner, + nowMs, + leaseMs, + ) + return this.deferClaims ? { ...claim, acquired: false, lease: undefined } : claim + } + } + + const completions: Array> = [] + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const stateStore = new DeferrableStateStore({ batchSize: 4 }) + const factory = createFactory( + config({ loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') } }), + { + mount: new FakeMountClient({ + [issuePath(971)]: issueFile(971), + [issuePath(972)]: issueFile(972, 'Ordinary product issue, not in factory scope'), + }), + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + logger: { + info: (message: string, detail?: unknown) => { + if (message === '[factory] periodic readiness reconciliation completed') { + completions.push(detail as Record) + } + }, + }, + }, + ) + try { + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + // One real pass that skips something, so a retained breakdown exists. + // (The startup backfill dispatches 971 and skips 972; a periodic pass + // then skips both — 971 as `already-tracked` — so assert on the shape, + // not on an exact count.) + await vi.waitFor(() => { + expect(completions.some((entry) => (entry.skipped as number) > 0)).toBe(true) + }, { timeout: 5_000 }) + const enumerated = completions.find((entry) => (entry.skipped as number) > 0) + expect(enumerated?.skipReasons).toMatchObject({ 'out-of-scope': 1 }) + // Internally consistent: the breakdown sums to the count beside it. + expect(Object.values(enumerated?.skipReasons as Record) + .reduce((sum, n) => sum + n, 0)).toBe(enumerated?.skipped) + + stateStore.deferClaims = true + await vi.waitFor(() => { + expect(completions.some((entry) => entry.discoveryDeferred === 'sweep-in-flight')).toBe(true) + }, { timeout: 5_000 }) + + // Every deferred line is internally consistent: zero skips, no breakdown. + const deferredLines = completions.filter((entry) => entry.discoveryDeferred === 'sweep-in-flight') + expect(deferredLines.length).toBeGreaterThan(0) + for (const entry of deferredLines) { + expect(entry).toMatchObject({ candidates: 0, dispatched: 0, skipped: 0 }) + expect(entry.skipReasons).toEqual({}) + } + // ...while the published surface still retains the real measurement. + const retained = factory.status().readinessReconcile + expect(retained?.candidates).toBe(2) + expect((retained?.skipped ?? 0)).toBeGreaterThan(0) + expect(retained?.skipReasons).toMatchObject({ 'out-of-scope': 1 }) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + // Two sequential waits on a live 50ms loop; vitest's 5s default is not a + // budget this fits in, and this repo configures no `testTimeout`. + }, 30_000) + + it('clears an older deferral marker when the next pass fails', async () => { + class ControllableStateStore extends InMemoryStateStore { + mode: 'enumerate' | 'defer' | 'fail' = 'enumerate' + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + if (this.mode === 'fail') throw new Error('discovery failed after deferral') + const claim = await super.claimDiscoverySweep( + workspaceId, + this.mode === 'defer' ? 'another-process' : owner, + nowMs, + leaseMs, + ) + return this.mode === 'defer' ? { ...claim, acquired: false, lease: undefined } : claim + } + } + + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const stateStore = new ControllableStateStore({ batchSize: 4 }) + const factory = createFactory( + config({ loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') } }), + { + mount: new FakeMountClient({ [issuePath(981)]: issueFile(981) }), + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + logger: {}, + }, + ) + try { + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + await vi.waitFor(() => { + expect(factory.status().readinessReconcile?.lastEnumeratedAtMs).toBeDefined() + }, { timeout: 5_000 }) + + stateStore.mode = 'defer' + await vi.waitFor(() => { + expect(factory.status().readinessReconcile?.discoveryDeferred).toBe('sweep-in-flight') + }, { timeout: 5_000 }) + + stateStore.mode = 'fail' + await vi.waitFor(() => { + const failed = factory.status().readinessReconcile + expect(failed?.lastErrorClass).toBe('Error') + expect(failed?.lastFailureAtMs).toBeDefined() + expect(failed && Object.hasOwn(failed, 'discoveryDeferred')).toBe(false) + }, { timeout: 5_000 }) + + // The last real measurement remains useful; only the stale attribution + // to lease contention is removed by the newer failed pass. + expect(factory.status().readinessReconcile).toMatchObject({ candidates: 1 }) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 30_000) + it('publishes counts only: no issue key, path or title reaches the unauthenticated record', async () => { const { status } = await sweepReadiness({ [issuePath(921)]: issueFile(921), diff --git a/src/types.ts b/src/types.ts index f9a1d94f..1167f8fc 100644 --- a/src/types.ts +++ b/src/types.ts @@ -231,22 +231,23 @@ export interface FactoryReadinessReconcileStatus { /** Age of a pass that started and has neither completed nor failed. */ inFlightMs?: number /** - * Work units the last *completed* sweep pulled and evaluated (#355). + * Work units the last *enumerating* sweep pulled and evaluated (#355). * - * Same tense as `lastDurationMs`: written when a pass settles successfully, - * left alone by a pass that failed or is still running, so - * `lastCompletedAtMs` says which pass these describe. + * Written when a pass settles successfully AND enumerated; left alone by a + * pass that failed, one still running, and one that deferred to another + * process's lease. `lastEnumeratedAtMs` — not `lastCompletedAtMs` — is what + * dates them, because the latter advances on deferred passes too. * * Optional, and never defaulted to zero. A sweep that ran and found nothing - * publishes `0`; a daemon that has not completed a sweep publishes nothing + * publishes `0`; a daemon that has not enumerated a sweep publishes nothing * at all, and the whole point of the field is that those two are different * facts — `candidates: 0` blames discovery, an absent `candidates` blames * nobody yet. */ candidates?: number - /** Work units the last completed sweep actually dispatched. */ + /** Work units the last enumerating sweep actually dispatched. */ dispatched?: number - /** Work units the last completed sweep saw and declined. */ + /** Work units the last enumerating sweep saw and declined. */ skipped?: number /** * `skipped` split by cause. Zero-count codes are omitted; the codes @@ -254,12 +255,30 @@ export interface FactoryReadinessReconcileStatus { */ skipReasons?: Partial> /** - * The last completed sweep never enumerated anything: another process held - * the discovery lease, so it returned an empty report immediately. + * When the pass the counts above describe finished enumerating (#359 review). + * + * NOT `lastCompletedAtMs`, and the difference is the point. That timestamp + * advances on every settled pass including a deferred one, which enumerates + * nothing; this one advances only when a pass actually enumerated. Equal on + * a daemon that is sweeping normally; where they differ, the gap is how + * stale the counts are — the freshness a reader otherwise could not + * recover, since retained counts sat beside an ever-fresh completion stamp. + */ + lastEnumeratedAtMs?: number + /** + * The MOST RECENT pass never enumerated anything: another process held the + * discovery lease, so it returned an empty report immediately. * * Without this, that pass is indistinguishable from a sweep that queried the - * provider and legitimately found no ready work — both publish + * provider and legitimately found no ready work — both would publish * `candidates: 0` — and those are opposite diagnoses (#355). + * + * Independent of the counts above, which describe the last sweep that + * actually enumerated. A deferred pass records only this marker: its zeroes + * measure nothing and must not overwrite a real sweep's numbers, which on a + * persistently-held lease would erase them entirely (#358 review). So the two + * together mean "the counts are from an earlier pass"; this one alone means + * nothing has enumerated yet. */ discoveryDeferred?: 'sweep-in-flight' /** Free text; authenticated surfaces only. */ @@ -284,18 +303,36 @@ export interface FactoryPublicReadinessReconcileHealth { /** `inFlightMs` expressed in sweeps that should have run and did not. */ missedPasses?: number /** - * The last completed sweep's arithmetic, published (#355). + * The last enumerating sweep's arithmetic, published (#355). * * Counts only — no issue keys, no paths, no titles — and absent rather than - * zero until a sweep has completed, so "never ran" and "ran and found - * nothing" are two different readings of this surface rather than one. + * zero until a sweep has completed enumeration, so "never enumerated" and + * "enumerated and found nothing" are two different readings of this surface + * rather than one. A completed deferral still leaves these absent. */ candidates?: number dispatched?: number skipped?: number /** `skipped` split by a closed vocabulary of causes; zero counts omitted. */ skipReasons?: Partial> - /** The last completed sweep deferred to another process's discovery lease. */ + /** + * When the pass the counts describe finished enumerating. Dates them — + * `lastCompletedAtMs` does not, since it advances on deferred passes too. + */ + lastEnumeratedAtMs?: number + /** + * A producer supplied some enumeration counts, but the trio was incomplete + * or invalid and was rejected during normalization. This is not equivalent + * to a genuine first-pass deferral with no enumeration evidence. + */ + enumerationCountsInvalid?: true + /** + * The most recent pass deferred to another process's discovery lease. Present + * alongside the counts it means they are from an earlier pass. Present alone + * means nothing has enumerated yet only when `enumerationCountsInvalid` is + * absent; otherwise a supplied snapshot was unusable and prior enumeration + * is unknown. + */ discoveryDeferred?: 'sweep-in-flight' lastErrorClass?: string }