diff --git a/.agentworkforce/features/manifest.yaml b/.agentworkforce/features/manifest.yaml index 6d288168..64a5da1c 100644 --- a/.agentworkforce/features/manifest.yaml +++ b/.agentworkforce/features/manifest.yaml @@ -1,11 +1,11 @@ version: '1.1' -updated: '2026-08-18' +updated: '2026-08-20' catalog: category_count: 25 - feature_count: 321 + feature_count: 323 tier_counts: - 1: 49 - 2: 127 + 1: 50 + 2: 128 3: 12 4: 52 5: 65 @@ -157,6 +157,14 @@ categories: location: src/cli/fleet.ts, src/orchestrator/factory.ts verify_tier: 2 + - id: cli-diagnose-deployed + name: Diagnose a Deployed Instance + cli: factory diagnose --deployed + api: diagnoseDeployedFactory() + description: Ask a deployed Factory over HTTP why it is or is not dispatching — public health block, readiness failure count, error class, in-flight sweep age and missed passes — with no credential, and read the gated /evidence lastError when an operator token is supplied + location: src/cli/diagnose.ts, src/cli/fleet.ts + verify_tier: 2 + - id: cli-kill-loop name: Stop Loop by Heartbeat PID cli: factory kill-loop @@ -1484,6 +1492,13 @@ categories: location: src/ports/state.ts, src/state/ verify_tier: 1 + - id: api-public-health + name: Public Health Projection API + api: publicHealthFromHeartbeat() / normalizePublicHealth() + description: Project the loop heartbeat into the record safe to serve unauthenticated — liveness split from an amber status, readiness failure count, allowlisted error class, and a stalled state derived from an in-flight sweep's age — with free text, paths, URLs and tokens excluded by construction + location: src/orchestrator/public-health.ts, src/observability/error-class.ts + verify_tier: 1 + - id: api-reaper name: Reaper API api: FactoryReaper / terminatePids() / reapFactoryOrphansOnce() diff --git a/README.md b/README.md index 3c7f581e..93f99dfe 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,7 @@ After init, add the `factory` label to an open issue and run a dry run below. | `factory triage ` | Triage one issue and print the decision. | | `factory dispatch ` | Triage + dispatch one issue. Honors `--dry-run`. | | `factory babysit ` | Spawn a one-shot babysitter for an existing open PR, even when it was not created by Factory. | +| `factory diagnose --deployed ` | Ask a **deployed** instance why it is or is not dispatching. Reads the unauthenticated `/healthz` diagnostics block — readiness failure count, error class, in-flight sweep age — with no credential; `--token` (or `FACTORY_EVIDENCE_TOKEN`) also reads the gated `/evidence` message. Exits non-zero when the instance is not dispatching. See [deployed diagnostics](docs/deployed-diagnostics.md). | | `factory canary ` | Assert a known "Ready for Agent" issue is dispatch-ready by the real dry-run triage path. Prints `{ok,issue,status,reason}`; exits non-zero (with the skip reason) if it isn't. | | `factory reap-orphans [--include-held]` | Report stale factory processes and held agents. By default held agents are visible but retained; `--include-held` releases held agents whose configured deadline has elapsed. Kubernetes cleanup is reported as not applicable when no environment provider is configured. | | `factory featuremap check [--manifest ] [--base ]` | Validate the repository feature/test manifest and optionally report advisory drift for unchanged entries whose locations changed. | @@ -183,6 +184,14 @@ state advances from `retrying` to `degraded` after three consecutive periodic discovery failures and returns to `healthy` after a successful checkpoint. The same record includes the last error, attempt duration, and failure count. +Those counters only move when a sweep *settles*, so a sweep that hangs would +leave every field reading `healthy` forever. `readinessReconcile` therefore +also carries `intervalMs` and `inFlightMs`, and reports `stalled` when a pass +has been in flight for more than ten sweep intervals. The redacted subset of +this record is published on the deployed instance's `/healthz` and is what +`factory diagnose --deployed` reads — see +[deployed diagnostics](docs/deployed-diagnostics.md). + Dispatch lifecycle writes are claim-critical. Factory applies the `factory:in-progress` label/state before the dispatch comment, confirms the GitHub label by provider read-back, and retries either write three times. An diff --git a/docs/deployed-diagnostics.md b/docs/deployed-diagnostics.md new file mode 100644 index 00000000..d0566a64 --- /dev/null +++ b/docs/deployed-diagnostics.md @@ -0,0 +1,161 @@ +# Diagnosing a deployed Factory + +> Issue: [#295](https://github.com/AgentWorkforce/factory/issues/295) · companion: AgentWorkforce/factory-cloud `fix/295-healthz-diagnostics` — a deployed Factory had no +> operator-reachable diagnostics. The field naming the 2026-08-19/20 outage existed the whole time +> and was unreachable for ~10 hours. + +## The command + +```sh +factory diagnose --deployed https:// +``` + +No credential required. It reads the unauthenticated `/healthz` and answers one question — *is this +instance dispatching, and if not, why* — with a non-zero exit when the answer is no, so a lane brief +or a cron entry can act on `$?` alone. + +```sh +factory diagnose --deployed # human-readable +factory diagnose --deployed --json # the same diagnosis as JSON +factory diagnose --deployed --token # also read the gated /evidence +factory diagnose --deployed --timeout-ms 30000 +``` + +`--token` defaults to `FACTORY_EVIDENCE_TOKEN` when set. Without it the command still works; it just +cannot show the free-text `lastError`, and says so. + +Exit codes follow `factory canary`: `0` when the instance is dispatching, `1` when it is not or +cannot be reached. + +## Why the other routes do not work + +| route | why not | +|---|---| +| `factory status`, `factory loop-status` | inspect a **local** instance only | +| `wrangler tail` | Worker scope — the Factory process runs in the Container and its stdout does not surface | +| `wrangler containers ssh` | WebSocket 400: the container is private-networked with no sshd | +| `GET /evidence` | carries the answer, but is bearer-gated by a token minted per deploy and destroyed at the end of the run that created it | + +Log egress therefore has to be pull-through-the-Worker. `factory diagnose --deployed` is that pull. + +## What `/healthz` now carries + +The daemon writes a redacted projection of its loop heartbeat — `heartbeat.health`, built by +`publicHealthFromHeartbeat()` — and the container serves it verbatim. The container has no redaction +logic of its own by design: the boundary lives in one place, in this repo, with tests. + +```jsonc +// The daemon stamps this when it WRITES the heartbeat, so `ageMs` is 0 and +// `stale` false in the file; freshness is `updatedAtMs` against the clock of +// whoever serves it. Here now = 1787229155805 (2026-08-20T12:32:35.805Z). +{ + "schemaVersion": 1, + "ok": true, // process liveness — see below + "status": "degraded", // the amber + "stale": false, + "updatedAtMs": 1787229155805, + "ageMs": 0, + "loopStatus": "running", + "degradedSubsystems": ["readinessReconcile"], + "reason": "dispatch-gating subsystem not healthy: readinessReconcile", + "readinessReconcile": { + "state": "stalled", // not-running | healthy | retrying | degraded | stalled + "consecutiveFailures": 0, + "failureThreshold": 3, + "intervalMs": 60000, + "lastStartedAtMs": 1787224595805, // 11:16:35.805Z + "lastCompletedAtMs": 1787224535802, // 11:15:35.802Z — 60s EARLIER + "inFlightMs": 4560000, // now − lastStarted: this pass has run 76 minutes + "missedPasses": 76, + "lastErrorClass": "TimeoutError" + }, + "eventListener": { "state": "subscribed" }, + "fleetControlPlane": { "state": "closed", "consecutiveFailures": 0, "failureThreshold": 3 } +} +``` + +### Reading it + +- **`consecutiveFailures` / `lastErrorClass`** — the failing case. During the outage this read 7 then + 8 while `/healthz` said `ok: true` and published nothing but the string `degraded`. +- **`lastStartedAtMs` vs `lastCompletedAtMs`** — the *silent* case. A sweep that hangs takes neither + the success nor the failure path, so no state is written and every settled field keeps reading + green. `lastStarted > lastCompleted` is the only evidence that a pass is in flight, and `inFlightMs` + says for how long. +- **`fleetControlPlane`** — an `open` circuit fails every spawn and resume fast, so it gates dispatch + as hard as a failing sweep. `closed` is the healthy value. +- **`state: "stalled"`** — derived, not written: an in-flight pass older than ten sweep intervals. + A cold container legitimately spends minutes in its first pass (#36 measured 61 minutes while the + Relayfile mirror hydrated), so check `lastCompletedAtMs`: absent means "first pass since boot, + still hydrating"; present and hours old means "was fine, then wedged". + +### Why `ok` stays `true` while `status` goes amber + +`/healthz` is the Cloudflare **Container ping endpoint** (`pingEndpoint = 'localhost/healthz'` in the +Worker). A non-200 there is a liveness verdict the platform acts on: it recycles the container. That +would destroy the in-memory evidence of the wedge and restart the cold-start hydration — turning a +diagnosable degradation into a restart loop that also erases its own cause. + +So the two questions are split: + +- `ok` — *is this process alive?* Unchanged semantics, safe to keep driving the ping and the HTTP + status code. +- `status` (`ok` / `degraded` / `unknown`) and `degradedSubsystems` — *is dispatch gated?* No platform + reads these, so a monitor can alert on `status != "ok"` with no lifecycle side effect. + +A liveness endpoint that cannot go amber is not much of a signal — this one goes amber in a field +that cannot restart the box. + +## What never crosses + +`lastError` is dependency-controlled free text and routinely carries provider prose, filesystem paths +and URLs with credentials in the query string. It stays on the authenticated `/evidence` surface. The +public block carries only its **class**, through the same allowlist that guards +`IterationReport.skipped[].reason` (`src/observability/error-class.ts`): a pattern-checked class name, +falling back to `Error`. + +Every other field is constructed explicitly and validated for what it is — states against closed +enums, counters and timestamps coerced with range and sign checks, `degradedSubsystems` filtered to +a fixed set of names, and the one assembled string (`reason`) built from those same names, then +control-stripped and length-bounded. Nothing is spread, so a field added upstream cannot reach the +public surface by default. See `src/orchestrator/public-health.ts`. + +Regression coverage: `src/orchestrator/public-health.test.ts` and the `#295` block in +`src/orchestrator/factory.test.ts` feed a `lastError` containing a path, a URL and a token and assert +none of it appears in the published record. + +## Serving the block (factory-cloud) + +The container entrypoint passes the block through unchanged: + +```js +// container/entrypoint.mjs — publicHeartbeat() +return { + status: parsed.status, + updatedAt: parsed.updatedAt, + updatedAtMs: parsed.updatedAtMs, + eventListener: parsed.eventListener?.state, + readinessReconcile: parsed.readinessReconcile?.state, + health: parsed.health, // already redacted by the daemon +} +``` + +so `/healthz` answers `{ ok, phase, factoryProcess, heartbeat: { …, health } }`. `factory diagnose` +reads the block from `heartbeat.health`, and accepts a top-level `health` as well. + +Instances running a Factory older than this change publish no `health` block; `factory diagnose` +detects that and says so rather than reporting a false green. + +Two other shapes the command refuses to read as green: + +- **Event-driven short-sleep mode.** With `FACTORY_EVENT_DRIVEN_SLEEP_ENABLED=1` the Worker answers + `/healthz` itself and never probes the container, deliberately — anonymous polling must not be a + second wake path. That response (`phase: "worker-ready"`, `container: "not-probed"`) is Worker + liveness and carries no Factory health, so `factory diagnose` reports *cannot tell* and points at + `/evidence`, which does reach the container. +- **A container serving a heartbeat its daemon stopped updating.** The block's own `stale`/`ageMs` + are not measurements of a read — they are constants of the write: `ageMs` is always `0` and + `stale` always `false` in the file, whether that file is one second or one week old. Freshness + comes from `updatedAtMs` measured against the clock of whoever serves it, which is what the + container does on every request; that verdict (`ok: false`, HTTP 503) outranks anything the block + still claims. diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts new file mode 100644 index 00000000..e7e97703 --- /dev/null +++ b/src/cli/diagnose.test.ts @@ -0,0 +1,710 @@ +import { describe, expect, it } from 'vitest' + +import { runFleetCli } from './fleet' + +const BASE = 'https://factory.example.com' +// An ambient FACTORY_EVIDENCE_TOKEN on a developer machine or CI runner would +// otherwise make these commands request /evidence as well, which the stubs do +// not route (#300 review, CodeRabbit). +const HERMETIC_ENV = {} as NodeJS.ProcessEnv +const NOW_MS = 1_787_224_000_000 + +const buffer = () => { + let value = '' + return { + write(chunk: string) { + value += chunk + return true + }, + text() { + return value + }, + } +} + +interface StubRoutes { + healthz?: { status: number; body: unknown } + evidence?: { status: number; body: unknown } +} + +const stubFetch = (routes: StubRoutes, seen: string[] = []): typeof fetch => + (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = typeof input === 'string' ? input : input instanceof URL ? input.toString() : input.url + const authorization = new Headers(init?.headers ?? {}).get('authorization') + seen.push(`${url}${authorization ? ` auth=${authorization}` : ''}`) + const route = url.endsWith('/evidence') ? routes.evidence : routes.healthz + if (!route) throw Object.assign(new Error('fetch failed'), { name: 'TypeError' }) + return new Response(JSON.stringify(route.body), { + status: route.status, + headers: { 'content-type': 'application/json' }, + }) + }) as typeof fetch + +const healthy = { + ok: true, + phase: 'running', + factoryProcess: 'running', + health: { + schemaVersion: 1, + ok: true, + status: 'ok', + stale: false, + updatedAtMs: NOW_MS, + ageMs: 12_000, + loopStatus: 'running', + degradedSubsystems: [], + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: NOW_MS - 30_000, + lastCompletedAtMs: NOW_MS - 29_000, + }, + eventListener: { state: 'subscribed' }, + }, +} + +describe('factory diagnose --deployed (#295)', () => { + it('reports a healthy deployed instance and exits zero without any credential', async () => { + const seen: string[] = [] + const out = buffer() + const err = buffer() + + const code = await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: err, + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ healthz: { status: 200, body: healthy } }, seen), + }) + + expect(code).toBe(0) + expect(seen).toEqual([`${BASE}/healthz`]) + expect(out.text()).toContain('dispatching') + expect(out.text()).toContain('readinessReconcile') + }) + + // The 2026-08-19/20 outage: eight consecutive failures behind `ok: true`. + it('names the failing subsystem, its failure count and its error class', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + health: { + schemaVersion: 1, + ok: true, + status: 'degraded', + stale: false, + loopStatus: 'running', + degradedSubsystems: ['readinessReconcile'], + readinessReconcile: { + state: 'degraded', + consecutiveFailures: 8, + failureThreshold: 3, + intervalMs: 60_000, + lastErrorClass: 'DispatchLifecycleError', + }, + }, + }, + }, + }), + }) + + // A lane briefed to "find out why production is not dispatching" must get + // a non-zero exit and a named subsystem out of one command. + expect(code).not.toBe(0) + const text = out.text() + expect(text).toContain('readinessReconcile') + expect(text).toContain('8') + expect(text).toContain('DispatchLifecycleError') + expect(text).toContain('not dispatching') + }) + + // The 2026-08-20 case: every settled field green, one pass wedged for 77m. + it('calls out a stalled sweep and how many passes it has missed', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + health: { + schemaVersion: 1, + ok: true, + status: 'degraded', + stale: false, + loopStatus: 'running', + degradedSubsystems: ['readinessReconcile'], + readinessReconcile: { + state: 'stalled', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + inFlightMs: 4_620_000, + missedPasses: 77, + lastStartedAtMs: NOW_MS, + lastCompletedAtMs: NOW_MS - 60_003, + }, + }, + }, + }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { + dispatching: boolean + verdict: string + health?: { readinessReconcile?: { state?: string; missedPasses?: number } } + } + expect(report.dispatching).toBe(false) + expect(report.health?.readinessReconcile).toMatchObject({ state: 'stalled', missedPasses: 77 }) + expect(report.verdict).toContain('stalled') + }) + + // The deployed container serves the block inside its heartbeat projection. + it('reads the health block where the container actually serves it', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + heartbeat: { + status: 'running', + readinessReconcile: 'healthy', + eventListener: 'subscribed', + health: healthy.health, + }, + }, + }, + }), + }) + + expect(code).toBe(0) + const report = JSON.parse(out.text()) as { dispatching: boolean; health?: { status?: string } } + expect(report.dispatching).toBe(true) + expect(report.health?.status).toBe('ok') + }) + + it('says what is missing when the instance predates the diagnostics block', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + heartbeat: { status: 'running', readinessReconcile: 'degraded', eventListener: 'subscribed' }, + }, + }, + }), + }) + + expect(code).not.toBe(0) + expect(out.text()).toContain('state strings only') + expect(out.text()).toContain('degraded') + }) + + it('reads the gated evidence surface when an operator token is supplied', async () => { + const seen: string[] = [] + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--token', 'op-token', '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { status: 200, body: healthy }, + evidence: { + status: 200, + body: { + phase: 'running', + heartbeat: { status: 'running' }, + readinessReconcile: { + state: 'degraded', + consecutiveFailures: 8, + lastError: 'Refusing to dispatch AR-241: dispatch lifecycle is already terminal', + }, + }, + }, + }, seen), + }) + + expect(code).toBe(0) + expect(seen).toEqual([`${BASE}/healthz`, `${BASE}/evidence auth=Bearer op-token`]) + const report = JSON.parse(out.text()) as { evidence?: { fetched?: boolean; lastError?: string } } + expect(report.evidence?.fetched).toBe(true) + expect(report.evidence?.lastError).toContain('dispatch lifecycle is already terminal') + }) + + it('still diagnoses when the evidence token is rejected', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--token', 'stale-token', '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { status: 200, body: healthy }, + evidence: { status: 401, body: { error: 'unauthorized' } }, + }), + }) + + expect(code).toBe(0) + const report = JSON.parse(out.text()) as { evidence?: { fetched?: boolean; reason?: string } } + expect(report.evidence?.fetched).toBe(false) + expect(report.evidence?.reason).toContain('401') + }) + + it('reports an unreachable instance as a failure rather than silence', async () => { + const out = buffer() + const err = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: err, + env: HERMETIC_ENV, + diagnoseFetch: (async () => { + throw Object.assign(new Error('connect ECONNREFUSED 10.0.0.1:443'), { name: 'TypeError' }) + }) as typeof fetch, + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { reachable: boolean; errorClass?: string } + expect(report.reachable).toBe(false) + // Class only: an error message from a private-networked host names hosts + // and paths the report does not need. + expect(report.errorClass).toBe('TypeError') + }) + + it('requires the target url', async () => { + const err = buffer() + const code = await runFleetCli(['diagnose'], { stdout: buffer(), stderr: err }) + + expect(code).not.toBe(0) + expect(err.text()).toContain('--deployed') + }) + + it('is listed in help so a lane brief can name it', async () => { + const out = buffer() + await runFleetCli(['--help'], { stdout: out, stderr: buffer() }) + + expect(out.text()).toContain('diagnose --deployed ') + }) + // Review follow-up on #300 (P1, codex). The daemon stamps the block at write + // time, so its `ageMs` is 0 and `stale` false *in the file*. If the daemon + // dies and the container keeps serving that file, believing the embedded + // snapshot reports green forever — the exact failure this command exists to + // catch. The container computes liveness against its own clock; that verdict + // wins. + it('believes the container liveness verdict over a frozen health snapshot', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 503, + body: { + // The container's own staleness check, against its own clock. + ok: false, + phase: 'running', + factoryProcess: 'running', + heartbeat: { + status: 'running', + updatedAtMs: NOW_MS - 3_600_000, + readinessReconcile: 'healthy', + health: healthy.health, + }, + }, + }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string } + expect(report.dispatching).toBe(false) + expect(report.verdict).toMatch(/heartbeat|liveness|not alive/iu) + }) + + // Review follow-up on #300 (P2, codex). `new Date(1e300).toISOString()` + // throws, and a remote instance chooses these numbers. + it('renders an out-of-range remote timestamp as unknown instead of aborting', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + health: { + ...healthy.health, + readinessReconcile: { + ...healthy.health.readinessReconcile, + lastStartedAtMs: 1e300, + lastCompletedAtMs: 1e300, + }, + }, + }, + }, + }), + }) + + expect(code).toBe(0) + const text = out.text() + expect(text).toContain('readinessReconcile') + expect(text).not.toContain('Invalid time value') + }) + // Review follow-up on #300 (P1, cubic). A block with no readiness subsystem + // in it says nothing about dispatch; "no degraded subsystem listed" is not + // the same statement as "the sweep is healthy". + it('refuses to call an incomplete health block dispatching', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + health: { + schemaVersion: 1, + ok: true, + status: 'ok', + stale: false, + loopStatus: 'running', + degradedSubsystems: [], + }, + }, + }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string } + expect(report.dispatching).toBe(false) + expect(report.verdict).toMatch(/cannot tell|no readiness/iu) + }) + + // Review follow-up on factory-cloud#40 (P2, codex). In event-driven + // short-sleep mode the Worker answers /healthz itself and never probes the + // container, deliberately, so anonymous polling cannot defeat scale-to-zero. + // That response is Worker liveness — reading it as "Factory is dispatching" + // is exactly the false green this command exists to prevent. + it('does not read a worker-only short-sleep response as a dispatching Factory', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { ok: true, phase: 'worker-ready', container: 'not-probed', eventDrivenSleep: true }, + }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string } + expect(report.dispatching).toBe(false) + expect(report.verdict).toMatch(/short-sleep|not probed|worker/iu) + expect(report.verdict).toContain('/evidence') + }) + + // Review follow-up on #300 (P2, cubic). A hermetic env must be honoured, or + // a test — or an embedder — silently skips the authenticated read. + it('takes the evidence token from the injected environment', async () => { + const seen: string[] = [] + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: buffer(), + stderr: buffer(), + env: { FACTORY_EVIDENCE_TOKEN: 'env-token' } as NodeJS.ProcessEnv, + diagnoseFetch: stubFetch({ + healthz: { status: 200, body: healthy }, + evidence: { status: 200, body: { phase: 'running' } }, + }, seen), + }) + + expect(code).toBe(0) + expect(seen).toEqual([`${BASE}/healthz`, `${BASE}/evidence auth=Bearer env-token`]) + }) + // A container in `preflight` also answers ok:false, and "the Factory process + // is gone" is the wrong thing to tell someone whose instance is three + // minutes into a boot. The phase is right there in the response. + it('distinguishes a booting instance from a wedged one', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 503, + body: { ok: false, phase: 'preflight', factoryProcess: 'not-running' }, + }, + }), + }) + + expect(code).not.toBe(0) + const text = out.text() + expect(text).toContain('preflight') + expect(text).toMatch(/still starting|booting/iu) + expect(text).not.toContain('the Factory process is gone') + }) + // Found while running the built CLI against a short-sleep stub: the report + // said "instance predates #295" about an instance whose age it cannot know, + // and printed `phase` twice. Sending an operator to "upgrade the deployed + // Factory" when the real answer is "the Worker never asked the container" is + // the wrong-problem failure this command exists to prevent. + it('does not blame the instance version when the Worker simply did not probe it', async () => { + const out = buffer() + await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { ok: true, phase: 'worker-ready', container: 'not-probed', eventDrivenSleep: true }, + }, + }), + }) + + const text = out.text() + expect(text).not.toContain('predates') + expect(text).toMatch(/short-sleep/iu) + expect(text.match(/^ +phase +:/gmu)?.length ?? 0).toBe(1) + }) + + it('prints the phase once when the instance predates the block', async () => { + const out = buffer() + await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + heartbeat: { status: 'running', readinessReconcile: 'degraded', eventListener: 'subscribed' }, + }, + }, + }), + }) + + const text = out.text() + expect(text).toContain('predates') + expect(text.match(/^ +phase +:/gmu)?.length ?? 0).toBe(1) + }) + // Review follow-up on #300 (P3, cubic). Introduced by my own refactor: with + // only one legacy state string present, the other rendered as the literal + // "undefined" in the verdict. + it('says unknown, never undefined, for a legacy state string that is missing', async () => { + const out = buffer() + await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + heartbeat: { status: 'running', eventListener: 'subscribed' }, + }, + }, + }), + }) + + const report = JSON.parse(out.text()) as { verdict: string } + expect(report.verdict).not.toContain('undefined') + expect(report.verdict).toContain('readinessReconcile=unknown') + }) + // Review follow-up on #300 (Major, CodeRabbit). `live` was false for any + // non-200, so a fronting proxy answering 404/502 produced a confident + // statement about a Factory that was never asked. + it('does not claim a Factory diagnosis from a proxy answer', async () => { + for (const status of [404, 401, 502]) { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ healthz: { status, body: { error: 'nope' } } }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string } + expect(report.dispatching).toBe(false) + expect(report.verdict).toContain(`HTTP ${status}`) + expect(report.verdict).toMatch(/cannot tell/iu) + // The claims this response cannot support. + expect(report.verdict).not.toContain('reports itself not live') + expect(report.verdict).not.toContain('Factory process is gone') + } + }) + + it('still treats a 503 carrying the instance own verdict as the instance speaking', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { status: 503, body: { ok: false, phase: 'running', factoryProcess: 'running' } }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { verdict: string } + expect(report.verdict).toContain('reports itself not live') + }) + + // Review follow-up on #300 (Minor, CodeRabbit). `unknown` means the block did + // not say, which is not the same statement as "a subsystem is degraded". + it('reports an unreadable status as cannot tell rather than as degradation', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ok: true, + phase: 'running', + health: { + schemaVersion: 1, + ok: true, + status: 'sideways', + stale: false, + loopStatus: 'running', + degradedSubsystems: [], + readinessReconcile: { state: 'healthy', consecutiveFailures: 0, failureThreshold: 3 }, + }, + }, + }, + }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { verdict: string } + expect(report.verdict).toMatch(/cannot tell/iu) + expect(report.verdict).not.toContain('a subsystem is degraded') + }) + + // Review follow-up on #300 (Minor, CodeRabbit). A 404 means the deployment + // has no /evidence route and a 5xx means it failed; neither is a reason to + // send someone rotating a credential that works. + it('blames the token only when the token is what was refused', async () => { + const cases = [ + { status: 401, expect: /token/iu }, + { status: 403, expect: /token/iu }, + { status: 404, expect: /no \/evidence route|not exposed/iu }, + { status: 502, expect: /failed|error/iu }, + ] + for (const testCase of cases) { + const out = buffer() + await runFleetCli(['diagnose', '--deployed', BASE, '--token', 'op-token', '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { status: 200, body: healthy }, + evidence: { status: testCase.status, body: {} }, + }), + }) + + const report = JSON.parse(out.text()) as { evidence?: { reason?: string } } + expect(report.evidence?.reason).toMatch(testCase.expect) + if (testCase.status === 404 || testCase.status === 502) { + expect(report.evidence?.reason).not.toContain('token was not accepted') + } + } + }) + + // Review follow-up on #300 (Security, CodeRabbit). `factory diagnose + // my-evidence-token` puts a credential in argv; echoing it into the error + // puts it on stderr and from there into CI logs. + it('never echoes an unrecognized argument value into the error', async () => { + const err = buffer() + const code = await runFleetCli(['diagnose', BASE, 's3cr3t-evidence-token'], { + stdout: buffer(), + stderr: err, + env: HERMETIC_ENV, + }) + + expect(code).not.toBe(0) + expect(err.text()).not.toContain('s3cr3t-evidence-token') + expect(err.text()).toMatch(/argument 2|second argument|position/iu) + }) + // Review follow-up on #300 (P2, cubic). My own no-echo fix was incomplete: + // a token in the URL slot still reached stderr through the scheme check. + it('never echoes a value that landed in the url slot either', async () => { + for (const argv of [ + ['diagnose', 's3cr3t-evidence-token'], + ['diagnose', '--deployed', 's3cr3t-evidence-token'], + ['diagnose', '--url', 's3cr3t-evidence-token'], + ]) { + const err = buffer() + const code = await runFleetCli(argv, { stdout: buffer(), stderr: err, env: HERMETIC_ENV }) + + expect(code).not.toBe(0) + expect(err.text()).not.toContain('s3cr3t-evidence-token') + expect(err.text()).toMatch(/http/iu) + } + }) + + // Review follow-up on #300 (P2, cubic). A gateway can answer 200 or 503 with + // its own body; the container's health response always carries `ok`. + it('treats a 200 or 503 with no ok field as something other than the instance', async () => { + for (const status of [200, 503]) { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ healthz: { status, body: { message: 'gateway timeout page' } } }), + }) + + expect(code).not.toBe(0) + const report = JSON.parse(out.text()) as { live?: boolean; verdict: string } + expect(report.live).toBeUndefined() + expect(report.verdict).toMatch(/cannot tell/iu) + expect(report.verdict).not.toContain('reports itself not live') + } + }) +}) + diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts new file mode 100644 index 00000000..c7c619f2 --- /dev/null +++ b/src/cli/diagnose.ts @@ -0,0 +1,484 @@ +import { telemetryErrorClass } from '../observability/error-class.js' +import { normalizePublicHealth } from '../orchestrator/public-health.js' +import type { FactoryPublicHealth } from '../types' + +/** + * `factory diagnose --deployed ` (#295). + * + * The command a lane brief can name. Every other route to a deployed + * Factory's health is local-only (`factory status`, `factory loop-status`), + * Worker-scope (`wrangler tail`), blocked by network shape (the container is + * private-networked with no sshd), or gated behind a per-deploy credential + * that no longer exists. This pulls the answer through the Worker, which is + * the only path that survives that shape. + * + * It works with no credential at all: `/healthz` is unauthenticated, and + * since #295 it carries the failure count, the error class and the in-flight + * age. A token, when the operator has one, adds `/evidence` — the free-text + * `lastError` that must not be on the public surface. + */ +export interface DeployedEvidenceSummary { + fetched: boolean + httpStatus?: number + /** Why the evidence surface was not read. */ + reason?: string + phase?: string + lastError?: string + consecutiveFailures?: number +} + +export interface DeployedLegacyHealth { + phase?: string + factoryProcess?: string + heartbeatStatus?: string + heartbeatUpdatedAt?: string + readinessReconcile?: string + eventListener?: string +} + +export interface DeployedFactoryDiagnosis { + url: string + reachable: boolean + httpStatus?: number + /** + * The instance's own liveness verdict, computed against ITS clock. + * + * The daemon stamps the health block when it writes the heartbeat, so the + * block's `ageMs` is 0 and `stale` false *in the file*. If the daemon dies + * and the container keeps serving that file, the block stays green forever. + * The container recomputes liveness from `updatedAtMs` on every request, so + * its verdict is the fresh one and it wins (#300 review, P1). + * + * `false` only when the INSTANCE answered no — a 503 health body, or + * `ok: false`. A 404, 401 or 502 from a proxy in front of it is not the + * instance speaking at all, and leaves this `undefined` (#300 review, + * CodeRabbit). + */ + live?: boolean + /** The endpoint answered, but not with anything this command can read. */ + unreadable?: boolean + /** Allowlisted class of a transport failure; the message is not reported. */ + errorClass?: string + dispatching: boolean + verdict: string + health?: FactoryPublicHealth + /** + * The Worker answered without probing the container. + * + * In event-driven short-sleep mode `/healthz` terminates at the Worker on + * purpose, so anonymous polling cannot wake the container and defeat + * scale-to-zero. That answer is Worker liveness and says nothing about + * Factory (factory-cloud#40 review). + */ + workerOnly?: boolean + /** + * The container's own bootstrap phase, when it reports one. + * + * `booting`/`rendering-config`/`preflight` answer `ok: false` exactly like a + * wedged instance does, and telling someone three minutes into a boot that + * their Factory process is gone sends them to the wrong problem. + */ + phase?: string + /** Present when the instance predates the `/healthz` diagnostics block. */ + legacy?: DeployedLegacyHealth + evidence?: DeployedEvidenceSummary +} + +export interface DiagnoseDeployedOptions { + url: string + token?: string + timeoutMs?: number + fetch?: typeof fetch +} + +const DEFAULT_TIMEOUT_MS = 10_000 + +/** Container bootstrap phases: `ok: false` here means "not yet", not "wedged". */ +const BOOT_PHASES = new Set(['booting', 'rendering-config', 'preflight']) +const MAX_EVIDENCE_TEXT = 2_000 + +const endpoint = (base: string, path: string): string => `${base.replace(/\/+$/u, '')}${path}` + +/** + * Strip control characters before anything remote reaches a terminal. + * + * `/evidence` is authenticated and returns the operator's own instance, but + * its `lastError` is still dependency-controlled text, and a terminal + * interprets escape sequences. + */ +const forTerminal = (value: string): string => + // C0 and C1 alike (#300 review, P2, cubic): some terminals treat the C1 + // range as escape introducers, so stripping only C0 is not enough. + value.replace(/[\u0000-\u001F\u007F-\u009F]+/gu, ' ').trim().slice(0, MAX_EVIDENCE_TEXT) + +const asRecord = (value: unknown): Record => + value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : {} + +const asText = (value: unknown): string | undefined => + typeof value === 'string' && value.trim() ? forTerminal(value) : undefined + +const asCount = (value: unknown): number | undefined => + typeof value === 'number' && Number.isFinite(value) ? value : undefined + +async function getJson( + fetchImpl: typeof fetch, + url: string, + opts: { token?: string; timeoutMs: number }, +): Promise<{ status: number; body: unknown }> { + const response = await fetchImpl(url, { + method: 'GET', + headers: { + accept: 'application/json', + ...(opts.token ? { authorization: `Bearer ${opts.token}` } : {}), + }, + signal: AbortSignal.timeout(opts.timeoutMs), + }) + let body: unknown + try { + body = await response.json() + } catch { + body = undefined + } + return { status: response.status, body } +} + +function legacyHealth(body: Record): DeployedLegacyHealth { + const heartbeat = asRecord(body.heartbeat) + return { + ...(asText(body.phase) ? { phase: asText(body.phase) } : {}), + ...(asText(body.factoryProcess) ? { factoryProcess: asText(body.factoryProcess) } : {}), + ...(asText(heartbeat.status) ? { heartbeatStatus: asText(heartbeat.status) } : {}), + ...(asText(heartbeat.updatedAt) ? { heartbeatUpdatedAt: asText(heartbeat.updatedAt) } : {}), + ...(asText(heartbeat.readinessReconcile) + ? { readinessReconcile: asText(heartbeat.readinessReconcile) } + : {}), + ...(asText(heartbeat.eventListener) ? { eventListener: asText(heartbeat.eventListener) } : {}), + } +} + +function verdictFor(diagnosis: Omit): { + dispatching: boolean + verdict: string +} { + if (!diagnosis.reachable) { + return { + dispatching: false, + verdict: `unreachable: ${diagnosis.url} did not answer (${diagnosis.errorClass ?? 'no response'})`, + } + } + // Before any subsystem reading: a snapshot served by a container whose + // daemon has stopped updating it describes a process that is no longer + // there. The instance already said so. + if (diagnosis.live === false) { + if (diagnosis.phase !== undefined && BOOT_PHASES.has(diagnosis.phase)) { + return { + dispatching: false, + verdict: + `not dispatching yet: the instance is still starting (phase ${diagnosis.phase}). ` + + 'A cold container hydrates its Relayfile mirror before the first pass, which #36 measured ' + + 'at up to 61 minutes; re-run this in a few minutes before treating it as wedged.', + } + } + return { + dispatching: false, + verdict: + `not dispatching: the instance reports itself not live (HTTP ${diagnosis.httpStatus ?? '?'}` + + `${diagnosis.phase ? `, phase ${diagnosis.phase}` : ''}). ` + + 'Its loop heartbeat is stale or the Factory process is gone, so any health block it still ' + + 'serves describes the last write, not the present.', + } + } + if (diagnosis.unreadable) { + return { + dispatching: false, + verdict: + `cannot tell: the endpoint answered HTTP ${diagnosis.httpStatus ?? '?'} and carried no Factory ` + + 'health. Something other than the instance may be answering this URL — a gateway, an auth ' + + 'proxy, or a load balancer. Check the URL, then pass --token to read /evidence.', + } + } + if (diagnosis.workerOnly) { + return { + dispatching: false, + verdict: + 'cannot tell: this deployment answers /healthz at the Worker without probing the container ' + + '(event-driven short-sleep), so the response is Worker liveness and carries no Factory ' + + 'health. Pass --token to read /evidence, which does reach the container.', + } + } + const health = diagnosis.health + if (!health) { + const legacy = diagnosis.legacy ?? {} + if (!legacy.readinessReconcile && !legacy.eventListener) { + return { + dispatching: false, + verdict: + 'cannot tell: this response carried no Factory health at all — no diagnostics block and ' + + 'no subsystem state strings. Pass --token to read /evidence.', + } + } + return { + dispatching: false, + verdict: + 'cannot tell: this instance predates the /healthz diagnostics block (#295), so it publishes ' + + `state strings only — readinessReconcile=${legacy.readinessReconcile ?? 'unknown'}, ` + + `eventListener=${legacy.eventListener ?? 'unknown'}. ` + + 'Upgrade the deployed Factory, or pass --token to read /evidence.', + } + } + if (health.stale || health.loopStatus === 'stopping') { + return { + dispatching: false, + verdict: `not dispatching: ${health.reason ?? 'the loop heartbeat is not current'}`, + } + } + const readiness = health.readinessReconcile + if (readiness?.state === 'stalled') { + const missed = readiness.missedPasses ?? 0 + return { + dispatching: false, + verdict: + `not dispatching: the readiness sweep is stalled — one pass has been in flight for ` + + `${formatDuration(readiness.inFlightMs)} (${missed} missed passes at ` + + `${formatDuration(readiness.intervalMs)} cadence). The loop only re-arms when a sweep ` + + 'settles, so a hung pass stops dispatch permanently.', + } + } + if (readiness && (readiness.state === 'degraded' || readiness.state === 'retrying')) { + return { + dispatching: false, + verdict: + `not dispatching: readinessReconcile is ${readiness.state} after ` + + `${readiness.consecutiveFailures} consecutive failures ` + + `(threshold ${readiness.failureThreshold}), last failure class ` + + `${readiness.lastErrorClass ?? 'unknown'}. ` + + 'Pass --token to read the message at /evidence.', + } + } + if (health.eventListener?.state === 'not-listening') { + return { + dispatching: false, + verdict: 'not dispatching: the daemon is not listening for Relayfile events.', + } + } + if (health.status === 'unknown') { + return { + dispatching: false, + verdict: + 'cannot tell: the health block did not report a status this version understands. ' + + 'Pass --token to read /evidence.', + } + } + if (health.status !== 'ok') { + return { dispatching: false, verdict: `not dispatching: ${health.reason ?? 'a subsystem is degraded'}` } + } + // Review follow-up on #300 (P1, cubic). An empty `degradedSubsystems` on a + // block that never reported the readiness sweep is an absence of evidence, + // not evidence of health. + if (!readiness || readiness.state === 'unknown') { + return { + dispatching: false, + verdict: + 'cannot tell: the health block carries no readiness-reconcile state, so nothing here says ' + + 'whether discovery is running. Pass --token to read /evidence.', + } + } + if (readiness.state === 'not-running') { + return { + dispatching: false, + verdict: + 'not dispatching: the readiness loop is not running — this instance is not a live daemon.', + } + } + return { + dispatching: true, + verdict: + 'dispatching: readinessReconcile is healthy' + + (readiness?.intervalMs ? ` on a ${formatDuration(readiness.intervalMs)} cadence` : '') + + `, and the event listener is ${health.eventListener?.state ?? 'unknown'}.`, + } +} + +export function formatDuration(ms: number | undefined): string { + if (ms === undefined) return 'unknown' + if (ms < 1_000) return `${ms}ms` + const seconds = Math.floor(ms / 1_000) + if (seconds < 60) return `${seconds}s` + const minutes = Math.floor(seconds / 60) + if (minutes < 60) return `${minutes}m ${seconds % 60}s` + return `${Math.floor(minutes / 60)}h ${minutes % 60}m` +} + +/** Read a deployed instance's public health, plus `/evidence` when a token is held. */ +export async function diagnoseDeployedFactory( + options: DiagnoseDeployedOptions, +): Promise { + const fetchImpl = options.fetch ?? fetch + const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS + const url = options.url.replace(/\/+$/u, '') + + let base: Omit + try { + const health = await getJson(fetchImpl, endpoint(url, '/healthz'), { timeoutMs }) + const body = asRecord(health.body) + // The container serves the daemon's block inside its heartbeat projection; + // accept a top-level copy too, so a proxy that hoists it still works. + const published = normalizePublicHealth(body.health ?? asRecord(body.heartbeat).health) + const workerOnly = body.eventDrivenSleep === true || body.container === 'not-probed' + // A container answering its own health says `ok`; the 503 path is the + // service-level negative. Anything else that answers on this URL — a + // gateway 404, an auth proxy 401, a load balancer 502 — never asked the + // container, and cannot support a statement about Factory. + // The container's health response always carries a top-level `ok`. A + // status code alone is not enough: a gateway can answer 200 with an error + // page or 503 with its own, and neither asked the container (#300 review, + // cubic). + const instanceAnswered = typeof body.ok === 'boolean' + base = { + url, + reachable: true, + httpStatus: health.status, + ...(instanceAnswered + ? { live: health.status === 200 && body.ok !== false } + : { unreadable: true }), + ...(asText(body.phase) ? { phase: asText(body.phase) } : {}), + ...(workerOnly ? { workerOnly: true } : {}), + ...(published ? { health: published } : { legacy: legacyHealth(body) }), + } + } catch (error) { + base = { url, reachable: false, errorClass: telemetryErrorClass(error) } + } + + if (base.reachable) { + base = { ...base, evidence: await readEvidence(fetchImpl, url, options.token, timeoutMs) } + } + + return { ...base, ...verdictFor(base) } +} + +async function readEvidence( + fetchImpl: typeof fetch, + url: string, + token: string | undefined, + timeoutMs: number, +): Promise { + if (!token) { + return { + fetched: false, + reason: + 'no operator token supplied — pass --token or set FACTORY_EVIDENCE_TOKEN to read the ' + + 'free-text lastError at /evidence', + } + } + try { + const evidence = await getJson(fetchImpl, endpoint(url, '/evidence'), { token, timeoutMs }) + if (evidence.status !== 200) { + // Only 401/403 are statements about the credential. A 404 means this + // deployment exposes no /evidence route and a 5xx means the endpoint + // failed — sending someone to rotate a working token for either is the + // wrong-problem failure again (#300 review, CodeRabbit). + const reason = evidence.status === 401 || evidence.status === 403 + ? `/evidence answered HTTP ${evidence.status}; the token was not accepted` + : evidence.status === 404 + ? `/evidence answered HTTP 404; this deployment exposes no /evidence route (the token is not the problem)` + : `/evidence request failed with HTTP ${evidence.status}; the endpoint errored (the token is not the problem)` + return { fetched: false, httpStatus: evidence.status, reason } + } + const body = asRecord(evidence.body) + const readiness = asRecord(body.readinessReconcile) + return { + fetched: true, + httpStatus: evidence.status, + ...(asText(body.phase) ? { phase: asText(body.phase) } : {}), + ...(asText(readiness.lastError) ? { lastError: asText(readiness.lastError) } : {}), + ...(asCount(readiness.consecutiveFailures) !== undefined + ? { consecutiveFailures: asCount(readiness.consecutiveFailures) } + : {}), + } + } catch (error) { + return { fetched: false, reason: `/evidence request failed (${telemetryErrorClass(error)})` } + } +} + +/** Human-readable rendering; `--json` prints the diagnosis object instead. */ +export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): string { + const lines: string[] = [`factory diagnose — ${diagnosis.url}`] + lines.push( + ` reachable : ${diagnosis.reachable ? `yes (HTTP ${diagnosis.httpStatus ?? '?'})` : `no (${diagnosis.errorClass ?? 'no response'})`}`, + ) + + if (diagnosis.phase !== undefined) { + lines.push(` phase : ${diagnosis.phase}`) + } + if (diagnosis.live === false) { + lines.push(' instance liveness : NOT LIVE (the instance\'s own verdict, on its own clock)') + } + + const health = diagnosis.health + if (health) { + lines.push(` liveness (ok) : ${health.ok} (as of the last heartbeat write)`) + lines.push(` status : ${health.status}`) + lines.push( + ` loop : ${health.loopStatus ?? 'unknown'}, heartbeat ${formatDuration(health.ageMs)} old${health.stale ? ' (STALE)' : ''}`, + ) + if (health.degradedSubsystems.length > 0) { + lines.push(` degraded subsystems : ${health.degradedSubsystems.join(', ')}`) + } + const readiness = health.readinessReconcile + if (readiness) { + lines.push(' readinessReconcile:') + lines.push(` state : ${readiness.state}`) + lines.push( + ` consecutiveFailures: ${readiness.consecutiveFailures} (threshold ${readiness.failureThreshold})`, + ) + lines.push(` lastErrorClass : ${readiness.lastErrorClass ?? '—'}`) + lines.push(` cadence : ${formatDuration(readiness.intervalMs)}`) + if (readiness.inFlightMs !== undefined) { + lines.push( + ` pass in flight : ${formatDuration(readiness.inFlightMs)} (${readiness.missedPasses ?? 0} missed passes)`, + ) + } + lines.push(` lastStartedAt : ${formatInstant(readiness.lastStartedAtMs)}`) + lines.push(` lastCompletedAt : ${formatInstant(readiness.lastCompletedAtMs)}`) + lines.push(` lastFailureAt : ${formatInstant(readiness.lastFailureAtMs)}`) + } + lines.push(` eventListener : ${health.eventListener?.state ?? 'unknown'}`) + } else if (diagnosis.unreadable) { + lines.push(' health block : none — this response carried no Factory health') + } else if (diagnosis.workerOnly) { + // Not an old instance — an unprobed one. Saying "predates #295" here would + // send an operator to upgrade a Factory that is fine. + lines.push(' health block : not requested — the Worker answered without probing the container') + lines.push(' mode : event-driven short-sleep') + } else if (diagnosis.legacy?.readinessReconcile || diagnosis.legacy?.eventListener) { + lines.push(' health block : absent — state strings only (instance predates #295)') + lines.push(` readinessReconcile : ${diagnosis.legacy.readinessReconcile ?? 'unknown'}`) + lines.push(` eventListener : ${diagnosis.legacy.eventListener ?? 'unknown'}`) + } else { + // No block and no state strings: the response carried no Factory health at + // all. That is a statement about this response, not about the instance. + lines.push(' health block : absent, and no subsystem state strings in the response') + } + + const evidence = diagnosis.evidence + if (evidence) { + lines.push( + ` evidence : ${evidence.fetched ? `read (HTTP ${evidence.httpStatus ?? 200})` : `not read — ${evidence.reason ?? 'unavailable'}`}`, + ) + if (evidence.lastError) lines.push(` lastError : ${evidence.lastError}`) + } + + lines.push('') + lines.push(`verdict: ${diagnosis.verdict}`) + return `${lines.join('\n')}\n` +} + +const formatInstant = (ms: number | undefined): string => { + if (ms === undefined) return '—' + // Belt and braces with `normalizePublicHealth`'s range check: a renderer + // asked to explain an outage must never be the thing that throws. + const instant = new Date(ms) + return Number.isNaN(instant.getTime()) ? 'unknown' : instant.toISOString() +} diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 007c5af8..2bd4c5e9 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -8,6 +8,7 @@ import { ensureCloudSession, type CloudSession } from '@agent-relay/cloud' import { stringifyLogValue } from '../logging' import { resolveLocalFactoryConfig, type LocalClonePathOptions } from '../config/local-clone-paths' import { initializeFactory } from './init' +import { diagnoseDeployedFactory, renderDeployedDiagnosis } from './diagnose' import { FACTORY_EXIT, exitCodeForDispatchResult, @@ -159,6 +160,8 @@ export interface FleetCliDeps { cloudAccessTokenFetch?: typeof fetch /** Hermetic Cloud telemetry transport for CLI integration tests. */ cloudReporterFetch?: typeof fetch + /** Hermetic deployed-instance transport for `diagnose --deployed` tests. */ + diagnoseFetch?: typeof fetch isInteractive?: () => boolean confirmIntegrationConnect?: (provider: FactoryIntegrationProvider) => Promise openIntegrationUrl?: (url: string) => void | Promise @@ -203,6 +206,7 @@ type ParsedCommand = | { kind: 'factory-close-probe'; prNumber: number; repo: string; issue: string } | { kind: 'featuremap-check'; manifestPath?: string; baseRef?: string } | { kind: 'factory-init'; repo?: string; workspaceId?: string } + | { kind: 'factory-diagnose'; url: string; token?: string; json: boolean; timeoutMs?: number } | { kind: 'notion-intake'; manifestPath: string } | { kind: 'notion-manifest' @@ -232,7 +236,7 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom return 0 } const { globals, args } = parseGlobalOptions(argv) - const command = parseFleetCommand(args) + const command = parseFleetCommand(args, deps.env ?? process.env) if (command.kind === 'featuremap-check') { const report = await (deps.featureMapCheck ?? checkFeatureMap)({ @@ -243,6 +247,26 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom return 0 } + if (command.kind === 'factory-diagnose') { + // Deliberately ahead of every config/mount/fleet setup below: the whole + // point of #295 is that this runs from anywhere — a laptop, a lane, a + // runbook — against an instance this checkout knows nothing about. + const diagnosis = await diagnoseDeployedFactory({ + url: command.url, + ...(command.token ? { token: command.token } : {}), + ...(command.timeoutMs !== undefined ? { timeoutMs: command.timeoutMs } : {}), + ...(deps.diagnoseFetch ? { fetch: deps.diagnoseFetch } : {}), + }) + if (command.json) { + writeJson(out, diagnosis) + } else { + out.write(renderDeployedDiagnosis(diagnosis)) + } + // Same contract as `factory canary`: a lane that only reads $? must be + // able to tell "production is dispatching" from "it is not". + return diagnosis.dispatching ? FACTORY_EXIT.OK : FACTORY_EXIT.FAILED + } + if (command.kind === 'factory-init') { await initializeFactory({ repo: command.repo, workspaceId: command.workspaceId, stdout: out, stderr: err }) return 0 @@ -608,14 +632,14 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom } } -export function parseFleetCommand(args: string[]): ParsedCommand { +export function parseFleetCommand(args: string[], env: NodeJS.ProcessEnv = process.env): ParsedCommand { const [verb, ...rest] = args if (!verb) { throw new Error(usage()) } if (isFactoryAction(verb)) { - return parseFactoryCommand(args) + return parseFactoryCommand(args, env) } if (verb === 'fleet') { @@ -1436,7 +1460,7 @@ function resolveLocalMountFn( } } -function parseFactoryCommand(args: string[]): ParsedCommand { +function parseFactoryCommand(args: string[], env: NodeJS.ProcessEnv = process.env): ParsedCommand { const [action, issueOrPr, ...flags] = args if (action === 'init') { const values = [issueOrPr, ...flags].filter((value): value is string => Boolean(value)) @@ -1454,6 +1478,12 @@ function parseFactoryCommand(args: string[]): ParsedCommand { } return { kind: 'factory-init', repo, workspaceId } } + if (action === 'diagnose') { + return parseFactoryDiagnoseFlags( + [issueOrPr, ...flags].filter((value): value is string => Boolean(value)), + env, + ) + } if (action === 'start') { return { kind: 'factory', action, ...parseFactoryStartFlags([issueOrPr, ...flags]) } } @@ -1535,6 +1565,65 @@ function evaluateFactoryCanary( } } +function parseFactoryDiagnoseFlags(flags: string[], env: NodeJS.ProcessEnv): ParsedCommand { + let url: string | undefined + // The injected environment, not the process one: an embedder or a hermetic + // test that supplies `deps.env` must not silently skip the authenticated + // read (#300 review, P2, cubic). + let token = env.FACTORY_EVIDENCE_TOKEN?.trim() || undefined + let json = false + let timeoutMs: number | undefined + for (let index = 0; index < flags.length; index += 1) { + const flag = flags[index] + if (flag === '--deployed' || flag === '--url') { + url = requireValue(flags, ++index, flag) + continue + } + if (flag === '--token') { + token = requireValue(flags, ++index, '--token') + continue + } + if (flag === '--timeout-ms') { + const value = Number(requireValue(flags, ++index, '--timeout-ms')) + if (!Number.isInteger(value) || value <= 0) throw new Error('--timeout-ms requires a positive integer') + timeoutMs = value + continue + } + if (flag === '--json') { + json = true + continue + } + // A bare URL is the same request spelled shorter. + if (!flag.startsWith('-') && !url) { + url = flag + continue + } + // Never echo the value: `factory diagnose ` is a plausible + // slip, and this message reaches stderr and from there CI logs (#300 + // review, CodeRabbit). The position is enough to find it. + if (!flag.startsWith('-')) { + throw new Error( + `factory diagnose accepts one url; argument ${index + 1} is a second positional value. ` + + 'Pass a token with --token, not as a bare argument.', + ) + } + throw new Error(`Unknown factory diagnose option at argument ${index + 1}`) + } + if (!url) { + throw new Error('factory diagnose requires --deployed (the deployed instance base URL)') + } + if (!/^https?:\/\//u.test(url)) { + // Not echoed, for the same reason as the positional case above: the value + // in the url slot is exactly where a mistyped `--token` argument lands + // (#300 review, cubic). + throw new Error( + 'factory diagnose --deployed requires an http(s) url; the value given does not start with ' + + 'http:// or https://. If that value is a token, pass it with --token.', + ) + } + return { kind: 'factory-diagnose', url, json, ...(token ? { token } : {}), ...(timeoutMs !== undefined ? { timeoutMs } : {}) } +} + function parseFactoryStartFlags(args: Array): { mode: 'live' } { let mode: 'live' = 'live' const flags = args.filter((arg): arg is string => Boolean(arg)) @@ -2556,6 +2645,7 @@ function isCapability(value: string | undefined): value is Capability { function isFactoryAction(value: string): boolean { return value === 'init' || + value === 'diagnose' || value === 'start' || value === 'run-once' || value === 'loop' || @@ -2638,6 +2728,9 @@ Commands: status Print current factory status as JSON loop Run the bounded loop configured in factory.config.json loop-status Print heartbeat/liveness status for the loop + diagnose --deployed + Ask a DEPLOYED instance why it is or is not dispatching + (--token/FACTORY_EVIDENCE_TOKEN also reads /evidence) kill-loop Send SIGTERM to the heartbeat pid reap-orphans [--include-held] Report stale processes and held agents; opt in to release held agents past deadline diff --git a/src/hosted/orchestrator.ts b/src/hosted/orchestrator.ts index c30a5c1f..aa89ad58 100644 --- a/src/hosted/orchestrator.ts +++ b/src/hosted/orchestrator.ts @@ -22,6 +22,7 @@ import type { HostedFactoryWritebackRecord, } from './types' import { dispatchAgentIdentityKey } from '../dispatch/work-unit-identity' +import { telemetryErrorClass } from '../observability/error-class.js' const DEFAULT_LEASE_TTL_MS = 5 * 60_000 const DEFAULT_MAX_ISSUES_PER_RUN = 100 @@ -771,11 +772,6 @@ function hostedInvocationReleaseReason( return undefined } -function telemetryErrorClass(error: unknown): string { - const name = error instanceof Error ? error.name : '' - return /^[A-Za-z][A-Za-z0-9]{0,63}(?:Error|Exception)$/u.test(name) ? name : 'Error' -} - function dedupeIssues(issues: T[]): T[] { return [...new Map(issues.map((issue) => [issue.uuid, issue])).values()] } diff --git a/src/index.ts b/src/index.ts index 69b3f061..2f7e931d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -148,6 +148,13 @@ export { BatchTracker, DEFAULT_FACTORY_LOOP_HEARTBEAT_PATH, DEFAULT_FACTORY_LOOP_REGISTRY_PATH, + DEFAULT_PUBLIC_HEALTH_STALE_MS, + DEFAULT_READINESS_RECONCILE_INTERVAL_MS, + FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + READINESS_RECONCILE_STALL_INTERVALS, + derivedReadinessReconcileState, + publicHealthFromHeartbeat, + readinessReconcileInFlightMs, FactoryEnvironmentReaper, FactoryReaper, checkFactoryLoopLiveness, diff --git a/src/observability/error-class.ts b/src/observability/error-class.ts new file mode 100644 index 00000000..187c7145 --- /dev/null +++ b/src/observability/error-class.ts @@ -0,0 +1,39 @@ +/** + * The one allowlist for error identifiers that cross a telemetry, operator or + * public boundary. + * + * An error's `message` is dependency-controlled free text: it routinely + * carries provider prose, filesystem paths, URLs and query strings that may + * embed credentials. Its *class name* is code-controlled and carries none of + * that — but only if the reader refuses anything that does not look like a + * class name, because `name` is a writable property and a persisted record can + * be written by an older or hostile producer. + * + * Anything that fails the pattern collapses to `Error`, which still tells a + * reader "this failed" without letting the failure choose what gets published. + */ +export const TELEMETRY_ERROR_CLASS_PATTERN = /^[A-Za-z][A-Za-z0-9]{0,63}(?:Error|Exception)$/u + +/** What an unrecognised class name collapses to. */ +export const TELEMETRY_ERROR_CLASS_FALLBACK = 'Error' + +/** True when `value` is a class name the allowlist admits verbatim. */ +export function isTelemetryErrorClassName(value: unknown): value is string { + return typeof value === 'string' && TELEMETRY_ERROR_CLASS_PATTERN.test(value) +} + +/** The allowlisted class name of a thrown value. */ +export function telemetryErrorClass(error: unknown): string { + const name = error instanceof Error ? error.name : '' + return isTelemetryErrorClassName(name) ? name : TELEMETRY_ERROR_CLASS_FALLBACK +} + +/** + * The allowlisted class name of an already-persisted record's class field. + * + * Used when re-publishing a stored status: the producer wrote a string, and + * this side has to decide whether that string may cross the boundary. + */ +export function telemetryErrorClassName(value: unknown): string { + return isTelemetryErrorClassName(value) ? value : TELEMETRY_ERROR_CLASS_FALLBACK +} diff --git a/src/observability/index.ts b/src/observability/index.ts index bcb707ea..60933a57 100644 --- a/src/observability/index.ts +++ b/src/observability/index.ts @@ -1,3 +1,4 @@ +export * from './error-class.js' export * from './cloud-reporter.js' export * from './events.js' export * from './instance-identity.js' diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index f079ab20..8edb7837 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -11788,6 +11788,251 @@ describe('FactoryLoop', () => { } }) + + // #295. The deployed instance publishes exactly one unauthenticated record, + // and it is built from this file. These two tests are the pair the outage + // needed: the failing case (a counter and a class an operator can read + // without a credential) and the silent case (a pass that hangs and writes + // no state at all). + describe('operator-reachable diagnostics on the public health block (#295)', () => { + it('publishes the failure count and error class without the message that names a path', async () => { + class DiagnosableFailureStateStore extends InMemoryStateStore { + failClaims = false + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + if (this.failClaims) { + throw Object.assign( + new Error( + 'ENOENT: no such file or directory, open ' + + "'/srv/agent-workforce/.relay/workspace-key' while POSTing " + + 'https://relay.internal.example.com/v1/workspaces/ws_9f2?token=sk-live-abc', + ), + { name: 'DiscoveryClaimError' }, + ) + } + return await super.claimDiscoverySweep(workspaceId, owner, nowMs, leaseMs) + } + } + + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new DiagnosableFailureStateStore({ batchSize: 2 }) + const root = await mkdtemp(join(tmpdir(), 'factory-public-health-')) + const heartbeatPath = join(root, 'heartbeat.json') + const factory = createFactory(config({ + issueSource: 'github', + loop: { heartbeatPath, registryPath: join(root, 'registry.json') }, + }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + try { + stateStore.failClaims = true + await vi.waitFor(async () => { + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + expect(heartbeat?.health?.readinessReconcile?.consecutiveFailures).toBeGreaterThanOrEqual(3) + }, { timeout: 3_000 }) + + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + expect(heartbeat?.health).toMatchObject({ + // Liveness is unchanged: the process is up, and this is what the + // container ping endpoint reads. + ok: true, + // ...but the record can now go amber, which is what was missing. + status: 'degraded', + degradedSubsystems: ['readinessReconcile'], + readinessReconcile: { + state: 'degraded', + failureThreshold: 3, + lastErrorClass: 'DiscoveryClaimError', + }, + }) + + // MUST-NOT-FIRE: the authenticated surface keeps the message, the + // public block never sees it. + expect(heartbeat?.readinessReconcile?.lastError).toContain('/srv/agent-workforce') + const published = JSON.stringify(heartbeat?.health) + expect(published).not.toContain('/srv/agent-workforce') + expect(published).not.toContain('https://') + expect(published).not.toContain('sk-live-abc') + expect(published).not.toContain('ENOENT') + } finally { + stateStore.failClaims = false + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + // Review follow-up on #300 (P1, cubic). The startup backfill is the pass + // most likely to hang — #36 measured 61 minutes here on a cold container — + // and it recorded no timestamps at all, so a wedged FIRST pass left the + // derived state reading `healthy` with nothing to derive from. + it('marks a hung startup backfill as an in-flight pass', async () => { + class HangingBackfillMount extends CountingEventsMount { + readonly backfillStarted: Promise + #resolveBackfillStarted: () => void = () => undefined + #releaseBackfill: () => void = () => undefined + #hung = false + + constructor() { + super() + this.backfillStarted = new Promise((resolve) => { this.#resolveBackfillStarted = resolve }) + this.setSubRoot('/linear/issues', 'absent') + } + + releaseBackfill(): void { + this.#releaseBackfill() + } + + override async listTree(prefix: string): Promise { + if (!this.#hung) { + this.#hung = true + this.#resolveBackfillStarted() + await new Promise((resolve) => { this.#releaseBackfill = resolve }) + } + return super.listTree(prefix) + } + } + + const mount = new HangingBackfillMount() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + const started = factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 20 }, + }) + try { + await mount.backfillStarted + + // 10 missed passes at a 20ms cadence is 200ms of hang. + await vi.waitFor(() => { + const readiness = factory.status().readinessReconcile + expect(readiness?.state).toBe('stalled') + expect(readiness?.lastCompletedAtMs).toBeUndefined() + expect(readiness?.inFlightMs ?? 0).toBeGreaterThan(200) + }, { timeout: 3_000 }) + + // No completed pass on record is how an operator tells a cold boot + // still hydrating from a daemon that ran fine for hours and stopped. + expect(factory.status().readinessReconcile?.consecutiveFailures).toBe(0) + } finally { + mount.releaseBackfill() + await started + await factory.stop() + } + }) + + it('reports a hung sweep as stalled while every settled field still reads healthy', async () => { + class HangingPeriodicMount extends CountingEventsMount { + readonly periodicStarted: Promise + hangPeriodic = false + #resolvePeriodicStarted: () => void = () => undefined + #releasePeriodic: () => void = () => undefined + #hung = false + + constructor() { + super() + this.periodicStarted = new Promise((resolve) => { this.#resolvePeriodicStarted = resolve }) + this.setSubRoot('/linear/issues', 'absent') + } + + releasePeriodic(): void { + this.#releasePeriodic() + } + + override async listTree(prefix: string): Promise { + if (this.hangPeriodic && !this.#hung) { + this.#hung = true + this.#resolvePeriodicStarted() + await new Promise((resolve) => { this.#releasePeriodic = resolve }) + } + return super.listTree(prefix) + } + } + + const mount = new HangingPeriodicMount() + const root = await mkdtemp(join(tmpdir(), 'factory-stalled-health-')) + const heartbeatPath = join(root, 'heartbeat.json') + const factory = createFactory(config({ + issueSource: 'github', + loop: { + heartbeatPath, + registryPath: join(root, 'registry.json'), + // Keep the heartbeat refreshing while the sweep is wedged: the + // deployed instance ticks every ~30s throughout a hang, which is + // exactly why the hang looked green. + heartbeatStaleMs: 1_000, + }, + }), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + // 10 missed passes at a 20ms cadence is 200ms of hang. + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 20 }, + }) + try { + mount.hangPeriodic = true + await mount.periodicStarted + + await vi.waitFor(() => { + const readiness = factory.status().readinessReconcile + expect(readiness?.state).toBe('stalled') + expect(readiness?.inFlightMs ?? 0).toBeGreaterThan(200) + }, { timeout: 3_000 }) + + const readiness = factory.status().readinessReconcile + // The fields the old surface exposed all still read green — that is + // the point. Nothing wrote them, because a hang settles neither way. + expect(readiness?.consecutiveFailures).toBe(0) + expect(readiness?.lastError).toBeUndefined() + expect(readiness?.lastFailureAtMs).toBeUndefined() + // The relative order of the two timestamps is the whole signal. + expect(readiness?.lastStartedAtMs).toBeGreaterThan(readiness?.lastCompletedAtMs ?? 0) + + await vi.waitFor(async () => { + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + expect(heartbeat?.health).toMatchObject({ + ok: true, + status: 'degraded', + degradedSubsystems: ['readinessReconcile'], + readinessReconcile: { state: 'stalled', consecutiveFailures: 0 }, + }) + expect(heartbeat?.health?.readinessReconcile?.missedPasses ?? 0).toBeGreaterThanOrEqual(10) + }, { timeout: 3_000 }) + } finally { + mount.releasePeriodic() + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + }) + it('derives a repo-scoped subscription and startup backfill from the simple hoopsheet config', async () => { class ScopedStartupMount extends CountingEventsMount { readonly listTreePrefixes: string[] = [] diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index cd3d724f..1cb2dc1a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -126,6 +126,12 @@ import { type FactoryCloudCancellationReasonV1, type FactoryCloudEventInputV1, } from '../observability/events' +import { telemetryErrorClass } from '../observability/error-class' +import { + derivedReadinessReconcileState, + publicHealthFromHeartbeat, + readinessReconcileInFlightMs, +} from './public-health' import { boundedRunCostTotal, CostLedger, type RunCostTotal, type UnpricedModelCostRecord } from '../cost/ledger' import { createTicketDispatchDelivery, type TicketDispatchDelivery } from '../delivery/ticket-dispatch' import { @@ -713,6 +719,7 @@ export class FactoryLoop implements Factory { #readinessReconcileLastCompletedAtMs?: number #readinessReconcileLastFailureAtMs?: number #readinessReconcileLastError?: string + #readinessReconcileLastErrorClass?: string readonly #liveEventQueue: ChangeEvent[] = [] #liveEventDrainScheduled = false #liveEventDrainActive = false @@ -1433,9 +1440,26 @@ export class FactoryLoop implements Factory { this.#logger.info?.('[factory] running startup ready-issue backfill before draining buffered events', { highWatermarkRouteUnavailable: highWatermark.routeUnavailable, }) + // Review follow-up on #300 (P1, cubic). The startup backfill is a + // discovery pass like any other, and it is the one most likely to hang: + // #36 measured 61 minutes here while the Relayfile mirror hydrated on a + // cold container. Stamping it means a wedged FIRST pass is visible as + // in-flight, instead of leaving the timestamps empty and the derived + // state reading `healthy` forever. + // + // Only the timestamps. `consecutiveFailures` and `lastError` belong to + // the reconcile loop's own failure accounting, which owns the degraded + // threshold and the #297 reason allowlist; a startup failure is already + // counted by `liveStartupBackfillErrors` and reported through `#error`. + const backfillStartedAtMs = this.#clock.now() + this.#readinessReconcileLastStartedAtMs = backfillStartedAtMs try { await this.runOnce() + this.#readinessReconcileLastDurationMs = this.#elapsedSince(backfillStartedAtMs) + this.#readinessReconcileLastCompletedAtMs = this.#clock.now() } catch (error) { + this.#readinessReconcileLastDurationMs = this.#elapsedSince(backfillStartedAtMs) + this.#readinessReconcileLastFailureAtMs = this.#clock.now() // A startup backfill failure must not abort the daemon: log it and fall // back to the live event stream (plus any buffered events) instead of // leaving the factory down. @@ -1599,6 +1623,7 @@ export class FactoryLoop implements Factory { this.#readinessReconcileLastDurationMs = this.#elapsedSince(startedAtMs) this.#readinessReconcileLastCompletedAtMs = this.#clock.now() this.#readinessReconcileLastError = undefined + this.#readinessReconcileLastErrorClass = undefined this.#logger.info?.('[factory] periodic readiness reconciliation completed', { durationMs: this.#readinessReconcileLastDurationMs, candidates: report.pulled.length, @@ -1625,6 +1650,9 @@ export class FactoryLoop implements Factory { this.#readinessReconcileLastDurationMs = this.#elapsedSince(startedAtMs) this.#readinessReconcileLastFailureAtMs = this.#clock.now() this.#readinessReconcileLastError = errorMessage + // The class, unlike the message, is publishable: #295 puts it on the + // unauthenticated health surface through the same allowlist. + this.#readinessReconcileLastErrorClass = telemetryErrorClass(error) this.#increment('readinessReconcileErrors') this.#logger.warn?.('[factory] periodic readiness reconciliation failed; retry remains scheduled', { error: errorMessage, @@ -4581,17 +4609,54 @@ export class FactoryLoop implements Factory { #readinessReconcileStatus(): FactoryReadinessReconcileStatus { const consecutiveFailures = this.#readinessReconcileConsecutiveFailures - const state = this.#startMode !== 'live' + const settled: FactoryReadinessReconcileStatus['state'] = this.#startMode !== 'live' ? 'not-running' : consecutiveFailures >= READINESS_RECONCILE_FAILURE_THRESHOLD ? 'degraded' : consecutiveFailures > 0 ? 'retrying' : 'healthy' + // The counters above only move when a pass *settles*. A pass that hangs + // takes neither path, so `settled` would keep reporting the last finished + // pass — `healthy` — for as long as the process is stuck (#295). The + // in-flight age is the only field that can express that, so derive the + // state from it rather than trusting the last write. + const timestamps = { + intervalMs: this.#readinessReconcileIntervalMs, + ...(this.#readinessReconcileLastStartedAtMs !== undefined + ? { lastStartedAtMs: this.#readinessReconcileLastStartedAtMs } + : {}), + ...(this.#readinessReconcileLastCompletedAtMs !== undefined + ? { lastCompletedAtMs: this.#readinessReconcileLastCompletedAtMs } + : {}), + ...(this.#readinessReconcileLastFailureAtMs !== undefined + ? { lastFailureAtMs: this.#readinessReconcileLastFailureAtMs } + : {}), + } + const nowMs = this.#clock.now() + // Defence in depth (#300 review, CodeRabbit). These derivations are new + // code from another module on a path that `status()` and every heartbeat + // write depend on. A throw here would take out the liveness signal the + // crash reaper reads — the diagnostic causing the outage it exists to + // explain — so a failure costs the derived fields and nothing else. + let inFlightMs: number | undefined + let derived: FactoryReadinessReconcileStatus['state'] | 'unknown' = settled + try { + inFlightMs = readinessReconcileInFlightMs(timestamps, nowMs) + derived = derivedReadinessReconcileState({ ...timestamps, state: settled }, nowMs) + } catch (error) { + this.#logger.warn?.('[factory] readiness health derivation failed; reporting the settled state', { + error: describeError(error).errorMessage, + }) + inFlightMs = undefined + derived = settled + } return { - state, + state: derived === 'unknown' ? settled : derived, consecutiveFailures, failureThreshold: READINESS_RECONCILE_FAILURE_THRESHOLD, + intervalMs: this.#readinessReconcileIntervalMs, + ...(inFlightMs !== undefined ? { inFlightMs } : {}), ...(this.#readinessReconcileLastDurationMs !== undefined ? { lastDurationMs: this.#readinessReconcileLastDurationMs } : {}), @@ -4605,6 +4670,9 @@ export class FactoryLoop implements Factory { ? { lastFailureAtMs: this.#readinessReconcileLastFailureAtMs } : {}), ...(this.#readinessReconcileLastError ? { lastError: this.#readinessReconcileLastError } : {}), + ...(this.#readinessReconcileLastErrorClass + ? { lastErrorClass: this.#readinessReconcileLastErrorClass } + : {}), } } @@ -7293,6 +7361,28 @@ export class FactoryLoop implements Factory { readinessReconcile: this.#readinessReconcileStatus(), fleetControlPlane: this.#fleetControlPlane.status(), } + // The deployed container serves `/healthz` straight out of this file and + // has no redaction logic of its own, so publish the already-safe view here + // rather than leaving that boundary to whoever reads the file (#295). + // Derived against this daemon's clock: every duration in it is a + // difference between timestamps this process wrote. + // + // Guarded (#300 review, CodeRabbit): this heartbeat is what the crash + // reaper and `/healthz` read to decide the daemon is alive, and several + // callers of this method sit outside any try/catch. A projection failure + // must cost the diagnostics block, never the heartbeat — the omitted block + // is itself legible, since `factory diagnose` reports a missing one rather + // than a false green. + try { + heartbeat.health = publicHealthFromHeartbeat(heartbeat, { + nowMs: updatedAtMs, + staleMs: this.#config.loop.heartbeatStaleMs, + }) + } catch (error) { + this.#logger.warn?.('[factory] public health projection failed; heartbeat written without it', { + error: describeError(error).errorMessage, + }) + } await mkdir(dirname(path), { recursive: true }) await writeFile(path, `${JSON.stringify(heartbeat, null, 2)}\n`, 'utf8') await this.#writeInFlightRegistry(registryPath, path) @@ -19310,11 +19400,6 @@ const telemetryCategory = (value: string | undefined): string | undefined => { return normalized.slice(0, 120) || undefined } -const telemetryErrorClass = (error: unknown): string => { - const name = error instanceof Error ? error.name : '' - return /^[A-Za-z][A-Za-z0-9]{0,63}(?:Error|Exception)$/u.test(name) ? name : 'Error' -} - const isTimeoutError = (error: unknown): boolean => error instanceof Error && (error.name === 'TimeoutError' || error.name === 'AbortError') diff --git a/src/orchestrator/health-projection-guard.test.ts b/src/orchestrator/health-projection-guard.test.ts new file mode 100644 index 00000000..e6890337 --- /dev/null +++ b/src/orchestrator/health-projection-guard.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it, vi } from 'vitest' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +/** + * Review follow-up on #300 (Major, CodeRabbit). + * + * The health projection is new code on the heartbeat write path, and the + * heartbeat is what the crash reaper and `/healthz` read to decide whether the + * daemon is alive. If the projection ever throws, an unguarded call would fail + * every heartbeat write and make a healthy daemon look wedged — the diagnostic + * causing the outage it exists to explain. + * + * Nothing in `publicHealthFromHeartbeat` throws today; this pins the guard, not + * the absence of a throw. The module is mocked to throw, which is the only + * honest way to test a defence against a condition the code does not currently + * produce. + */ +vi.mock('./public-health', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + publicHealthFromHeartbeat: () => { + throw new TypeError('projection exploded') + }, + derivedReadinessReconcileState: () => { + throw new TypeError('derivation exploded') + }, + } +}) + +const { createFactory, readFactoryLoopHeartbeat } = await import('./factory') +const { FakeFleetClient, FakeMountClient } = await import('../testing') +const { FactoryConfigSchema } = await import('../config/schema') + +describe('health projection failures never break the heartbeat (#295)', () => { + it('still writes a heartbeat, and still reports status, when the projection throws', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-health-guard-')) + const heartbeatPath = join(root, 'heartbeat.json') + const mount = new FakeMountClient() + mount.setSubRoot('/linear/issues', 'absent') + const logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() } + const factory = createFactory( + FactoryConfigSchema.parse({ + workspaceId: 'factory-health-guard', + issueSource: 'github', + repos: { + byLabel: { pear: 'AgentWorkforce/pear' }, + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + stateIds: { + readyForAgent: 'state-ready-for-agent', + agentImplementing: 'state-agent-implementing', + done: 'state-done', + inPlanning: 'state-in-planning', + humanReview: 'state-human-review', + }, + loop: { heartbeatPath, registryPath: join(root, 'registry.json'), heartbeatStaleMs: 1_000 }, + }), + { mount, fleet: new FakeFleetClient(), logger }, + ) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + try { + // The liveness contract the reaper depends on: the file exists and its + // timestamp advances, projection or no projection. + await vi.waitFor(async () => { + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + expect(heartbeat?.pid).toBe(process.pid) + expect(heartbeat?.updatedAtMs).toEqual(expect.any(Number)) + }, { timeout: 3_000 }) + + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + // The block is omitted rather than half-written... + expect(heartbeat?.health).toBeUndefined() + // ...and the authenticated detail an operator reads is untouched. + expect(heartbeat?.readinessReconcile?.state).toEqual(expect.any(String)) + + // status() must survive the same failure: a CLI asking a wedged daemon + // what is wrong should not get an exception instead of an answer. + expect(() => factory.status()).not.toThrow() + expect(factory.status().readinessReconcile?.state).toEqual(expect.any(String)) + + // The failure is not swallowed silently — it is logged where an operator + // reading logs will find it. + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('health'), + expect.anything(), + ) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) +}) diff --git a/src/orchestrator/index.ts b/src/orchestrator/index.ts index efe4178b..982532f1 100644 --- a/src/orchestrator/index.ts +++ b/src/orchestrator/index.ts @@ -1,4 +1,13 @@ export { BatchTracker, issueKey } from './batch-tracker' +export { + DEFAULT_PUBLIC_HEALTH_STALE_MS, + DEFAULT_READINESS_RECONCILE_INTERVAL_MS, + FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + READINESS_RECONCILE_STALL_INTERVALS, + derivedReadinessReconcileState, + publicHealthFromHeartbeat, + readinessReconcileInFlightMs, +} from './public-health' export type { DependencyAdmission, DependencyBlocker, ParkedIssue } from './batch-tracker' export { dependencyIdentity, findDependencyCycle, parseBlockedBy, resolveDependency } from './dependencies' export type { DeclaredDependency, ResolvedDependency } from './dependencies' diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts new file mode 100644 index 00000000..36645b24 --- /dev/null +++ b/src/orchestrator/public-health.test.ts @@ -0,0 +1,437 @@ +import { describe, expect, it } from 'vitest' + +import { + FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + READINESS_RECONCILE_STALL_INTERVALS, + normalizePublicHealth, + publicHealthFromHeartbeat, +} from './public-health' +import type { FactoryLoopHeartbeat } from '../types' + +const BOOT_MS = 1_787_224_000_000 + +function heartbeat(overrides: Partial = {}): FactoryLoopHeartbeat { + return { + pid: 42, + status: 'running', + iteration: 0, + maxIterations: 0, + updatedAt: new Date(BOOT_MS).toISOString(), + updatedAtMs: BOOT_MS, + eventListener: { state: 'subscribed' }, + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: BOOT_MS - 30_000, + lastCompletedAtMs: BOOT_MS - 29_000, + lastDurationMs: 1_000, + }, + ...overrides, + } +} + +describe('publicHealthFromHeartbeat (#295)', () => { + it('carries the failure count and an allowlisted error class', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'degraded', + consecutiveFailures: 8, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: BOOT_MS - 30_000, + lastCompletedAtMs: BOOT_MS - 600_000, + lastFailureAtMs: BOOT_MS - 29_000, + lastError: 'Refusing to dispatch AR-241: dispatch lifecycle is already terminal', + lastErrorClass: 'TypeError', + }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.schemaVersion).toBe(FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION) + expect(health.readinessReconcile).toMatchObject({ + state: 'degraded', + consecutiveFailures: 8, + failureThreshold: 3, + lastErrorClass: 'TypeError', + }) + expect(health.status).toBe('degraded') + expect(health.degradedSubsystems).toEqual(['readinessReconcile']) + }) + + // MUST-NOT-FIRE. `lastError` is a free-text, dependency-controlled string + // that already carries provider text, filesystem paths and URLs. It is + // readable at /evidence behind a bearer token; nothing derived from it may + // reach the unauthenticated health surface except the allowlisted class. + it('keeps provider text, filesystem paths, URLs and tokens off the public surface', () => { + const hostile = + 'ENOENT: no such file or directory, open ' + + "'/srv/agent-workforce/.relay/workspace-key' while POSTing " + + 'https://relay.internal.example.com/v1/workspaces/ws_9f2?token=sk-live-abcdef0123456789' + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'degraded', + consecutiveFailures: 7, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: BOOT_MS - 30_000, + lastFailureAtMs: BOOT_MS - 29_000, + lastError: hostile, + // A writer that puts free text where the class belongs must not be + // able to smuggle it through either. + lastErrorClass: hostile, + }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + const rendered = JSON.stringify(health) + expect(rendered).not.toContain('/srv/agent-workforce') + expect(rendered).not.toContain('.relay/workspace-key') + expect(rendered).not.toContain('https://') + expect(rendered).not.toContain('relay.internal.example.com') + expect(rendered).not.toContain('sk-live-abcdef0123456789') + expect(rendered).not.toContain('ENOENT') + // The operator still learns that the subsystem is failing and how often. + expect(health.readinessReconcile?.consecutiveFailures).toBe(7) + expect(health.readinessReconcile?.lastErrorClass).toBe('Error') + }) + + it('drops the free-text reason from the event-listener state', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + eventListener: { state: 'not-listening', reason: 'mount /srv/agent-workforce is unavailable' }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.eventListener).toEqual({ state: 'not-listening' }) + expect(JSON.stringify(health)).not.toContain('/srv/agent-workforce') + expect(health.degradedSubsystems).toContain('eventListener') + }) + + // The observed 2026-08-20 case: every state string reads green while the + // sweep that started at 11:16:35Z has neither completed nor failed. The + // relative order of the two timestamps is the entire signal. + it('derives stalled from lastStarted > lastCompleted past the stall threshold', () => { + const startedAtMs = BOOT_MS - 77 * 60_000 + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: startedAtMs, + lastCompletedAtMs: startedAtMs - 60_003, + }, + }), + { nowMs: BOOT_MS }, + ) + + expect(health.readinessReconcile).toMatchObject({ + state: 'stalled', + inFlightMs: 77 * 60_000, + missedPasses: 77, + }) + expect(health.status).toBe('degraded') + expect(health.degradedSubsystems).toEqual(['readinessReconcile']) + }) + + it('does not call a pass in flight for less than the stall threshold stalled', () => { + const startedAtMs = BOOT_MS - (READINESS_RECONCILE_STALL_INTERVALS - 1) * 60_000 + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: startedAtMs, + lastCompletedAtMs: startedAtMs - 1_000, + }, + }), + { nowMs: BOOT_MS }, + ) + + expect(health.readinessReconcile?.state).toBe('healthy') + expect(health.readinessReconcile?.inFlightMs).toBe((READINESS_RECONCILE_STALL_INTERVALS - 1) * 60_000) + expect(health.status).toBe('ok') + }) + + it('reports no in-flight pass when the last pass completed after it started', () => { + const health = publicHealthFromHeartbeat(heartbeat(), { nowMs: BOOT_MS }) + + expect(health.readinessReconcile?.inFlightMs).toBeUndefined() + expect(health.readinessReconcile?.state).toBe('healthy') + expect(health.ok).toBe(true) + expect(health.status).toBe('ok') + }) + + // Deliverable (2). `ok` is the container ping verdict, and a 503 there + // recycles the container — which destroys the evidence and restarts the + // cold-start hydration. The amber goes in `status`, which no platform + // interprets, so a monitor can alert on it without causing a restart loop. + it('keeps ok true for a live process while status goes amber', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'degraded', + consecutiveFailures: 8, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: BOOT_MS - 30_000, + lastFailureAtMs: BOOT_MS - 29_000, + }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.ok).toBe(true) + expect(health.status).toBe('degraded') + }) + + it('reports a missing heartbeat as unknown rather than healthy', () => { + const health = publicHealthFromHeartbeat(undefined, { nowMs: BOOT_MS }) + + expect(health).toMatchObject({ ok: false, status: 'unknown', stale: true }) + expect(health.readinessReconcile).toBeUndefined() + }) + + it('reports a stale heartbeat as not ok', () => { + const health = publicHealthFromHeartbeat(heartbeat(), { nowMs: BOOT_MS + 120_000, staleMs: 60_000 }) + + expect(health).toMatchObject({ ok: false, status: 'unknown', stale: true, ageMs: 120_000 }) + }) + + it('coerces hostile non-numeric counters instead of passing them through', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: '/srv/agent-workforce' as never, + consecutiveFailures: '7; DROP TABLE' as never, + failureThreshold: Number.NaN, + lastStartedAtMs: 'https://example.com' as never, + }, + }), + { nowMs: BOOT_MS }, + ) + + const rendered = JSON.stringify(health) + expect(rendered).not.toContain('/srv/agent-workforce') + expect(rendered).not.toContain('DROP TABLE') + expect(rendered).not.toContain('https://') + expect(health.readinessReconcile).toMatchObject({ state: 'unknown', consecutiveFailures: 0 }) + }) + // Review follow-up on #300 (P2, codex). `starting` is the state a live + // daemon reports before `#startLiveSubscription` installs the subscription: + // no listener is registered, so reporting green would be the same false + // green this issue exists to remove. Startup can be lengthy. + it('treats a listener that is still starting as not yet dispatch-capable', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ eventListener: { state: 'starting' } }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.degradedSubsystems).toContain('eventListener') + expect(health.status).toBe('degraded') + // Still alive — this is amber during startup, not a dead process. + expect(health.ok).toBe(true) + }) + + it('does not fault the listener on an instance that is not running live', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + eventListener: { state: 'not-listening', reason: 'factory mode is dispatch-owner' }, + readinessReconcile: { state: 'not-running', consecutiveFailures: 0, failureThreshold: 3 }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + // A bounded `factory loop` is not supposed to be listening; only a live + // daemon's silence is a fault. + expect(health.degradedSubsystems).toEqual([]) + expect(health.status).toBe('ok') + }) + + // Review follow-up on #300 (P2, codex). A finite number is not a valid date: + // `new Date(1e300).toISOString()` throws, and a remote record reaches a + // renderer that would abort the whole diagnosis. + it('drops timestamps outside the representable Date range', () => { + const health = normalizePublicHealth({ + schemaVersion: 1, + ok: true, + status: 'ok', + stale: false, + updatedAtMs: 1e300, + loopStatus: 'running', + degradedSubsystems: [], + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + lastStartedAtMs: 1e300, + lastCompletedAtMs: -1e300, + lastFailureAtMs: Number.MAX_VALUE, + intervalMs: 60_000, + }, + }) + + expect(health?.updatedAtMs).toBeUndefined() + expect(health?.readinessReconcile?.lastStartedAtMs).toBeUndefined() + expect(health?.readinessReconcile?.lastCompletedAtMs).toBeUndefined() + expect(health?.readinessReconcile?.lastFailureAtMs).toBeUndefined() + // Durations are not dates and stay as they are. + expect(health?.readinessReconcile?.intervalMs).toBe(60_000) + }) + + it('drops an out-of-range timestamp written into the heartbeat itself', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + lastStartedAtMs: 1e300, + }, + }), + { nowMs: BOOT_MS }, + ) + + expect(health.readinessReconcile?.lastStartedAtMs).toBeUndefined() + expect(health.readinessReconcile?.inFlightMs).toBeUndefined() + }) + // Review follow-up on #300 (P1, cubic). An open fleet control-plane circuit + // rejects every spawn and resume, so dispatch is gated just as hard as by a + // failing readiness sweep — and the health record said nothing about it. + it('reports an open fleet control-plane circuit as dispatch-gating', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + fleetControlPlane: { + state: 'open', + consecutiveFailures: 4, + timeoutMs: 10_000, + failureThreshold: 3, + resetTimeoutMs: 30_000, + lastFailureAtMs: BOOT_MS - 5_000, + retryAtMs: BOOT_MS + 25_000, + lastError: 'roster probe failed: connect ECONNREFUSED /run/relay/broker.sock', + }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.degradedSubsystems).toContain('fleetControlPlane') + expect(health.status).toBe('degraded') + expect(health.fleetControlPlane).toMatchObject({ + state: 'open', + consecutiveFailures: 4, + retryAtMs: BOOT_MS + 25_000, + }) + // MUST-NOT-FIRE: the circuit's lastError is free text with a socket path. + const rendered = JSON.stringify(health) + expect(rendered).not.toContain('/run/relay/broker.sock') + expect(rendered).not.toContain('ECONNREFUSED') + }) + + it('does not fault a closed fleet control-plane circuit', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + fleetControlPlane: { + state: 'closed', + consecutiveFailures: 0, + timeoutMs: 10_000, + failureThreshold: 3, + resetTimeoutMs: 30_000, + }, + }), + { nowMs: BOOT_MS + 1_000 }, + ) + + expect(health.degradedSubsystems).toEqual([]) + expect(health.fleetControlPlane).toMatchObject({ state: 'closed' }) + }) + + // Review follow-up on #300 (P2, cubic). A zero cadence made every in-flight + // pass instantly stalled and `missedPasses` Infinity, which JSON renders as + // null — a broken record about a working sweep. + it('falls back to the default cadence when the recorded interval is not positive', () => { + const health = publicHealthFromHeartbeat( + heartbeat({ + readinessReconcile: { + state: 'healthy', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 0, + lastStartedAtMs: BOOT_MS - 120_000, + lastCompletedAtMs: BOOT_MS - 180_000, + }, + }), + { nowMs: BOOT_MS }, + ) + + expect(health.readinessReconcile?.state).toBe('healthy') + expect(health.readinessReconcile?.missedPasses).toBe(2) + expect(Number.isFinite(health.readinessReconcile?.missedPasses ?? 0)).toBe(true) + expect(health.readinessReconcile?.intervalMs).toBeUndefined() + }) + // Review follow-up on #300 (Minor, CodeRabbit). The writer refuses to publish + // a non-positive cadence; a reader that accepts one from a remote process + // undoes that guarantee for everyone downstream of it. + it('re-applies the writer cadence and sign invariants when reading a remote record', () => { + const health = normalizePublicHealth({ + schemaVersion: 1, + ok: true, + status: 'degraded', + stale: false, + loopStatus: 'running', + degradedSubsystems: ['readinessReconcile'], + readinessReconcile: { + state: 'stalled', + consecutiveFailures: 3, + failureThreshold: 3, + intervalMs: 0, + inFlightMs: -5_000, + missedPasses: -12, + lastDurationMs: -1, + }, + }) + + expect(health?.readinessReconcile?.intervalMs).toBeUndefined() + expect(health?.readinessReconcile?.inFlightMs).toBeUndefined() + expect(health?.readinessReconcile?.missedPasses).toBeUndefined() + expect(health?.readinessReconcile?.lastDurationMs).toBeUndefined() + // The states and counters still come through: dropping a bad duration must + // not cost the operator the signal. + expect(health?.readinessReconcile).toMatchObject({ state: 'stalled', consecutiveFailures: 3 }) + }) + // Review follow-up on #300 (P2, cubic). "1.5 missed passes" is not a thing. + it('reports missed passes as a whole number', () => { + const health = normalizePublicHealth({ + schemaVersion: 1, + ok: true, + status: 'degraded', + stale: false, + loopStatus: 'running', + degradedSubsystems: ['readinessReconcile'], + readinessReconcile: { + state: 'stalled', + consecutiveFailures: 0, + failureThreshold: 3, + intervalMs: 60_000, + inFlightMs: 90_000, + missedPasses: 1.5, + }, + }) + + expect(health?.readinessReconcile?.missedPasses).toBe(1) + // A duration is genuinely fractional; only the count is not. + expect(health?.readinessReconcile?.inFlightMs).toBe(90_000) + }) +}) + diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts new file mode 100644 index 00000000..fcba744f --- /dev/null +++ b/src/orchestrator/public-health.ts @@ -0,0 +1,427 @@ +import { telemetryErrorClassName } from '../observability/error-class.js' +import type { FleetControlPlaneStatus } from '../fleet/control-plane-circuit' +import type { + FactoryEventListenerStatus, + FactoryLoopHeartbeat, + FactoryPublicEventListenerHealth, + FactoryPublicFleetControlPlaneHealth, + FactoryPublicHealth, + FactoryPublicReadinessReconcileHealth, + FactoryPublicSubsystemState, + FactoryReadinessReconcileStatus, +} from '../types' + +/** + * Public health projection (#295). + * + * A deployed Factory's only unauthenticated surface is the container's + * `/healthz`. Until now it carried subsystem *state strings* and nothing else, + * so an operator could see `degraded` without learning how badly, since when, + * or what class of failure — and could not see a wedged sweep at all, because + * a hang writes no state. The fields that answer those questions live in the + * loop heartbeat next to `lastError`, which is free text and must not be + * published. + * + * This module is that boundary. It builds the public record **by + * construction**: every field is named here, every number is coerced, every + * string is either a closed enum or passes the shared telemetry allowlist. + * Nothing is spread, so a field added upstream — or written by an older or + * hostile producer — cannot reach the public surface by default. + */ +export const FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION = 1 + +/** Default readiness-reconcile cadence, mirrored from the live daemon. */ +export const DEFAULT_READINESS_RECONCILE_INTERVAL_MS = 60_000 + +/** + * How many missed passes make an in-flight sweep `stalled`. + * + * A cold container legitimately spends minutes in its first pass — #36 + * measured a post-boot reconcile at 61 minutes while the Relayfile mirror + * hydrated — so a small multiple would cry wolf on every boot. Ten missed + * passes (ten minutes at the default cadence) is past any warm-path sweep, + * and `inFlightMs`/`missedPasses`/`lastCompletedAtMs` ship alongside so a + * reader can still tell "first pass since boot, still hydrating" from "was + * fine for hours, then stopped". + */ +export const READINESS_RECONCILE_STALL_INTERVALS = 10 + +/** Heartbeat age past which the whole record is treated as unknown, not green. */ +export const DEFAULT_PUBLIC_HEALTH_STALE_MS = 60_000 + +const READINESS_RECONCILE_STATES: readonly FactoryPublicSubsystemState[] = [ + 'not-running', + 'healthy', + 'retrying', + 'degraded', + 'stalled', +] + +const EVENT_LISTENER_STATES: readonly FactoryEventListenerStatus['state'][] = [ + 'not-listening', + 'starting', + 'subscribed', + 'polling', +] + +const FLEET_CONTROL_PLANE_STATES: readonly FleetControlPlaneStatus['state'][] = [ + 'closed', + 'open', + 'half-open', +] + +/** Subsystems whose degradation stops issues from being dispatched. */ +const DISPATCH_GATING_SUBSYSTEMS = ['readinessReconcile', 'eventListener', 'fleetControlPlane'] as const + +const finiteNumber = (value: unknown): number | undefined => + typeof value === 'number' && Number.isFinite(value) ? value : undefined + +/** A cadence is a denominator: zero or negative would make every derived ratio nonsense. */ +const positiveNumber = (value: unknown): number | undefined => { + const parsed = finiteNumber(value) + return parsed !== undefined && parsed > 0 ? parsed : undefined +} + +const counter = (value: unknown): number => { + const parsed = finiteNumber(value) + return parsed !== undefined && parsed >= 0 ? Math.floor(parsed) : 0 +} + +/** + * The widest instant `Date` can represent (ECMA-262 time-value limit). + * + * Review follow-up on #300 (P2, codex): a finite number is not a valid date. + * `new Date(1e300).toISOString()` throws, and these numbers arrive from a + * remote process — so a hostile or corrupted record could abort a renderer + * that was asked to explain an outage. A timestamp outside the range is + * dropped rather than published. + */ +const MAX_TIME_VALUE_MS = 8.64e15 + +const timestamp = (value: unknown): number | undefined => { + const parsed = finiteNumber(value) + return parsed !== undefined && Math.abs(parsed) <= MAX_TIME_VALUE_MS ? parsed : undefined +} + +const optionalTimestamp = (key: K, value: unknown): Partial> => { + const parsed = timestamp(value) + return parsed === undefined ? {} : { [key]: parsed } as Partial> +} + +const plainRecord = (value: unknown): Record | undefined => + value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : undefined + +const optionalNumber = (key: K, value: unknown): Partial> => { + const parsed = finiteNumber(value) + return parsed === undefined ? {} : { [key]: parsed } as Partial> +} + +/** + * A duration that cannot be negative, dropped rather than republished. + * + * The writer never emits these; a remote process on another version might + * (#300 review, CodeRabbit). Dropping the field keeps the promise this + * module's doc comment makes to its callers about the shape they get. + */ +const optionalDuration = (key: K, value: unknown): Partial> => { + const parsed = finiteNumber(value) + return parsed === undefined || parsed < 0 ? {} : { [key]: parsed } as Partial> +} + +/** A count of whole passes: fractions are not a thing an operator can read. */ +const optionalCount = (key: K, value: unknown): Partial> => { + const parsed = finiteNumber(value) + return parsed === undefined || parsed < 0 + ? {} + : { [key]: Math.floor(parsed) } as Partial> +} + +const optionalPositive = (key: K, value: unknown): Partial> => { + const parsed = positiveNumber(value) + return parsed === undefined ? {} : { [key]: parsed } as Partial> +} + +/** Control characters stripped, length bounded: this text can reach a terminal. */ +const boundedText = (value: string): string => + // C0 and C1 alike (#300 review, P2, cubic): some terminals interpret the + // C1 range as escape introducers. + value.replace(/[\u0000-\u001F\u007F-\u009F]+/gu, ' ').trim().slice(0, 300) + +const enumValue = (value: unknown, allowed: readonly T[]): T | 'unknown' => + typeof value === 'string' && (allowed as readonly string[]).includes(value) ? value as T : 'unknown' + +/** + * How long the current pass has been running, or `undefined` when none is. + * + * A sweep that hangs takes neither the success nor the failure path, so no + * field is written while it is stuck. The only evidence is that its start + * timestamp is newer than both settle timestamps — which is why the *relative + * order* of these three numbers, not any state string, is the signal. + */ +export function readinessReconcileInFlightMs( + status: Pick< + FactoryReadinessReconcileStatus, + 'lastStartedAtMs' | 'lastCompletedAtMs' | 'lastFailureAtMs' + >, + nowMs: number, +): number | undefined { + const startedAtMs = timestamp(status.lastStartedAtMs) + if (startedAtMs === undefined) return undefined + const settledAtMs = Math.max( + timestamp(status.lastCompletedAtMs) ?? Number.NEGATIVE_INFINITY, + timestamp(status.lastFailureAtMs) ?? Number.NEGATIVE_INFINITY, + ) + if (settledAtMs >= startedAtMs) return undefined + return Math.max(0, nowMs - startedAtMs) +} + +/** + * The state a reader should believe, which is not always the one on record. + * + * `state` as persisted is last-write-wins over the last *settled* pass, so it + * reports `healthy` for as long as a hang lasts. This re-derives it against + * the clock: an in-flight pass past the stall threshold is `stalled` no matter + * what the last completed pass said. + */ +export function derivedReadinessReconcileState( + status: Pick< + FactoryReadinessReconcileStatus, + 'state' | 'intervalMs' | 'lastStartedAtMs' | 'lastCompletedAtMs' | 'lastFailureAtMs' + >, + nowMs: number, +): FactoryPublicSubsystemState | 'unknown' { + const reported = enumValue(status.state, READINESS_RECONCILE_STATES) + // A daemon that is not running has no pass in flight; its own state wins. + if (reported === 'not-running') return reported + const inFlightMs = readinessReconcileInFlightMs(status, nowMs) + if (inFlightMs === undefined) return reported + const intervalMs = positiveNumber(status.intervalMs) ?? DEFAULT_READINESS_RECONCILE_INTERVAL_MS + return inFlightMs > intervalMs * READINESS_RECONCILE_STALL_INTERVALS ? 'stalled' : reported +} + +function readinessReconcileHealth( + status: FactoryReadinessReconcileStatus, + nowMs: number, +): FactoryPublicReadinessReconcileHealth { + // Review follow-up on #300 (P2, cubic): a recorded `intervalMs: 0` made + // every in-flight pass instantly stalled and `missedPasses` Infinity, which + // JSON renders as null. An unusable cadence falls back to the default and is + // not republished as though it were real. + const intervalMs = positiveNumber(status.intervalMs) + const inFlightMs = readinessReconcileInFlightMs(status, nowMs) + const cadenceMs = intervalMs ?? DEFAULT_READINESS_RECONCILE_INTERVAL_MS + return { + state: derivedReadinessReconcileState(status, nowMs), + consecutiveFailures: counter(status.consecutiveFailures), + failureThreshold: counter(status.failureThreshold), + ...(intervalMs !== undefined ? { intervalMs } : {}), + ...(finiteNumber(status.lastDurationMs) !== undefined + ? { lastDurationMs: finiteNumber(status.lastDurationMs) } + : {}), + ...optionalTimestamp('lastStartedAtMs', status.lastStartedAtMs), + ...optionalTimestamp('lastCompletedAtMs', status.lastCompletedAtMs), + ...optionalTimestamp('lastFailureAtMs', status.lastFailureAtMs), + ...(inFlightMs !== undefined + ? { inFlightMs, missedPasses: Math.floor(inFlightMs / cadenceMs) } + : {}), + // `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. + ...(status.lastErrorClass !== undefined || status.lastError !== undefined + ? { lastErrorClass: telemetryErrorClassName(status.lastErrorClass) } + : {}), + } +} + +function fleetControlPlaneHealth( + status: FleetControlPlaneStatus, +): FactoryPublicFleetControlPlaneHealth { + return { + state: enumValue(status.state, FLEET_CONTROL_PLANE_STATES), + consecutiveFailures: counter(status.consecutiveFailures), + failureThreshold: counter(status.failureThreshold), + ...optionalTimestamp('lastFailureAtMs', status.lastFailureAtMs), + ...optionalTimestamp('retryAtMs', status.retryAtMs), + // `lastError` stays behind /evidence: a roster probe failure names the + // broker socket path. + } +} + +/** + * Project a loop heartbeat into the record safe to serve unauthenticated. + * + * `nowMs` must come from the *writer's* clock, not a remote reader's: every + * derived duration here is a difference against timestamps the daemon + * produced, and comparing them to a laptop's clock would report skew as + * stall. Readers get `ageMs` instead and can bound the staleness themselves. + */ +export function publicHealthFromHeartbeat( + heartbeat: FactoryLoopHeartbeat | undefined, + opts: { nowMs?: number; staleMs?: number } = {}, +): FactoryPublicHealth { + const nowMs = opts.nowMs ?? Date.now() + const staleMs = opts.staleMs ?? DEFAULT_PUBLIC_HEALTH_STALE_MS + if (!heartbeat) { + return { + schemaVersion: FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + ok: false, + status: 'unknown', + stale: true, + reason: 'heartbeat missing', + degradedSubsystems: [], + } + } + + const updatedAtMs = timestamp(heartbeat.updatedAtMs) + const ageMs = updatedAtMs === undefined ? undefined : Math.max(0, nowMs - updatedAtMs) + const stale = ageMs === undefined || ageMs > staleMs + const loopStatus = enumValue(heartbeat.status, ['running', 'idle', 'stopping'] as const) + + const readinessReconcile = heartbeat.readinessReconcile + ? readinessReconcileHealth(heartbeat.readinessReconcile, nowMs) + : undefined + const fleetControlPlane = heartbeat.fleetControlPlane + ? fleetControlPlaneHealth(heartbeat.fleetControlPlane) + : undefined + const eventListener: FactoryPublicEventListenerHealth | undefined = heartbeat.eventListener + // Only the state. `reason` is assembled free text and stays behind the + // authenticated surface. + ? { state: enumValue(heartbeat.eventListener.state, EVENT_LISTENER_STATES) } + : undefined + + // A daemon that is not running a readiness loop is not a live dispatcher — + // a bounded `factory loop` reports `not-running` here and is not supposed to + // hold a subscription. Only a live instance's listener is dispatch-gating. + const liveDispatcher = readinessReconcile !== undefined && readinessReconcile.state !== 'not-running' + const degradedSubsystems = DISPATCH_GATING_SUBSYSTEMS.filter((name) => { + if (name === 'readinessReconcile') { + return readinessReconcile !== undefined && + readinessReconcile.state !== 'healthy' && + readinessReconcile.state !== 'not-running' + } + if (name === 'fleetControlPlane') { + // An open circuit fails every spawn fast; half-open is one probe away + // from either. Both mean dispatch is not admitting work normally. + return fleetControlPlane !== undefined && fleetControlPlane.state !== 'closed' + } + // Review follow-up on #300 (P2, codex): `starting` is what a live daemon + // reports before `#startLiveSubscription` installs the subscription. No + // listener is registered, and startup can be lengthy, so anything short of + // a registered subscription or an active poll is amber — not green. + return liveDispatcher && + eventListener !== undefined && + eventListener.state !== 'subscribed' && + eventListener.state !== 'polling' + }) + + // Deliberate split (#295, deliverable 2). + // + // `ok` answers "is this process alive", because that is the question the + // platform asks: the container ping endpoint is `/healthz`, and a non-200 + // there recycles the container. Recycling a wedged Factory destroys the + // in-memory evidence of the wedge and restarts the cold-start hydration + // that #36 measured at 61 minutes, so a dispatch-gating degradation must + // not be able to reach into container lifecycle. + // + // `status` is the amber a liveness bit cannot express. No platform reads + // it, so a monitor can alert on `status !== "ok"` — or on + // `degradedSubsystems` being non-empty — and get the signal that was + // missing during the outage, with no restart-loop risk. + const ok = !stale && loopStatus !== 'stopping' && loopStatus !== 'unknown' + const status = !ok ? 'unknown' : degradedSubsystems.length > 0 ? 'degraded' : 'ok' + const reason = stale + ? 'heartbeat stale' + : loopStatus === 'stopping' + ? 'loop stopping' + : degradedSubsystems.length > 0 + ? `dispatch-gating subsystem not healthy: ${degradedSubsystems.join(', ')}` + : undefined + + return { + schemaVersion: FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + ok, + status, + stale, + ...(updatedAtMs !== undefined ? { updatedAtMs } : {}), + ...(ageMs !== undefined ? { ageMs } : {}), + loopStatus, + degradedSubsystems: [...degradedSubsystems], + ...(reason ? { reason } : {}), + ...(readinessReconcile ? { readinessReconcile } : {}), + ...(eventListener ? { eventListener } : {}), + ...(fleetControlPlane ? { fleetControlPlane } : {}), + } +} + +/** + * Re-read a health record that arrived over the wire. + * + * A reader of a remote `/healthz` holds parsed JSON from a process it does + * not control and that may be several versions behind. Running it back + * through the same coercions the writer used means a caller can rely on the + * shape — and means an unrecognised state or a hostile string cannot reach a + * terminal or a downstream report just because it arrived over HTTP. + * + * The derived fields are read, not recomputed: they were derived against the + * writer's clock, and a reader's clock would report skew as stall. + * + * Returns `undefined` when the record is absent, which is how a caller tells + * "instance predates the diagnostics block" from "instance is unhealthy". + */ +export function normalizePublicHealth(value: unknown): FactoryPublicHealth | undefined { + const record = plainRecord(value) + if (!record) return undefined + const readiness = plainRecord(record.readinessReconcile) + const listener = plainRecord(record.eventListener) + const fleet = plainRecord(record.fleetControlPlane) + const degradedSubsystems = Array.isArray(record.degradedSubsystems) + ? DISPATCH_GATING_SUBSYSTEMS.filter((name) => (record.degradedSubsystems as unknown[]).includes(name)) + : [] + const status = enumValue(record.status, ['ok', 'degraded'] as const) + return { + schemaVersion: finiteNumber(record.schemaVersion) ?? FACTORY_PUBLIC_HEALTH_SCHEMA_VERSION, + ok: record.ok === true, + status, + stale: record.stale === true, + ...optionalTimestamp('updatedAtMs', record.updatedAtMs), + ...optionalDuration('ageMs', record.ageMs), + loopStatus: enumValue(record.loopStatus, ['running', 'idle', 'stopping'] as const), + degradedSubsystems: [...degradedSubsystems], + ...(typeof record.reason === 'string' + ? { reason: boundedText(record.reason) } + : {}), + ...(readiness + ? { + readinessReconcile: { + state: enumValue(readiness.state, READINESS_RECONCILE_STATES), + consecutiveFailures: counter(readiness.consecutiveFailures), + failureThreshold: counter(readiness.failureThreshold), + ...optionalPositive('intervalMs', readiness.intervalMs), + ...optionalDuration('lastDurationMs', readiness.lastDurationMs), + ...optionalTimestamp('lastStartedAtMs', readiness.lastStartedAtMs), + ...optionalTimestamp('lastCompletedAtMs', readiness.lastCompletedAtMs), + ...optionalTimestamp('lastFailureAtMs', readiness.lastFailureAtMs), + ...optionalDuration('inFlightMs', readiness.inFlightMs), + ...optionalCount('missedPasses', readiness.missedPasses), + ...(readiness.lastErrorClass !== undefined + ? { lastErrorClass: telemetryErrorClassName(readiness.lastErrorClass) } + : {}), + }, + } + : {}), + ...(listener ? { eventListener: { state: enumValue(listener.state, EVENT_LISTENER_STATES) } } : {}), + ...(fleet + ? { + fleetControlPlane: { + state: enumValue(fleet.state, FLEET_CONTROL_PLANE_STATES), + consecutiveFailures: counter(fleet.consecutiveFailures), + failureThreshold: counter(fleet.failureThreshold), + ...optionalTimestamp('lastFailureAtMs', fleet.lastFailureAtMs), + ...optionalTimestamp('retryAtMs', fleet.retryAtMs), + }, + } + : {}), + } +} diff --git a/src/types.ts b/src/types.ts index ca012aa4..a35523c2 100644 --- a/src/types.ts +++ b/src/types.ts @@ -142,17 +142,116 @@ export interface FactoryLoopHeartbeat { readinessReconcile?: FactoryReadinessReconcileStatus /** Daemon-owned dispatch admission state; status readers must prefer this over a fresh local Factory instance. */ fleetControlPlane?: FleetControlPlaneStatus + /** + * Redacted projection of this record, safe to serve unauthenticated (#295). + * + * The deployed container reads this file to answer `/healthz` and has no + * redaction logic of its own, so the daemon publishes the already-safe view + * rather than leaving the boundary to whoever serves it. + */ + health?: FactoryPublicHealth } +/** + * `stalled` is derived, not written: a sweep that hangs takes neither the + * success nor the failure path, so nothing updates `state` while it is stuck. + * See `derivedReadinessReconcileState`. + */ +export type FactoryReadinessReconcileState = + | 'not-running' + | 'healthy' + | 'retrying' + | 'degraded' + | 'stalled' + export interface FactoryReadinessReconcileStatus { - state: 'not-running' | 'healthy' | 'retrying' | 'degraded' + state: FactoryReadinessReconcileState consecutiveFailures: number failureThreshold: number + /** Sweep cadence — the denominator that turns `inFlightMs` into missed passes. */ + intervalMs?: number lastDurationMs?: number lastStartedAtMs?: number lastCompletedAtMs?: number lastFailureAtMs?: number + /** Age of a pass that started and has neither completed nor failed. */ + inFlightMs?: number + /** Free text; authenticated surfaces only. */ lastError?: string + /** Allowlisted class name of `lastError`; publishable. */ + lastErrorClass?: string +} + +/** A subsystem state as published, plus the value an unrecognised one collapses to. */ +export type FactoryPublicSubsystemState = FactoryReadinessReconcileState + +export interface FactoryPublicReadinessReconcileHealth { + state: FactoryPublicSubsystemState | 'unknown' + consecutiveFailures: number + failureThreshold: number + intervalMs?: number + lastDurationMs?: number + lastStartedAtMs?: number + lastCompletedAtMs?: number + lastFailureAtMs?: number + inFlightMs?: number + /** `inFlightMs` expressed in sweeps that should have run and did not. */ + missedPasses?: number + lastErrorClass?: string +} + +export interface FactoryPublicEventListenerHealth { + state: FactoryEventListenerStatus['state'] +} + +/** + * The broker/fleet mutation gate, redacted (#300 review). + * + * An open circuit fails every spawn and resume fast, so it gates dispatch as + * hard as a failing readiness sweep. Its `lastError` is free text — a roster + * probe failure names sockets and paths — so only the state, the counters and + * the retry instant cross. + */ +export interface FactoryPublicFleetControlPlaneHealth { + state: FleetControlPlaneStatus['state'] | 'unknown' + consecutiveFailures: number + failureThreshold: number + lastFailureAtMs?: number + retryAtMs?: number +} + +/** + * The unauthenticated health record (#295). + * + * `ok` is process liveness — the question the container ping endpoint asks, + * and the only one whose answer may recycle a container. `status` is the + * amber: dispatch-gating degradation that an operator or monitor must see, + * carried where no platform will act on it. + */ +export interface FactoryPublicHealth { + schemaVersion: number + ok: boolean + status: 'ok' | 'degraded' | 'unknown' + /** + * Stamped when the daemon WROTE this record, not when it was read. + * + * A record served out of a file the daemon has stopped updating still says + * `stale: false`, because it was fresh at write time. Freshness is therefore + * `updatedAtMs` measured against the serving process's clock — which is what + * the container's own liveness verdict does, and why that verdict outranks + * this field (#300 review). + */ + stale: boolean + updatedAtMs?: number + ageMs?: number + loopStatus?: FactoryLoopHeartbeatStatus | 'unknown' + /** Dispatch-gating subsystems that are not healthy right now. */ + degradedSubsystems: string[] + /** Why this is not plain `ok`, assembled from closed vocabularies only. */ + reason?: string + readinessReconcile?: FactoryPublicReadinessReconcileHealth + eventListener?: FactoryPublicEventListenerHealth + fleetControlPlane?: FactoryPublicFleetControlPlaneHealth } export interface FactoryInFlightRegistryAgent {