From 0a1b81e695c77ec98b619db8430c08601903baef Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 24 Aug 2026 02:23:45 +0200 Subject: [PATCH] feat(health): publish the sweep's candidate/dispatched/skipped counts (#355) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A readiness sweep on the live container completed in 843ms with `state: healthy`, `consecutiveFailures: 0`, no `lastError`, `fleetControlPlane: closed`, `eventListener: subscribed` and a free dispatch slot — and dispatched none of seven eligible issues in routed repos. Every published subsystem read green through a total dispatch outage, because none of them measures what the sweep actually did. The sweep's own completion log already carries the three numbers that split this in half: - `candidates > 0` — it SAW those issues and REJECTED them, so the bug is in eligibility evaluation. - `candidates == 0` — it NEVER PULLED them, so the bug is upstream in discovery/ingestion. Two different bugs with two different owners, and the line separating them goes to container stdout, which does not reach `wrangler tail`. This puts it on the evidence surface instead. `readinessReconcile` now carries `candidates`, `dispatched` and `skipped` from the last *completed* sweep — the same tense as `lastDurationMs`, dated by `lastCompletedAtMs`, left untouched by a pass that failed. They travel as a group, all present or none, so nothing can publish one pass's `dispatched` beside another's `candidates`. Absent and zero are deliberately different. A sweep that ran and found nothing publishes `0`; a daemon that has not finished a sweep publishes nothing at all. That distinction is the whole point — the projection uses `optionalCount`, not `counter()`, which would have collapsed both into `0` — and it is what tells "never ran" from "ran and found nothing". Two fields make a zero readable rather than merely visible: - `skipReasons` splits `skipped` by `FACTORY_SWEEP_SKIP_REASON_CODES`, a closed fourteen-value vocabulary recorded at the skip site, never matched out of the operator text — a rename would otherwise collapse every bucket into `other`. `dispatch-terminal` and `dispatch-retry-limit` are the two that never clear on their own. `#dispatchBlockReason` returns its code alongside its message for the same reason. - `discoveryDeferred: 'sweep-in-flight'` names the sweep that returned immediately because another process held the discovery lease. It completes healthy in milliseconds having enumerated nothing, so on counts alone it is indistinguishable from a sweep that queried every repo and found no work — opposite diagnoses, identical `candidates: 0`. Counts only, by construction, per the existing `fleetControlPlane` and #315 occupant rules: no issue keys, paths or titles. `skipReasons` is the only field whose KEYS arrive from a remote record, and an object key publishes as readily as a value, so the reader rebuilds it from its own copy of the vocabulary and folds anything unrecognised into `other` — which keeps the parts summing to `skipped` rather than silently dropping a bucket. `factory diagnose` renders both, and the final "dispatching: readinessReconcile is healthy" verdict — the line that was true and useless tonight — now ends with the last sweep's arithmetic. Tests: a must-fire/must-not-fire pair over a real live daemon and a real sweep (never a hand-set status field), plus a CONTROL asserting each expectation throws on the other's fixture, so a hard-wired counter cannot satisfy both. Also pinned: the periodic reconcile as well as the startup backfill, absent-until-swept, the skip-reason split, the deferred sweep, and redaction of the serialized record. Verified by ablation — removing the periodic recorder, the backfill recorder, the deferred field, or the absent/zero distinction each fails exactly one of them, and hard-wiring the counters fails the control specifically. Co-Authored-By: Claude Opus 5 --- docs/deployed-diagnostics.md | 39 ++- src/cli/diagnose.ts | 44 ++- src/index.ts | 6 + src/orchestrator/factory.test.ts | 26 +- src/orchestrator/factory.ts | 133 +++++++- src/orchestrator/public-health.test.ts | 137 +++++++++ src/orchestrator/public-health.ts | 89 ++++++ src/orchestrator/sweep-counters.test.ts | 384 ++++++++++++++++++++++++ src/orchestrator/sweep-skip-reason.ts | 89 ++++++ src/types.ts | 53 +++- 10 files changed, 975 insertions(+), 25 deletions(-) create mode 100644 src/orchestrator/sweep-counters.test.ts create mode 100644 src/orchestrator/sweep-skip-reason.ts diff --git a/docs/deployed-diagnostics.md b/docs/deployed-diagnostics.md index 2ef2fa05..be1aa7c4 100644 --- a/docs/deployed-diagnostics.md +++ b/docs/deployed-diagnostics.md @@ -68,7 +68,12 @@ logic of its own by design: the boundary lives in one place, in this repo, with "inFlightSinceMs": 1787224595805, // when the oldest sweep still running began "inFlightMs": 4560000, // this pass has run 76 minutes "missedPasses": 76, - "lastErrorClass": "TimeoutError" + "lastErrorClass": "TimeoutError", + // The last COMPLETED sweep's arithmetic (#355). Absent until one completes. + "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 } }, "eventListener": { "state": "subscribed" }, "fleetControlPlane": { "state": "closed", "consecutiveFailures": 0, "failureThreshold": 3 } @@ -86,6 +91,38 @@ logic of its own by design: the boundary lives in one place, in this repo, with #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. +- **`candidates` / `dispatched` / `skipped`** — the *green-but-idle* case, and the fastest question to + ask when nothing is being dispatched and every state above reads healthy. On 2026-08-23 a sub-second + sweep with `state: healthy`, `consecutiveFailures: 0` and a free dispatch slot declined seven + eligible issues, and no surface anyone could reach said which half of the pipeline was at fault. + + - `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`. + +- **`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. + + 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. + - **`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. diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index a9e9d4d6..2a1ed36a 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -1,6 +1,6 @@ import { telemetryErrorClass } from '../observability/error-class.js' import { normalizePublicHealth } from '../orchestrator/public-health.js' -import type { FactoryPublicHealth } from '../types' +import type { FactoryPublicHealth, FactoryPublicReadinessReconcileHealth } from '../types' /** * `factory diagnose --deployed ` (#295). @@ -315,15 +315,49 @@ function verdictFor(diagnosis: Omit): string { + const entries = Object.entries(counts).filter(([, count]) => count > 0) + return entries.length === 0 ? '—' : entries.map(([code, count]) => `${code}=${count}`).join(', ') +} + export function formatDuration(ms: number | undefined): string { if (ms === undefined) return 'unknown' if (ms < 1_000) return `${ms}ms` @@ -466,6 +500,12 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st lines.push(` lastStartedAt : ${formatInstant(readiness.lastStartedAtMs)}`) lines.push(` lastCompletedAt : ${formatInstant(readiness.lastCompletedAtMs)}`) lines.push(` lastFailureAt : ${formatInstant(readiness.lastFailureAtMs)}`) + // #355. `candidates === 0` and an absent `candidates` are opposite + // diagnoses, so the renderer must not collapse them into one dash. + lines.push(` last sweep : ${formatSweepOutcome(readiness)}`) + if (readiness.skipReasons) { + lines.push(` skip reasons : ${formatSkipReasons(readiness.skipReasons)}`) + } } const capacity = health.dispatchCapacity if (capacity) { diff --git a/src/index.ts b/src/index.ts index 2f7e931d..68816348 100644 --- a/src/index.ts +++ b/src/index.ts @@ -490,6 +490,12 @@ export type { TriageDecision, TriageEngine, } from './types' +export { + FACTORY_SWEEP_SKIP_REASON_CODES, + factorySweepSkipReasonCode, + factorySweepSkipReasonCounts, +} from './orchestrator/sweep-skip-reason' +export type { FactorySweepSkipReasonCode } from './orchestrator/sweep-skip-reason' export type { FleetControlPlaneState, FleetControlPlaneStatus } from './fleet/control-plane-circuit' export { LOAD_EVIDENCE_CONTRACT, diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 3d165078..bdacc1a7 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -3209,6 +3209,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-141', key: 'AR-141', path: dependentPath }, reason: 'parked on dependencies: AgentWorkforce/pear#140', + code: 'parked-dependency', }) expect(factory.status().parked).toEqual([ expect.objectContaining({ @@ -3355,6 +3356,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: expect.objectContaining({ key: '36' }), reason: 'parked on dependencies: AgentWorkforce/pear#35', + code: 'parked-dependency', }) expect(factory.status().parked).toEqual([ expect.objectContaining({ issue: expect.objectContaining({ key: '36' }) }), @@ -3512,6 +3514,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toEqual([{ issue: { uuid: 'AgentWorkforce/pear#46', key: '46', path: canonicalPath }, reason: 'live state is not ready-for-agent', + code: 'not-ready', }]) expect(fleet.spawns).toEqual([]) }) @@ -3809,6 +3812,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toEqual([{ issue: { uuid: 'AgentWorkforce/pear#49', key: '49', path }, reason: 'live state is not ready-for-agent', + code: 'not-ready', }]) expect(fleet.spawns).toEqual([]) }) @@ -5480,6 +5484,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: racedPath }, reason: 'live state changed during dispatch', + code: 'dispatch-failed', }) expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ @@ -5572,6 +5577,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: blockedPath }, reason: 'dispatch lifecycle already terminal', + code: 'dispatch-failed', }) expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ @@ -5605,6 +5611,7 @@ describe('FactoryLoop', () => { // Sanitized: `run-once` prints the report as JSON, so the reason // carries a classification rather than raw provider text. reason: 'dispatch failed (TypeError)', + code: 'dispatch-failed', }) expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) expect(factory.status().counters.dispatchItemFailuresSkipped).toBe(1) @@ -5727,6 +5734,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'AgentWorkforce/pear#59', key: '59', path: blockedPath }, reason: 'dispatch failed (Error)', + code: 'dispatch-failed', }) expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) expect(fleet.spawns).toEqual([]) @@ -5918,6 +5926,10 @@ describe('FactoryLoop', () => { expect(report.skipped).toEqual([{ issue: { uuid: 'AgentWorkforce/pear#53', key: '53', path }, reason: 'active dispatch claim or live agent still owns the issue', + // The batch already holds this issue, so it is the `already-tracked` + // gate that declines it — not the readiness gate that carries the + // same operator text one branch later. + code: 'already-tracked', }]) expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-53-babysit-pear']) expect(fleet.spawns[0]).toMatchObject({ @@ -6234,6 +6246,7 @@ describe('FactoryLoop', () => { issue: '55', path, reason: 'active dispatch claim or live agent still owns the issue', + code: 'not-ready', }, ]) } finally { @@ -6705,8 +6718,8 @@ describe('FactoryLoop', () => { expect(report.pulled.map((issue) => issue.key)).toEqual(['AR-1', 'AR-2', 'AR-3', 'AR-4']) expect(report.dispatched.map((result) => result.issue.key)).toEqual(['AR-1', 'AR-2']) - expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-3', key: 'AR-3', path: issuePath(3) }, reason: 'queued or escalated' }) - expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-4', key: 'AR-4', path: issuePath(4) }, reason: 'live state is not ready-for-agent' }) + expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-3', key: 'AR-3', path: issuePath(3) }, reason: 'queued or escalated', code: 'queued-or-escalated' }) + expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-4', key: 'AR-4', path: issuePath(4) }, reason: 'live state is not ready-for-agent', code: 'not-ready' }) expect(fleet.spawns).toHaveLength(4) expect(factory.status().inFlight.map((issue) => issue.key)).toEqual(['AR-1', 'AR-2']) expect(factory.status().queued.map((issue) => issue.key)).toEqual(['AR-3']) @@ -7208,7 +7221,7 @@ describe('FactoryLoop', () => { expect(report.pulled).toEqual([{ uuid: 'uuid-67-canonical', key: 'AR-67', path: canonicalPath }]) expect(report.dispatched).toEqual([]) expect(report.skipped).toEqual([ - { issue: { uuid: 'uuid-67-canonical', key: 'AR-67', path: canonicalPath }, reason: 'live state is not ready-for-agent' }, + { issue: { uuid: 'uuid-67-canonical', key: 'AR-67', path: canonicalPath }, reason: 'live state is not ready-for-agent', code: 'not-ready' }, ]) expect(fleet.spawns).toEqual([]) }) @@ -7287,7 +7300,7 @@ describe('FactoryLoop', () => { expect(report.dispatched).toEqual([]) expect(report.skipped).toEqual([ - { issue: { uuid: 'uuid-365', key: 'AR-365', path: issuePath(365) }, reason: 'dispatch already terminal' }, + { issue: { uuid: 'uuid-365', key: 'AR-365', path: issuePath(365) }, reason: 'dispatch already terminal', code: 'dispatch-terminal' }, ]) expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-365-impl-pear', 'ar-365-review']) expect(factory.status().counters.dispatchTerminalReopened).toBeUndefined() @@ -7758,6 +7771,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: '40c7e780-59ad-47ee-8809-3a9b8434d8fb', key: 'AR-173', path: capturedStaleDoneCanonicalPath }, reason: 'live state is not ready-for-agent', + code: 'not-ready', }) expect(mount.readPaths).not.toContain(byIdBareAliasPath) expect(mount.readPaths).not.toContain(byIdCanonicalShapedAliasPath) @@ -11010,6 +11024,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-21', key: 'AR-21', path: unscopedPath }, reason: 'not factory-e2e scope', + code: 'out-of-scope', }) expect(report.triaged).toEqual([]) expect(report.dispatched).toEqual([]) @@ -11095,6 +11110,7 @@ describe('FactoryLoop', () => { expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-261', key: 'AR-261', path: draftPath }, reason: 'not reconciled real Linear issue', + code: 'not-dispatchable', }) expect(report.triaged).toEqual([]) expect(report.dispatched).toEqual([]) @@ -19231,7 +19247,7 @@ describe('FactoryLoop', () => { const report = await factory.runOnce() expect(report.dispatched).toEqual([]) - expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-20', key: 'AR-20', path: issuePath(20) }, reason: 'queued or escalated' }) + expect(report.skipped).toContainEqual({ issue: { uuid: 'uuid-20', key: 'AR-20', path: issuePath(20) }, reason: 'queued or escalated', code: 'queued-or-escalated' }) expect(fleet.spawns).toEqual([]) const slackRoots = mount.writes.filter((write) => isSlackRootWritePath(write.path)) expect(slackRoots).toHaveLength(1) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index efa118fe..191d4fcd 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -150,6 +150,8 @@ import { publicHealthFromHeartbeat, readinessReconcileInFlightMs, } from './public-health' +import { factorySweepSkipReasonCounts } from './sweep-skip-reason' +import type { FactorySweepSkipReasonCode } from './sweep-skip-reason' import { boundedRunCostTotal, CostLedger, type RunCostTotal, type UnpricedModelCostRecord } from '../cost/ledger' import { createTicketDispatchDelivery, type TicketDispatchDelivery } from '../delivery/ticket-dispatch' import { @@ -845,6 +847,25 @@ export class FactoryLoop implements Factory { #readinessReconcileLastFailureAtMs?: number #readinessReconcileLastError?: string #readinessReconcileLastErrorClass?: string + /** + * The last *completed* sweep's arithmetic (#355). + * + * Held as one record rather than four 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 — + * "this daemon has not finished a sweep" and "a sweep finished and found + * nothing" are the two readings #355 has to tell apart. + */ + #readinessReconcileLastSweep?: { + candidates: number + dispatched: number + skipped: number + skipReasons: Partial> + discoveryDeferred?: 'sweep-in-flight' + } readonly #liveEventQueue: ChangeEvent[] = [] #liveEventDrainScheduled = false #liveEventDrainActive = false @@ -1610,7 +1631,11 @@ export class FactoryLoop implements Factory { const backfillStartedAtMs = this.#clock.now() this.#readinessReconcileLastStartedAtMs = backfillStartedAtMs try { - await this.runOnce() + // The startup backfill is a discovery pass like any other, and on a + // 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()) this.#readinessReconcileLastDurationMs = this.#elapsedSince(backfillStartedAtMs) this.#readinessReconcileLastCompletedAtMs = this.#clock.now() } catch (error) { @@ -1853,11 +1878,18 @@ export class FactoryLoop implements Factory { this.#readinessReconcileLastCompletedAtMs = this.#clock.now() 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.#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, + discoveryDeferred: report.discoveryDeferred, }) } catch (error) { // #297: all four relayfile overload reason codes share one message, and @@ -2919,6 +2951,7 @@ export class FactoryLoop implements Factory { issue: entry.issue.key, path: entry.issue.path, reason: entry.reason, + code: entry.code, }) } // Backstop for the skip-by-default catch below: see #292. Reset only by @@ -2956,7 +2989,11 @@ export class FactoryLoop implements Factory { retryAfterSeconds: overload.retryAfterSeconds, sweepOverloads: this.#discoverySweepOverloads, }) - recordSkip({ issue: issueRefFromPath(path), reason: perItemDispatchSkipReason(error) }) + recordSkip({ + issue: issueRefFromPath(path), + reason: perItemDispatchSkipReason(error), + code: 'read-failed', + }) } readyIssueReads += 1 // Relayfile served this work unit's read: the dependency is shedding @@ -3022,7 +3059,7 @@ export class FactoryLoop implements Factory { if (!mayRecoverGithubOrphan) { const dispatchBlock = await this.#dispatchBlockReason(issue) if (dispatchBlock) { - recordSkip({ issue: issueRef(issue), reason: dispatchBlock }) + recordSkip({ issue: issueRef(issue), ...dispatchBlock }) continue } } @@ -3033,20 +3070,25 @@ export class FactoryLoop implements Factory { const recoveredOrphan = orphanResult.recovered const batch = await this.#batch() if (batch.isInFlight(issue) || batch.isQueued(issue)) { - recordSkip({ issue: issueRef(issue), reason: orphanResult.reason ?? 'already tracked' }) + recordSkip({ + issue: issueRef(issue), + reason: orphanResult.reason ?? 'already tracked', + code: 'already-tracked', + }) continue } if (!wasReady && !recoveredOrphan) { if (mayRecoverGithubOrphan) { const dispatchBlock = await this.#dispatchBlockReason(issue) if (dispatchBlock) { - recordSkip({ issue: issueRef(issue), reason: dispatchBlock }) + recordSkip({ issue: issueRef(issue), ...dispatchBlock }) continue } } recordSkip({ issue: issueRef(issue), reason: orphanResult.reason ?? 'live state is not ready-for-agent', + code: 'not-ready', }) continue } @@ -3055,18 +3097,22 @@ export class FactoryLoop implements Factory { if (recoveredOrphan) { const dispatchBlock = await this.#dispatchBlockReason(issue) if (dispatchBlock) { - recordSkip({ issue: issueRef(issue), reason: dispatchBlock }) + recordSkip({ issue: issueRef(issue), ...dispatchBlock }) continue } } if (!isInFactoryScope(issue, this.#config.safety)) { - recordSkip({ issue: issueRef(issue), reason: 'not factory-e2e scope' }) + recordSkip({ issue: issueRef(issue), reason: 'not factory-e2e scope', code: 'out-of-scope' }) continue } if (!isDispatchableIssue(issue)) { - recordSkip({ issue: issueRef(issue), reason: 'not reconciled real Linear issue' }) + recordSkip({ + issue: issueRef(issue), + reason: 'not reconciled real Linear issue', + code: 'not-dispatchable', + }) continue } @@ -3080,12 +3126,17 @@ export class FactoryLoop implements Factory { // decays the durable overload ratchet (#297). this.#discoverySweepProgress = true if (result.agents.length === 0 && !dryRun) { + const code: FactorySweepSkipReasonCode = result.hold?.kind === 'dependency-cycle' + ? 'dependency-cycle' + : result.hold?.kind === 'dependency' + ? 'parked-dependency' + : 'queued-or-escalated' const reason = result.hold?.kind === 'dependency-cycle' ? `dependency cycle detected: ${result.hold.cycle?.join(' -> ') ?? 'unknown cycle'}` : result.hold?.kind === 'dependency' ? `parked on dependencies: ${result.hold.blockers?.join(', ') ?? 'unresolved dependency'}` : 'queued or escalated' - recordSkip({ issue: decision.issue, reason }) + recordSkip({ issue: decision.issue, reason, code }) } else { dispatched.push(result) } @@ -3162,7 +3213,11 @@ export class FactoryLoop implements Factory { error: describeError(error).errorMessage, }) } - recordSkip({ issue: issueRef(issue), reason: perItemDispatchSkipReason(error) }) + recordSkip({ + issue: issueRef(issue), + reason: perItemDispatchSkipReason(error), + code: 'dispatch-failed', + }) continue } finally { if (recoveredIdentity) this.#reconciledGithubInProgress.delete(recoveredIdentity) @@ -4500,7 +4555,7 @@ export class FactoryLoop implements Factory { const blockReason = await this.#dispatchBlockReason(decision.issue) if (blockReason) { - const error = new Error(`Refusing to dispatch ${decision.issue.key}: ${blockReason}`) + const error = new Error(`Refusing to dispatch ${decision.issue.key}: ${blockReason.reason}`) this.#error(error, decision.issue) throw error } @@ -4973,6 +5028,24 @@ export class FactoryLoop implements Factory { return { state: 'starting' } } + /** + * Snapshot a settled sweep's counts onto the readiness surface (#355). + * + * 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. + */ + #recordReadinessSweepOutcome(report: IterationReport): void { + this.#readinessReconcileLastSweep = { + candidates: report.pulled.length, + dispatched: report.dispatched.length, + skipped: report.skipped.length, + skipReasons: factorySweepSkipReasonCounts(report.skipped), + ...(report.discoveryDeferred ? { discoveryDeferred: report.discoveryDeferred } : {}), + } + } + #readinessReconcileStatus(): FactoryReadinessReconcileStatus { const consecutiveFailures = this.#readinessReconcileConsecutiveFailures // #296 owns the numerator here, #295/#300 own the derivation. The earliest @@ -5049,6 +5122,22 @@ export class FactoryLoop implements Factory { ...(this.#readinessReconcileLastFailureAtMs !== undefined ? { lastFailureAtMs: this.#readinessReconcileLastFailureAtMs } : {}), + // Spread whole or not at all: see `#readinessReconcileLastSweep`. Zeroes + // are published, which is the entire point — an absent `candidates` and + // a zero `candidates` are different diagnoses (#355). + ...(this.#readinessReconcileLastSweep + ? { + candidates: this.#readinessReconcileLastSweep.candidates, + dispatched: this.#readinessReconcileLastSweep.dispatched, + skipped: this.#readinessReconcileLastSweep.skipped, + ...(Object.keys(this.#readinessReconcileLastSweep.skipReasons).length > 0 + ? { skipReasons: { ...this.#readinessReconcileLastSweep.skipReasons } } + : {}), + ...(this.#readinessReconcileLastSweep.discoveryDeferred + ? { discoveryDeferred: this.#readinessReconcileLastSweep.discoveryDeferred } + : {}), + } + : {}), ...(this.#readinessReconcileLastError ? { lastError: this.#readinessReconcileLastError } : {}), ...(this.#readinessReconcileLastErrorClass ? { lastErrorClass: this.#readinessReconcileLastErrorClass } @@ -7966,20 +8055,32 @@ export class FactoryLoop implements Factory { } } - async #dispatchBlockReason(issue: IssueRef): Promise { + /** + * Why durable dispatch state refuses this issue, or `undefined` to proceed. + * + * Returns the operator text *and* its #355 code together rather than letting + * the sweep re-derive one from the other: these four conditions are the ones + * most likely to explain a sweep that saw eligible issues and dispatched + * nothing, and two of them (`terminal`, `retry-limit`) never clear on their + * own. Deriving the code by matching the message would put that distinction + * one rename away from collapsing into `other`. + */ + async #dispatchBlockReason( + issue: IssueRef, + ): Promise<{ reason: string; code: FactorySweepSkipReasonCode } | undefined> { const key = issueStateKey(issue) const state = await this.#state.getDispatchAttempts(this.#workspaceId, key) if (!state) return undefined - if (state.terminal) return 'dispatch already terminal' - if (state.inFlight) return 'dispatch already in-flight' + if (state.terminal) return { reason: 'dispatch already terminal', code: 'dispatch-terminal' } + if (state.inFlight) return { reason: 'dispatch already in-flight', code: 'dispatch-in-flight' } const now = this.#clock.now() if (state.backoffUntilMs > now) { - return 'dispatch backoff active' + return { reason: 'dispatch backoff active', code: 'dispatch-backoff' } } if (state.attempts >= this.#config.dispatch.maxAttempts) { state.terminal = true await this.#state.recordDispatchAttempt(this.#workspaceId, key, state) - return 'dispatch retry limit reached' + return { reason: 'dispatch retry limit reached', code: 'dispatch-retry-limit' } } return undefined } diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 40d9f43f..26c4221e 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -743,3 +743,140 @@ describe('publicHealthFromHeartbeat (#295)', () => { }) }) + +describe('sweep counters on the public surface (#355)', () => { + const swept = ( + overrides: Partial> = {}, + ) => publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: BOOT_MS - 30_000, + lastCompletedAtMs: BOOT_MS - 29_000, + lastDurationMs: 1_000, + ...overrides, + }, + }), + { nowMs: BOOT_MS + 1_000 }, + ).readinessReconcile + + it('publishes a completed sweep that found nothing as zero, and one that never ran as absent', () => { + const ran = swept({ candidates: 0, dispatched: 0, skipped: 0 }) + expect(ran?.candidates).toBe(0) + expect(Object.hasOwn(ran ?? {}, 'candidates')).toBe(true) + + const neverRan = swept() + expect(Object.hasOwn(neverRan ?? {}, 'candidates')).toBe(false) + expect(Object.hasOwn(neverRan ?? {}, 'dispatched')).toBe(false) + expect(Object.hasOwn(neverRan ?? {}, 'skipped')).toBe(false) + }) + + // A record carrying one of the three and not the others is a producer this + // version does not understand. Publishing the fragment would invite + // "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(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({ + candidates: 4, + dispatched: 1, + skipped: 3, + }) + }) + + 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' }) + // Only the one value the vocabulary has. + expect(swept({ + candidates: 0, + dispatched: 0, + skipped: 0, + discoveryDeferred: 'whatever the producer felt like' as 'sweep-in-flight', + })?.discoveryDeferred).toBeUndefined() + }) + + // MUST-NOT-FIRE. `skipReasons` is the only field here whose *keys* come from + // a remote record, and an object key is as publishable as a value: a + // producer on another version could otherwise put an issue key or a + // filesystem path onto the unauthenticated surface by using it as one. + it('rebuilds the skip breakdown from its own vocabulary, so no remote key can cross', () => { + const readiness = swept({ + candidates: 9, + dispatched: 0, + skipped: 9, + skipReasons: { + 'out-of-scope': 4, + // Not in the vocabulary, and carrying exactly what must never publish. + ["AR-350 /linear/issues/AR-350__uuid.json"]: 3, + ['dispatch-terminal']: 2, + } as Record, + }) + + expect(JSON.stringify(readiness)).not.toContain('AR-350') + expect(JSON.stringify(readiness)).not.toContain('/linear/issues') + // Folded into `other`, not dropped: the parts still sum to `skipped`, so a + // reader comparing them does not conclude the counter is broken. + expect(readiness?.skipReasons).toEqual({ 'out-of-scope': 4, 'dispatch-terminal': 2, other: 3 }) + expect(Object.values(readiness?.skipReasons ?? {}).reduce((sum, n) => sum + n, 0)) + .toBe(readiness?.skipped) + }) + + it('drops counts a reader cannot use, and the breakdown entirely when it is empty', () => { + expect(swept({ + candidates: 1, + dispatched: 0, + skipped: 1, + skipReasons: { + 'out-of-scope': Number.NaN, + 'dispatch-backoff': -3, + 'not-ready': 0, + } as Record, + })?.skipReasons).toBeUndefined() + expect(swept({ + candidates: 1, + dispatched: 0, + skipped: 1, + skipReasons: { 'not-ready': 1.9 } as Record, + })?.skipReasons).toEqual({ 'not-ready': 1 }) + }) + + it('re-reads its own published record without turning a zero back into an absence', () => { + const published = swept({ candidates: 0, dispatched: 0, skipped: 0 }) + const reread = normalizePublicHealth({ + schemaVersion: FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + ok: true, + status: 'ok', + stale: false, + degradedSubsystems: [], + readinessReconcile: published, + }) + expect(reread?.readinessReconcile).toMatchObject({ candidates: 0, dispatched: 0, skipped: 0 }) + expect(Object.hasOwn(reread?.readinessReconcile ?? {}, 'candidates')).toBe(true) + }) + + it('applies the same key rebuild to a record that arrived over the wire', () => { + const reread = normalizePublicHealth({ + schemaVersion: FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + ok: true, + status: 'ok', + stale: false, + degradedSubsystems: [], + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + candidates: 7, + dispatched: 0, + skipped: 7, + skipReasons: { '/srv/agent-workforce/.relay/workspace-key': 7 }, + }, + }) + expect(JSON.stringify(reread)).not.toContain('workspace-key') + expect(reread?.readinessReconcile?.skipReasons).toEqual({ other: 7 }) + }) +}) diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index ca13b01d..7942db3d 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -2,6 +2,11 @@ import { createHash, randomBytes } from 'node:crypto' 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 { + FACTORY_SWEEP_SKIP_REASON_CODES, + factorySweepSkipReasonCode, +} from './sweep-skip-reason' +import type { FactorySweepSkipReasonCode } from './sweep-skip-reason' import type { FactoryDispatchCapacityStatus, FactoryEventListenerStatus, @@ -164,6 +169,88 @@ const boundedText = (value: string): string => // C1 range as escape introducers. value.replace(/[\u0000-\u001F\u007F-\u009F]+/gu, ' ').trim().slice(0, 300) +/** + * The last sweep's skip breakdown, rebuilt key by key (#355). + * + * Numbers only, and the keys come from this module's own copy of the + * vocabulary rather than from the record: a producer on another version — or a + * corrupted one — could otherwise put an arbitrary string on an + * unauthenticated surface simply by using it as an object key, which is the + * one thing every other field here is careful not to allow. An unknown key's + * count is folded into `other` rather than dropped, so the parts still sum to + * `skipped`. + */ +const skipReasonCounts = ( + value: unknown, +): Partial> | undefined => { + const record = plainRecord(value) + if (!record) return undefined + const counts: Partial> = {} + for (const [key, raw] of Object.entries(record)) { + const parsed = finiteNumber(raw) + if (parsed === undefined || parsed < 0) continue + const floored = Math.floor(parsed) + if (floored === 0) continue + const code = factorySweepSkipReasonCode(key) + counts[code] = (counts[code] ?? 0) + floored + } + // Emitted in vocabulary order so two samples of the same surface diff + // cleanly, and dropped entirely when empty: `skipped` already carries the + // total, so an empty breakdown states nothing the reader did not have. + const ordered = FACTORY_SWEEP_SKIP_REASON_CODES.filter((code) => counts[code] !== undefined) + if (ordered.length === 0) return undefined + return Object.fromEntries(ordered.map((code) => [code, counts[code] as number])) +} + +/** + * The last completed 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 + * 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. + */ +const sweepOutcome = ( + // Deliberately `unknown` per field rather than the status type: the same + // code serves the writer, which holds a real status, and the reader, which + // holds parsed JSON from a process it does not control. Casting the latter + // into the former to share the function would be the one unchecked + // assumption on a path whose whole job is not making any. + status: { + candidates?: unknown + dispatched?: unknown + skipped?: unknown + skipReasons?: unknown + discoveryDeferred?: unknown + }, +): Partial> => { + const candidates = optionalCount('candidates', status.candidates) + const dispatched = optionalCount('dispatched', status.dispatched) + const skipped = optionalCount('skipped', status.skipped) + // 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 {} + } + const skipReasons = skipReasonCounts(status.skipReasons) + return { + ...candidates, + ...dispatched, + ...skipped, + ...(skipReasons ? { skipReasons } : {}), + ...(status.discoveryDeferred === 'sweep-in-flight' + ? { discoveryDeferred: 'sweep-in-flight' as const } + : {}), + } +} + const DISPATCH_CAPACITY_STATES: readonly FactoryPublicDispatchCapacityHealth['state'][] = [ 'healthy', 'waiting', @@ -299,6 +386,7 @@ function readinessReconcileHealth( ...(inFlightMs !== undefined ? { inFlightMs, missedPasses: Math.floor(inFlightMs / cadenceMs) } : {}), + ...sweepOutcome(status), // `lastError` itself never crosses. Its class does, through the same // allowlist that guards IterationReport.skipped[].reason — and a record // that carries an error but no admissible class still says so. @@ -641,6 +729,7 @@ export function normalizePublicHealth(value: unknown): FactoryPublicHealth | und ...optionalTimestamp('lastFailureAtMs', readiness.lastFailureAtMs), ...optionalDuration('inFlightMs', readiness.inFlightMs), ...optionalCount('missedPasses', readiness.missedPasses), + ...sweepOutcome(readiness), ...(readiness.lastErrorClass !== undefined ? { lastErrorClass: telemetryErrorClassName(readiness.lastErrorClass) } : {}), diff --git a/src/orchestrator/sweep-counters.test.ts b/src/orchestrator/sweep-counters.test.ts new file mode 100644 index 00000000..f8816409 --- /dev/null +++ b/src/orchestrator/sweep-counters.test.ts @@ -0,0 +1,384 @@ +import { describe, expect, it, vi } from 'vitest' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { + FactoryConfigSchema, + createFactory, + publicHealthFromHeartbeat, + type FactoryConfig, + type LinearIssue, + type TriageDecision, + type TriageEngine, +} from '../index' +import { FakeFleetClient, FakeMountClient } from '../testing' +import { InMemoryStateStore } from '../state/in-memory-state-store' +import type { DiscoverySweepClaim } from '../ports/state' +import { normalizePublicHealth } from './public-health' +import type { FactoryPublicReadinessReconcileHealth, FactoryReadinessReconcileStatus } from '../types' + +/** + * The sweep counters, end to end (#355). + * + * A healthy sub-second sweep declined seven eligible issues with a free + * dispatch slot and every published subsystem read green. `candidates` splits + * that in half: non-zero means the sweep saw them and rejected them, zero + * means it never pulled them, and those are two different bugs with two + * different owners. + * + * Every assertion below drives the real writer — a live daemon running a real + * sweep over a real mount — rather than hand-setting the field on a status + * fixture. A fixture would prove the projection copies a number it was handed; + * only the sweep proves the number is the sweep's. + */ + +const ready = '11111111-1111-4111-8111-111111111111' +const implementing = '22222222-2222-4222-8222-222222222222' +const done = '33333333-3333-4333-8333-333333333333' +const planning = '44444444-4444-4444-8444-444444444444' + +const config = (overrides: Record = {}): FactoryConfig => FactoryConfigSchema.parse({ + workspaceId: 'factory-sweep-counters', + repos: { + byLabel: { pear: 'AgentWorkforce/pear' }, + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + triage: { maxImplementers: 4 }, + batchSize: 4, + stateIds: { readyForAgent: ready, agentImplementing: implementing, done, inPlanning: planning }, + verification: { enabled: false }, + ...overrides, +}) + +const issuePath = (n: number) => `/linear/issues/AR-${n}__uuid-${n}.json` + +const issueFile = (n: number, title = `[factory-e2e] Fix factory issue ${n}`) => ({ + provider: 'linear', + objectType: 'issue', + objectId: `uuid-${n}`, + payload: { + id: `uuid-${n}`, + identifier: `AR-${n}`, + title, + description: 'Implement the requested fix in src/orchestrator/factory.ts and verify it with tests.', + stateId: ready, + url: `https://linear.app/agent-relay/issue/AR-${n}/factory-issue-${n}`, + labels: [{ name: 'pear' }], + labelIds: ['label-id-not-used-by-parser'], + team: { key: 'AR', name: 'Agent Relay' }, + project: { name: 'Factory' }, + state: { id: ready, name: 'Ready for Agent' }, + }, +}) + +class StaticTriage implements TriageEngine { + async triage(issue: LinearIssue): Promise { + const number = issue.key.match(/\d+/)?.[0] ?? '0' + return { + issue: { uuid: issue.uuid, key: issue.key, path: issue.path }, + routes: [{ repo: 'AgentWorkforce/pear', clonePath: '/work/pear', rationale: 'test route' }], + scope: 'single', + implementers: [{ + name: `ar-${number}-impl`, + role: 'implementer', + capability: 'spawn:codex', + model: 'codex', + task: `Implement ${issue.key}`, + repo: 'AgentWorkforce/pear', + clonePath: '/work/pear', + node: 'self', + }], + reviewer: { + name: `ar-${number}-review`, + role: 'reviewer', + capability: 'spawn:claude', + model: 'claude', + task: `Review ${issue.key}`, + repo: 'AgentWorkforce/pear', + clonePath: '/work/pear', + node: 'self', + }, + thin: false, + confidence: 'high', + rationale: 'static test decision', + } + } +} + +/** + * Run one live daemon over `files` and return the readiness record its own + * startup sweep produced. + * + * The startup backfill is a full discovery pass and settles before `start()` + * resolves, so this needs no timer advance and no polling — which is what + * keeps these assertions off the nondeterminism in #342. + */ +async function sweepReadiness( + files: Record, + overrides: Record = {}, +): Promise<{ status: FactoryReadinessReconcileStatus; spawns: string[] }> { + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const mount = new FakeMountClient(files) + const fleet = new FakeFleetClient() + const factory = createFactory( + config({ + loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') }, + ...overrides, + }), + { mount, fleet, triage: new StaticTriage(), logger: process.env.SWEEP_DEBUG ? console : {} }, + ) + try { + await factory.start({ + mode: 'live', + // Long enough that no periodic pass can race the assertion: the startup + // backfill is the sweep under test, and a second one landing mid-read + // would make the numbers describe a pass the test never set up. + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 600_000 }, + }) + const status = factory.status().readinessReconcile + if (!status) throw new Error('live daemon published no readinessReconcile status') + return { status, spawns: fleet.spawns.map((spawn) => spawn.name) } + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } +} + +/** The public record a container would serve for this readiness status. */ +const published = (status: FactoryReadinessReconcileStatus): FactoryPublicReadinessReconcileHealth => { + const health = publicHealthFromHeartbeat({ + pid: 1, + status: 'running', + iteration: 1, + maxIterations: 1, + updatedAt: new Date(1_700_000_000_000).toISOString(), + updatedAtMs: 1_700_000_000_000, + readinessReconcile: status, + }, { nowMs: 1_700_000_000_000 }) + if (!health.readinessReconcile) throw new Error('public health carried no readinessReconcile block') + return health.readinessReconcile +} + +/** The must-fire expectation, as a function so the control can aim it elsewhere. */ +const expectSawAndDispatched = (readiness: FactoryPublicReadinessReconcileHealth): void => { + expect(readiness.candidates).toBeGreaterThan(0) + expect(readiness.dispatched).toBeGreaterThan(0) +} + +/** The must-not-fire expectation: a real zero, never an absence. */ +const expectSweptNothing = (readiness: FactoryPublicReadinessReconcileHealth): void => { + expect(readiness.candidates).toBe(0) + expect(readiness.dispatched).toBe(0) + expect(readiness.skipped).toBe(0) +} + +describe('readiness sweep counters (#355)', () => { + it('publishes a non-zero candidate count for a sweep that found and dispatched ready work', async () => { + const { status, spawns } = await sweepReadiness({ + [issuePath(901)]: issueFile(901), + [issuePath(902)]: issueFile(902), + }) + + 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 }) + }) + + it('publishes zero — not undefined — for a sweep that completed and found nothing', async () => { + const { status, spawns } = await sweepReadiness({}) + + expect(spawns).toEqual([]) + expect(status).toMatchObject({ candidates: 0, dispatched: 0, skipped: 0 }) + const readiness = published(status) + expectSweptNothing(readiness) + // `toMatchObject` is satisfied by an absent key holding `undefined`, and + // absent-vs-zero is the entire distinction this field exists to carry. + expect(Object.hasOwn(readiness, 'candidates')).toBe(true) + expect(Object.hasOwn(readiness, 'dispatched')).toBe(true) + expect(Object.hasOwn(readiness, 'skipped')).toBe(true) + // And it survives the wire: a reader re-normalising a served record must + // not turn the published zero back into an absence. + const roundTripped = normalizePublicHealth(JSON.parse(JSON.stringify({ + schemaVersion: 1, + ok: true, + status: 'ok', + stale: false, + degradedSubsystems: [], + readinessReconcile: readiness, + }))) + expect(roundTripped?.readinessReconcile).toMatchObject({ candidates: 0, dispatched: 0, skipped: 0 }) + expect(Object.hasOwn(roundTripped?.readinessReconcile ?? {}, 'candidates')).toBe(true) + }) + + it('CONTROL: the two expectations are not interchangeable, so a swap would fail the suite', async () => { + // The pair above is only evidence if each assertion can tell the two + // sweeps apart. If `candidates` were hard-wired — to a constant, to + // `pulled.length` of the wrong pass, or to zero — one of these four would + // stop throwing, and the must-fire/must-not-fire pair would pass for a + // reason that has nothing to do with what the sweep saw. + const found = published((await sweepReadiness({ + [issuePath(903)]: issueFile(903), + })).status) + const empty = published((await sweepReadiness({})).status) + + expect(() => expectSweptNothing(found)).toThrow() + expect(() => expectSawAndDispatched(empty)).toThrow() + expect(() => expectSawAndDispatched(found)).not.toThrow() + expect(() => expectSweptNothing(empty)).not.toThrow() + }) + + it('leaves the counters absent until a sweep completes, so "never ran" stays readable', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const factory = createFactory( + config({ loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') } }), + { mount: new FakeMountClient(), fleet: new FakeFleetClient(), triage: new StaticTriage(), logger: {} }, + ) + try { + const status = factory.status().readinessReconcile + expect(status?.state).toBe('not-running') + expect(status && Object.hasOwn(status, 'candidates')).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. + expect(status && Object.hasOwn(published(status), 'candidates')).toBe(false) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('splits `skipped` by a bounded reason code when the sweep saw work and rejected it', async () => { + // The exact shape #355 suspects: the sweep pulls every ready issue and the + // eligibility gate declines some of them. `candidates` alone says "it saw + // them"; the code says which gate, which is what names the owner. + const { status, spawns } = await sweepReadiness({ + [issuePath(911)]: issueFile(911), + [issuePath(912)]: issueFile(912, 'Ordinary product issue, not in factory scope'), + }) + + expect(spawns).toEqual(['ar-911-impl-pear', 'ar-911-review']) + expect(status).toMatchObject({ + candidates: 2, + dispatched: 1, + skipped: 1, + skipReasons: { 'out-of-scope': 1 }, + }) + expect(published(status).skipReasons).toEqual({ 'out-of-scope': 1 }) + }) + + it('records the PERIODIC sweep, not just the startup backfill', async () => { + // Everything above drives the startup backfill, which is one of two call + // sites. The deployed outage is the *periodic* loop: on the live container + // it had been the only sweep running for hours. A recorder wired to the + // backfill alone would satisfy every other test in this file and publish + // nothing for the pass that matters. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + const mount = new FakeMountClient() + const fleet = new FakeFleetClient() + const factory = createFactory( + config({ loop: { registryPath: join(root, 'registry.json'), heartbeatPath: join(root, 'heartbeat.json') } }), + { mount, fleet, triage: new StaticTriage(), logger: {} }, + ) + try { + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + // The backfill swept an empty mount, so anything non-zero from here can + // only have come from a later pass. + expect(factory.status().readinessReconcile).toMatchObject({ candidates: 0, dispatched: 0 }) + + mount.files.set(issuePath(931), { content: issueFile(931) }) + + await vi.waitFor(() => { + expect(factory.status().counters.readinessReconcileSweeps).toBeGreaterThanOrEqual(1) + expect(factory.status().readinessReconcile).toMatchObject({ + candidates: 1, + dispatched: 1, + skipped: 0, + }) + }, { timeout: 5_000 }) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('names a sweep that deferred to another owner, so its zero is not read as an empty provider', async () => { + // The nastiest reading of `candidates: 0`. A sweep that never claimed the + // discovery lease returns an empty report immediately and completes + // *healthy* in milliseconds — indistinguishable, on counts alone, from one + // that queried every routed repo and legitimately found no ready work. + class LeaseHeldElsewhereStateStore extends InMemoryStateStore { + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const claim = await super.claimDiscoverySweep(workspaceId, 'another-process', nowMs, leaseMs) + return { ...claim, acquired: false, lease: undefined } + } + } + + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-counters-')) + 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 }), + triage: new StaticTriage(), + logger: {}, + }, + ) + try { + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 600_000 }, + }) + const status = factory.status().readinessReconcile + expect(status).toMatchObject({ + state: 'healthy', + candidates: 0, + dispatched: 0, + skipped: 0, + discoveryDeferred: 'sweep-in-flight', + }) + expect(published(status!)).toMatchObject({ + candidates: 0, + discoveryDeferred: 'sweep-in-flight', + }) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('publishes counts only: no issue key, path or title reaches the unauthenticated record', async () => { + const { status } = await sweepReadiness({ + [issuePath(921)]: issueFile(921), + [issuePath(922)]: issueFile(922, 'Ordinary product issue, not in factory scope'), + }) + + const serialized = JSON.stringify(published(status)) + expect(serialized).not.toContain('AR-921') + expect(serialized).not.toContain('AR-922') + expect(serialized).not.toContain('/linear/issues') + expect(serialized).not.toContain('Ordinary product issue') + expect(serialized).not.toContain('factory-e2e') + expect(serialized).not.toContain('AgentWorkforce/pear') + // Everything it does say is a number or a published code. + for (const [key, value] of Object.entries(JSON.parse(serialized) as Record)) { + if (key === 'state') continue + if (key === 'skipReasons') { + for (const count of Object.values(value as Record)) { + expect(typeof count).toBe('number') + } + continue + } + expect(typeof value).not.toBe('object') + } + }) +}) diff --git a/src/orchestrator/sweep-skip-reason.ts b/src/orchestrator/sweep-skip-reason.ts new file mode 100644 index 00000000..72a57280 --- /dev/null +++ b/src/orchestrator/sweep-skip-reason.ts @@ -0,0 +1,89 @@ +/** + * Why a discovery sweep declined to dispatch a work unit, as a closed set (#355). + * + * `IterationReport.skipped[].reason` is free text assembled at the skip site: + * it names issue keys, dependency blockers and provider conditions, so it + * cannot cross onto the unauthenticated health surface. But the *count* of + * skips, split by cause, is the measurement that separates the two bugs #355 + * is stuck between — a sweep that saw seven eligible issues and rejected them + * has a cause, and the cause is what says who owns the fix. + * + * So the code is recorded at the skip site alongside the text, never derived + * from it: matching on a message is a rename away from silently collapsing + * every bucket into `other`, and this vocabulary is the thing an operator + * reads when nothing else is available. + * + * Codes are the *structural* reason the sweep stopped, deliberately finer + * than the four `state`/`reason` buckets an operator can already see: + * `dispatch-terminal` and `dispatch-retry-limit` mean the work unit is + * permanently declined and needs a human, while `dispatch-backoff` and + * `already-tracked` clear on their own. + */ +export const FACTORY_SWEEP_SKIP_REASON_CODES = [ + /** Relayfile shed this work unit's read; the sweep continued without it. */ + 'read-failed', + /** Durable dispatch state says terminal — this issue will never be retried. */ + 'dispatch-terminal', + /** Durable dispatch state says a dispatch is already running for it. */ + 'dispatch-in-flight', + /** Durable dispatch state is in error cooldown. */ + 'dispatch-backoff', + /** Durable dispatch state exhausted `dispatch.maxAttempts`. */ + 'dispatch-retry-limit', + /** The in-memory batch already holds it in flight or queued. */ + 'already-tracked', + /** Live provider state is not ready-for-agent, and no orphan was recovered. */ + 'not-ready', + /** Outside the configured factory scope (label / title prefix / repo route). */ + 'out-of-scope', + /** In scope but not a dispatchable reconciled issue. */ + 'not-dispatchable', + /** Dispatched into a dependency park. */ + 'parked-dependency', + /** Dispatched into a park because its dependencies form a cycle. */ + 'dependency-cycle', + /** Dispatch returned without agents: queued on capacity, or escalated to a human. */ + 'queued-or-escalated', + /** Dispatch threw; the sweep kept the rest of the pass (#292). */ + 'dispatch-failed', + /** Recorded by a producer this vocabulary does not know. */ + 'other', +] as const + +export type FactorySweepSkipReasonCode = typeof FACTORY_SWEEP_SKIP_REASON_CODES[number] + +/** + * Coerce an arbitrary value onto the vocabulary. + * + * Unrecognised codes collapse to `other` rather than being dropped: a skip + * that vanished from the breakdown would make the parts stop summing to + * `skipped`, and a reader comparing the two would conclude the counter was + * broken rather than that the producer was newer. + */ +export const factorySweepSkipReasonCode = (value: unknown): FactorySweepSkipReasonCode => + typeof value === 'string' && (FACTORY_SWEEP_SKIP_REASON_CODES as readonly string[]).includes(value) + ? value as FactorySweepSkipReasonCode + : 'other' + +/** + * The per-cause breakdown, as counts only. + * + * Zero-count codes are omitted — the vocabulary is fixed and published, so an + * absent key reads as zero unambiguously, and emitting fourteen zeroes on + * every heartbeat would bury the one or two that are non-zero. The total is + * carried separately by `skipped`, which is always present, so "ran and + * skipped nothing" stays distinguishable from "never ran". + */ +export function factorySweepSkipReasonCounts( + skipped: Iterable<{ code?: unknown }> | undefined, +): Partial> { + const counts: Partial> = {} + if (!skipped) return counts + for (const entry of skipped) { + const code = factorySweepSkipReasonCode( + entry !== null && typeof entry === 'object' ? entry.code : undefined, + ) + counts[code] = (counts[code] ?? 0) + 1 + } + return counts +} diff --git a/src/types.ts b/src/types.ts index cdb05eb1..f9a1d94f 100644 --- a/src/types.ts +++ b/src/types.ts @@ -8,6 +8,7 @@ import type { AgentWorktreeManager } from './ports/worktree' import type { CloseProbePrInput, CloseProbePrResult } from './github/probe-closer' import type { GhRunner, GithubMergeGate } from './github/merge-gate' import type { AgentProcessFinder, ProcessIdentity } from './orchestrator/process-identity' +import type { FactorySweepSkipReasonCode } from './orchestrator/sweep-skip-reason' import type { DispatchRelayflowOptions, RelayflowPolicyRegistry } from './dispatch/relayflow-registry' import type { VerificationGate } from './environments/verification-pipeline' import type { CostLedger } from './cost/ledger' @@ -229,6 +230,38 @@ export interface FactoryReadinessReconcileStatus { lastFailureAtMs?: number /** Age of a pass that started and has neither completed nor failed. */ inFlightMs?: number + /** + * Work units the last *completed* 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. + * + * 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 + * 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. */ + dispatched?: number + /** Work units the last completed sweep saw and declined. */ + skipped?: number + /** + * `skipped` split by cause. Zero-count codes are omitted; the codes + * themselves are a fixed published vocabulary, so an absent key is a zero. + */ + skipReasons?: Partial> + /** + * The last completed sweep 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 + * `candidates: 0` — and those are opposite diagnoses (#355). + */ + discoveryDeferred?: 'sweep-in-flight' /** Free text; authenticated surfaces only. */ lastError?: string /** Allowlisted class name of `lastError`; publishable. */ @@ -250,6 +283,20 @@ export interface FactoryPublicReadinessReconcileHealth { inFlightMs?: number /** `inFlightMs` expressed in sweeps that should have run and did not. */ missedPasses?: number + /** + * The last completed 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. + */ + 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. */ + discoveryDeferred?: 'sweep-in-flight' lastErrorClass?: string } @@ -535,7 +582,11 @@ export interface IterationReport { pulled: IssueRef[] triaged: TriageDecision[] dispatched: DispatchResult[] - skipped: Array<{ issue: IssueRef; reason: string }> + /** + * `reason` is free text for an operator; `code` is the closed vocabulary + * that may cross onto the unauthenticated health surface (#355). + */ + skipped: Array<{ issue: IssueRef; reason: string; code?: FactorySweepSkipReasonCode }> dryRun: boolean slackDegraded?: boolean /**