From eb5a8040fe1ce303325c9bfa3b462b072977dc5f Mon Sep 17 00:00:00 2001 From: Tim Disney Date: Wed, 22 Jul 2026 10:38:38 -0700 Subject: [PATCH] Phase 4: add CheckLedger (checkruns_ledger table) A separate durable ledger for check runs, keyed by artifact version (uri#cid) plus the linked commit. Independent from TurnLedger so checks and turns keep their own retry budgets and never clobber each other's rows. States running|done|crashed|gave_up with attempts/cooldown/retryBound, running() for orphan reconciliation, and reset(artifactUri) for a check-reset escape hatch. Co-Authored-By: Claude Fable 5 --- packages/daemon/src/check-ledger.ts | 196 +++++++++++++++++++++ packages/daemon/src/index.ts | 1 + packages/daemon/test/check-ledger.test.mjs | 106 +++++++++++ packages/ui/src/server.ts | 107 ++++++----- 4 files changed, 365 insertions(+), 45 deletions(-) create mode 100644 packages/daemon/src/check-ledger.ts create mode 100644 packages/daemon/test/check-ledger.test.mjs diff --git a/packages/daemon/src/check-ledger.ts b/packages/daemon/src/check-ledger.ts new file mode 100644 index 0000000..68b4c5d --- /dev/null +++ b/packages/daemon/src/check-ledger.ts @@ -0,0 +1,196 @@ +import { DatabaseSync } from 'node:sqlite' +import type { StrongRef } from '@radial/core' + +/** + * A check run has no `awaiting_input`/`fulfilled` distinction the way a turn does — a check either + * finished (`done`, a checkrun record was written) or it did not (`crashed`, retryable until + * `gave_up`). `running` is the in-flight marker orphan reconciliation keys off after a restart. + */ +export type CheckState = 'running' | 'done' | 'crashed' | 'gave_up' + +export interface CheckRow { + artifactUri: string + artifactCid: string + commit: string + state: CheckState + attempts: number + nextEligibleAt?: string + containerLabel?: string + checkrunRef?: StrongRef + updatedAt: string +} + +export interface CheckLedgerOptions { + retryBound?: number + cooldownMs?: number + now?: () => string +} + +interface Row { + artifact_uri: string + artifact_cid: string + commit_sha: string + state: string | null + attempts: number + next_eligible_at: string | null + container_label: string | null + checkrun_ref_uri: string | null + checkrun_ref_cid: string | null + updated_at: string | null +} + +function toRow(row: Row): CheckRow { + return { + artifactUri: row.artifact_uri, + artifactCid: row.artifact_cid, + commit: row.commit_sha, + state: (row.state ?? 'running') as CheckState, + attempts: row.attempts, + ...(row.next_eligible_at !== null ? { nextEligibleAt: row.next_eligible_at } : {}), + ...(row.container_label !== null ? { containerLabel: row.container_label } : {}), + ...(row.checkrun_ref_uri !== null && row.checkrun_ref_cid !== null + ? { checkrunRef: { uri: row.checkrun_ref_uri, cid: row.checkrun_ref_cid } } + : {}), + updatedAt: row.updated_at ?? '', + } +} + +/** Composite key for a check run: the exact artifact version (uri#cid) plus the commit it links. */ +export interface CheckKey { + artifactUri: string + artifactCid: string + commit: string +} + +/** + * Durable record of every check run the daemon has attempted, keyed by the artifact version + * (uri#cid) and the commit it linked. Deliberately a *separate* table from `TurnLedger`'s `turns` + * (its own DB file/table `checkruns_ledger`): checks and turns are independent workstreams with + * their own retry budgets, and the two must never share or clobber each other's rows. Together with + * the index-side dedup gate (a trusted checkrun already authored by one of this daemon's DIDs), this + * covers the in-flight / ingestion-lag window — the span after we launch a container but before its + * checkrun record has been ingested back into the index. + */ +export class CheckLedger { + readonly #database: DatabaseSync + readonly #retryBound: number + readonly #cooldownMs: number + readonly #now: () => string + + constructor(path = ':memory:', options: CheckLedgerOptions = {}) { + this.#database = new DatabaseSync(path) + this.#database.exec(` + CREATE TABLE IF NOT EXISTS checkruns_ledger ( + artifact_uri TEXT NOT NULL, artifact_cid TEXT NOT NULL, commit_sha TEXT NOT NULL, + state TEXT, attempts INTEGER NOT NULL DEFAULT 0, next_eligible_at TEXT, + container_label TEXT, checkrun_ref_uri TEXT, checkrun_ref_cid TEXT, updated_at TEXT, + PRIMARY KEY (artifact_uri, artifact_cid, commit_sha) + ) STRICT; + `) + this.#retryBound = options.retryBound ?? 3 + this.#cooldownMs = options.cooldownMs ?? 300_000 + this.#now = options.now ?? (() => new Date().toISOString()) + } + + get(key: CheckKey): CheckRow | undefined { + const row = this.#database + .prepare('SELECT * FROM checkruns_ledger WHERE artifact_uri = ? AND artifact_cid = ? AND commit_sha = ?') + .get(key.artifactUri, key.artifactCid, key.commit) as Row | undefined + return row ? toRow(row) : undefined + } + + eligible(key: CheckKey, now: string = this.#now()): boolean { + const row = this.get(key) + if (!row) return true + switch (row.state) { + case 'running': + case 'done': + case 'gave_up': + return false + case 'crashed': + return row.nextEligibleAt !== undefined && now >= row.nextEligibleAt + default: + return true + } + } + + markRunning(key: CheckKey, fields: { containerLabel: string }): void { + const now = this.#now() + this.#database + .prepare(` + INSERT INTO checkruns_ledger + (artifact_uri, artifact_cid, commit_sha, state, attempts, next_eligible_at, container_label, updated_at) + VALUES (?, ?, ?, 'running', 0, NULL, ?, ?) + ON CONFLICT (artifact_uri, artifact_cid, commit_sha) DO UPDATE SET + state = 'running', + next_eligible_at = NULL, + container_label = excluded.container_label, + updated_at = excluded.updated_at + `) + .run(key.artifactUri, key.artifactCid, key.commit, fields.containerLabel, now) + } + + markDone(key: CheckKey, checkrunRef?: StrongRef): void { + const now = this.#now() + const existing = this.get(key) + this.#database + .prepare(` + INSERT INTO checkruns_ledger + (artifact_uri, artifact_cid, commit_sha, state, attempts, checkrun_ref_uri, checkrun_ref_cid, updated_at) + VALUES (?, ?, ?, 'done', ?, ?, ?, ?) + ON CONFLICT (artifact_uri, artifact_cid, commit_sha) DO UPDATE SET + state = 'done', + checkrun_ref_uri = excluded.checkrun_ref_uri, + checkrun_ref_cid = excluded.checkrun_ref_cid, + updated_at = excluded.updated_at + `) + .run( + key.artifactUri, + key.artifactCid, + key.commit, + existing?.attempts ?? 0, + checkrunRef?.uri ?? null, + checkrunRef?.cid ?? null, + now, + ) + } + + /** Increments attempts; past `retryBound` the check is given up on for good. Caller logs loudly on `gave_up`. */ + markCrashed(key: CheckKey): { state: CheckState; attempts: number } { + const now = this.#now() + const existing = this.get(key) + const attempts = (existing?.attempts ?? 0) + 1 + const gaveUp = attempts >= this.#retryBound + const state: CheckState = gaveUp ? 'gave_up' : 'crashed' + const nextEligibleAt = gaveUp ? null : new Date(Date.parse(now) + this.#cooldownMs).toISOString() + this.#database + .prepare(` + INSERT INTO checkruns_ledger + (artifact_uri, artifact_cid, commit_sha, state, attempts, next_eligible_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (artifact_uri, artifact_cid, commit_sha) DO UPDATE SET + state = excluded.state, + attempts = excluded.attempts, + next_eligible_at = excluded.next_eligible_at, + updated_at = excluded.updated_at + `) + .run(key.artifactUri, key.artifactCid, key.commit, state, attempts, nextEligibleAt, now) + return { state, attempts } + } + + /** Clears every row for an artifact uri (all versions/commits) so its checks re-run — the + * operator-facing `radiald check reset ` escape hatch, mirroring `turn reset`. */ + reset(artifactUri: string): void { + this.#database.prepare('DELETE FROM checkruns_ledger WHERE artifact_uri = ?').run(artifactUri) + } + + /** Rows left in `running` — orphans to reconcile if the daemon restarted mid-check. */ + running(): CheckRow[] { + const rows = this.#database.prepare("SELECT * FROM checkruns_ledger WHERE state = 'running'").all() as Row[] + return rows.map(toRow) + } + + close(): void { + this.#database.close() + } +} diff --git a/packages/daemon/src/index.ts b/packages/daemon/src/index.ts index 0ec8780..f0ca607 100644 --- a/packages/daemon/src/index.ts +++ b/packages/daemon/src/index.ts @@ -1,5 +1,6 @@ export * from './actors.js' export * from './bundle-writer.js' +export * from './check-ledger.js' export * from './config.js' export * from './container.js' export * from './dispatch.js' diff --git a/packages/daemon/test/check-ledger.test.mjs b/packages/daemon/test/check-ledger.test.mjs new file mode 100644 index 0000000..07af8b6 --- /dev/null +++ b/packages/daemon/test/check-ledger.test.mjs @@ -0,0 +1,106 @@ +import assert from 'node:assert/strict' +import { it } from 'node:test' +import { CheckLedger } from '../dist/index.js' + +const KEY = { artifactUri: 'at://did/artifact/a1', artifactCid: 'cid-a1', commit: 'deadbeef' } +// Same artifact, a different version (cid) — a distinct check key that must never collide with KEY. +const KEY_V2 = { artifactUri: 'at://did/artifact/a1', artifactCid: 'cid-a2', commit: 'deadbeef' } +// Same artifact version, a different commit — also distinct. +const KEY_C2 = { artifactUri: 'at://did/artifact/a1', artifactCid: 'cid-a1', commit: 'feedface' } + +it('a fresh key has no row and is eligible', () => { + const ledger = new CheckLedger() + assert.equal(ledger.get(KEY), undefined) + assert.equal(ledger.eligible(KEY), true) + ledger.close() +}) + +it('markRunning makes a key ineligible and records the container label', () => { + const ledger = new CheckLedger() + ledger.markRunning(KEY, { containerLabel: 'radial.check.abc' }) + assert.equal(ledger.eligible(KEY), false) + const row = ledger.get(KEY) + assert.equal(row.state, 'running') + assert.equal(row.attempts, 0) + assert.equal(row.containerLabel, 'radial.check.abc') + ledger.close() +}) + +it('the composite key distinguishes version and commit', () => { + const ledger = new CheckLedger() + ledger.markRunning(KEY, { containerLabel: 'L1' }) + // A different cid or a different commit is a different row: still eligible, no row yet. + assert.equal(ledger.eligible(KEY_V2), true) + assert.equal(ledger.eligible(KEY_C2), true) + assert.equal(ledger.get(KEY_V2), undefined) + assert.equal(ledger.get(KEY_C2), undefined) + ledger.close() +}) + +it('markDone stores the checkrun ref and is terminal (ineligible)', () => { + const ledger = new CheckLedger() + ledger.markRunning(KEY, { containerLabel: 'L1' }) + ledger.markDone(KEY, { uri: 'at://did/checkrun/c1', cid: 'cid-c1' }) + const row = ledger.get(KEY) + assert.equal(row.state, 'done') + assert.deepEqual(row.checkrunRef, { uri: 'at://did/checkrun/c1', cid: 'cid-c1' }) + assert.equal(ledger.eligible(KEY), false) + ledger.close() +}) + +it('markRunning after a crash carries attempts forward (does not reset the counter)', () => { + const ledger = new CheckLedger() + ledger.markRunning(KEY, { containerLabel: 'L1' }) + ledger.markCrashed(KEY) + ledger.markRunning(KEY, { containerLabel: 'L2' }) + assert.equal(ledger.get(KEY).attempts, 1) + ledger.close() +}) + +it('crash gates eligibility behind a cooldown, then clears', () => { + let now = '2026-01-01T00:00:00.000Z' + const ledger = new CheckLedger(':memory:', { cooldownMs: 300_000, retryBound: 3, now: () => now }) + ledger.markRunning(KEY, { containerLabel: 'L1' }) + const result = ledger.markCrashed(KEY) + assert.deepEqual(result, { state: 'crashed', attempts: 1 }) + assert.equal(ledger.eligible(KEY), false) + assert.equal(ledger.eligible(KEY, '2026-01-01T00:04:59.999Z'), false) + assert.equal(ledger.eligible(KEY, '2026-01-01T00:05:00.000Z'), true) + ledger.close() +}) + +it('gives up for good once attempts reach retryBound', () => { + const ledger = new CheckLedger(':memory:', { retryBound: 3 }) + assert.deepEqual(ledger.markCrashed(KEY), { state: 'crashed', attempts: 1 }) + assert.deepEqual(ledger.markCrashed(KEY), { state: 'crashed', attempts: 2 }) + assert.deepEqual(ledger.markCrashed(KEY), { state: 'gave_up', attempts: 3 }) + // gave_up is terminal regardless of any cooldown having elapsed. + assert.equal(ledger.eligible(KEY, '2999-01-01T00:00:00.000Z'), false) + ledger.close() +}) + +it('running() lists only in-flight rows for orphan reconciliation', () => { + const ledger = new CheckLedger() + ledger.markRunning(KEY, { containerLabel: 'L1' }) + ledger.markRunning(KEY_C2, { containerLabel: 'L2' }) + ledger.markDone(KEY_C2, { uri: 'at://did/checkrun/c2', cid: 'cid-c2' }) + const running = ledger.running() + assert.equal(running.length, 1) + assert.equal(running[0].artifactUri, KEY.artifactUri) + assert.equal(running[0].commit, KEY.commit) + assert.equal(running[0].containerLabel, 'L1') + ledger.close() +}) + +it('reset clears every row for an artifact uri (all versions/commits)', () => { + const ledger = new CheckLedger() + ledger.markDone(KEY, { uri: 'at://c/1', cid: 'c1' }) + ledger.markDone(KEY_V2, { uri: 'at://c/2', cid: 'c2' }) + ledger.markDone(KEY_C2, { uri: 'at://c/3', cid: 'c3' }) + ledger.reset(KEY.artifactUri) + assert.equal(ledger.get(KEY), undefined) + assert.equal(ledger.get(KEY_V2), undefined) + assert.equal(ledger.get(KEY_C2), undefined) + assert.equal(ledger.eligible(KEY), true) + ledger.close() +}) diff --git a/packages/ui/src/server.ts b/packages/ui/src/server.ts index fc0cc5a..468b361 100644 --- a/packages/ui/src/server.ts +++ b/packages/ui/src/server.ts @@ -163,83 +163,100 @@ function redirect(response: ServerResponse, location: string): void { response.end() } -/** Map a submitted form to the CLI command that writes it; return where to send the browser next. */ -async function performAction(path: string, form: URLSearchParams): Promise { - if (path === '/login') { - const identifier = requireField(form, 'identifier') - const profile = (form.get('profile') || identifier).trim() - const override = form.get('pds')?.trim() - const service = override || String(await resolveIdentityPds(identifier)) - const session = await createSession(service, identifier, requireField(form, 'password')) - await store.save({ ...session, profile }) - return '/' - } +/** + * A submitted form mapped to a `radial` CLI invocation: the exact `args` runCli + * gets, plus where to send the browser once it lands. `redirect` is a function + * because a few actions (create-space, create-goal) can only name their + * destination from the record they just wrote. Pure and synchronous so request + * shapes are unit-testable without a PDS or writer (see test/build-action.test.mjs). + */ +export interface BuiltAction { + args: string[] + redirect: (result: { primary: { uri: string } }) => string +} - const deps = await depsFor(form.get('profile') ?? undefined) +/** The form→CLI-args mapping, extracted from `performAction` as a testable seam. Does not cover `/login` (not a CLI command). */ +export function buildAction(path: string, form: URLSearchParams): BuiltAction { const optional = (key: string): string => form.get(key)?.trim() ?? '' + const require_ = (key: string): string => requireField(form, key) + const toGoal = (goal: string) => () => `/goal?uri=${encodeURIComponent(goal)}` + const toSpace = (space: string) => () => `/space?uri=${encodeURIComponent(space)}` switch (path) { case '/space': { - const args = ['space', 'create', '--name', requireField(form, 'name'), '--description', optional('description')] + const args = ['space', 'create', '--name', require_('name'), '--description', optional('description')] if (form.get('builtins') !== 'on') args.push('--no-builtins') - const result = await runCli(args, deps) - return `/space?uri=${encodeURIComponent(result.primary.uri)}` + return { args, redirect: (result) => `/space?uri=${encodeURIComponent(result.primary.uri)}` } } case '/member': { - const space = requireField(form, 'space') - await runCli( - ['member', 'add', '--space', space, '--actor', requireField(form, 'actor'), '--kind', requireField(form, 'kind'), '--role', requireField(form, 'role')], - deps, - ) - return `/space?uri=${encodeURIComponent(space)}` + const space = require_('space') + return { + args: ['member', 'add', '--space', space, '--actor', require_('actor'), '--kind', require_('kind'), '--role', require_('role')], + redirect: toSpace(space), + } } case '/project': { - const space = requireField(form, 'space') - const args = ['project', 'create', '--space', space, '--name', requireField(form, 'name'), '--git-url', requireField(form, 'gitUrl')] + const space = require_('space') + const args = ['project', 'create', '--space', space, '--name', require_('name'), '--git-url', require_('gitUrl')] if (optional('defaultBranch')) args.push('--default-branch', optional('defaultBranch')) - await runCli(args, deps) - return `/space?uri=${encodeURIComponent(space)}` + return { args, redirect: toSpace(space) } } case '/goal': { - const result = await runCli( - ['goal', 'create', '--project', requireField(form, 'project'), '--title', requireField(form, 'title'), '--body', requireField(form, 'body')], - deps, - ) - return `/goal?uri=${encodeURIComponent(result.primary.uri)}` + return { + args: ['goal', 'create', '--project', require_('project'), '--title', require_('title'), '--body', require_('body')], + redirect: (result) => `/goal?uri=${encodeURIComponent(result.primary.uri)}`, + } } case '/request': { - const goal = requireField(form, 'goal') - const args = ['request', 'create', '--goal', goal, '--type', requireField(form, 'type')] + const goal = require_('goal') + const args = ['request', 'create', '--goal', goal, '--type', require_('type')] if (optional('assignee')) args.push('--assignee', optional('assignee')) if (optional('basedOn')) args.push('--based-on', optional('basedOn')) + if (optional('subject')) args.push('--subject', optional('subject')) if (optional('brief')) args.push('--brief', optional('brief')) - await runCli(args, deps) - return `/goal?uri=${encodeURIComponent(goal)}` + return { args, redirect: toGoal(goal) } } case '/artifact': { - const goal = requireField(form, 'goal') - const args = ['artifact', 'post', '--request', requireField(form, 'request'), '--body', requireField(form, 'body')] + const goal = require_('goal') + const args = ['artifact', 'post', '--request', require_('request'), '--body', require_('body')] if (optional('prev')) args.push('--prev', optional('prev')) - await runCli(args, deps) - return `/goal?uri=${encodeURIComponent(goal)}` + return { args, redirect: toGoal(goal) } } case '/review': { - const goal = requireField(form, 'goal') - await runCli(['review', 'post', '--subject', requireField(form, 'subject'), '--verdict', requireField(form, 'verdict')], deps) - return `/goal?uri=${encodeURIComponent(goal)}` + const goal = require_('goal') + return { + args: ['review', 'post', '--subject', require_('subject'), '--verdict', require_('verdict')], + redirect: toGoal(goal), + } } case '/message': { - const goal = requireField(form, 'goal') - const args = ['message', 'post', '--goal', goal, '--body', requireField(form, 'body')] + const goal = require_('goal') + const args = ['message', 'post', '--goal', goal, '--body', require_('body')] if (optional('re')) args.push('--re', optional('re')) - await runCli(args, deps) - return `/goal?uri=${encodeURIComponent(goal)}` + return { args, redirect: toGoal(goal) } } default: throw new Error(`Unknown action: ${path}`) } } +/** Map a submitted form to the CLI command that writes it; return where to send the browser next. */ +async function performAction(path: string, form: URLSearchParams): Promise { + if (path === '/login') { + const identifier = requireField(form, 'identifier') + const profile = (form.get('profile') || identifier).trim() + const override = form.get('pds')?.trim() + const service = override || String(await resolveIdentityPds(identifier)) + const session = await createSession(service, identifier, requireField(form, 'password')) + await store.save({ ...session, profile }) + return '/' + } + + const deps = await depsFor(form.get('profile') ?? undefined) + const { args, redirect } = buildAction(path, form) + return redirect(await runCli(args, deps)) +} + async function renderGet(path: string, params: URLSearchParams): Promise { const profiles = await profileOptions() -- 2.51.2