/** * Run bookkeeping on bun:sqlite. * * The tracker's labels remain the source of truth for *what* the fleet is * doing; this store exists so the dispatcher can answer cap questions ("am I * already at two workers?", "have I spent $25 today?") without a round-trip to * GitHub on every tick, and so a restart can reconcile the runs it left * mid-flight. Losing the file is survivable — it is a cache with a memory, not * a ledger. */ import { Database } from "bun:sqlite"; import { createHash, randomUUID } from "node:crypto"; import { existsSync, mkdirSync, readdirSync, rmSync, unlinkSync } from "node:fs"; import { dirname, join } from "node:path"; import { backupTimestamp, publishTempFile } from "./backups.ts"; import { stateDir } from "./config.ts"; import { DECISION_TTL_MS, DEFAULT_CAPS, DIGEST_BACKLOG_LIMIT, TERMINAL_REVIEW_ADJUDICATION_STATES, } from "./types.ts"; /** * `expiresAt` for a watch (#459). It has no human deadline — a condition that * simply has not fired yet is a different thing from a question nobody * answered, and silently expiring it could drop a release. Far enough ahead * that `expireDueDecisions` (`expiresAt <= now`) never closes it. */ const DECISION_WATCH_NEVER_EXPIRES = Number.MAX_SAFE_INTEGER; import { dispatchInfra, neverStarted } from "./failure-class.ts"; import type { AdmissionHoldReason, BaseFreeze, BaseHealth, DecisionDraft, FailureClass, FileLane, DecisionRecord, DecisionState, DigestBacklog, DispatchSummary, EpicApproval, DaemonStop, DaemonStopDraft, FrictionAdmissionReason, FrictionKind, FrictionObservation, FrictionSignal, GraphToolsObservation, GroomingDraft, GroomingRecord, GroomingVerdict, HistoricalInfraCandidate, HandoffWithdrawal, HandoffWithdrawalResult, HeldNotice, InterruptCategory, HeldNoticeDraft, IntakeDraft, IntakeItem, InstallSurfaceObservation, IntakeState, MaterialEvent, MaterialEventDraft, LabelOp, MergeLock, ModelEscalation, OrchestratorDownMode, OrchestratorIncident, OrchestratorIncidentDraft, PromotedBy, ReleaseComposition, ReleaseCompositionOverride, ReportDeliveryState, ReportDraft, ReportEnqueue, ReportKind, ReportRecord, RecoveryAction, ReviewClearance, ReviewReason, ReviewAdjudicationAdmission, ReviewAdjudicationProvenance, ReviewAdjudicationRecord, ReviewAdjudicationState, ReviewCorrectionProvenance, ReviewLaunchMode, ReviewRevisionDraft, ReviewRevisionEnqueue, ReviewRevisionOutcome, ReviewRevisionRecord, ReviewRevisionRetry, RunPatch, RunRecord, RunState, SessionRole, SessionSpendRole, SessionSpendRow, SettlementFlag, Store, TurnOverride, VerbDecision, VerbLedgerDraft, VerbLedgerEntry, VerbName, VerbRefusal, } from "./types.ts"; /** * States backed by a worker process. These are what worker capacity counts: * a slot is a process, and only a claimed or running attempt has one. * * Exported because it is also the answer to "is anything still writing to this * run's transcript?" — `omp-conductor tail` needs that and must not re-derive * it, or the two definitions drift the first time a state is added. */ export const LIVE_STATES: readonly RunState[] = ["claimed", "running"]; /** * States that keep an *issue* occupied. Pending and green pushes belong here * but not in {@link LIVE_STATES}: their workers are finished and their * worktrees removed, so they must not consume slots, while their live PRs must * still block duplicate attempts. */ export const ACTIVE_STATES: readonly RunState[] = [...LIVE_STATES, "pushed-pending", "pushed-green"]; const LIVE_PLACEHOLDERS = LIVE_STATES.map(() => "?").join(", "); const ACTIVE_PLACEHOLDERS = ACTIVE_STATES.map(() => "?").join(", "); const FRICTION_HOLD_REASONS: ReadonlySet = new Set([ "failed-attempts", "continuations", "stale-base", "critical-base-verify-error", "parent-lookup-error", "issue-state-lookup-error", "open-pr-lookup-error", "unroutable:no-repo-label", "unroutable:multiple-repo-labels", "unroutable:unknown-repo", ]); const FRICTION_ISSUE_SAMPLES = 5; const FRICTION_TEXT_SAMPLES = 3; const FRICTION_TEXT_LIMIT = 160; function isFrictionHoldReason(reason: string): reason is FrictionAdmissionReason { return FRICTION_HOLD_REASONS.has(reason as FrictionAdmissionReason); } /** * The admission hold reasons the grooming store persists as `blocked` verdicts * (#735): the mechanical holds that will clear by themselves once the lane * frees or the dependency closes. Every other hold reason is flow control of * the moment (capacity, spend, parking) rather than durable per-issue * inventory, so admission's reconcile records and clears exactly these. */ const BLOCKED_GROOMING_HOLDS: Record = { "file-lane": true, "depends-on": true, }; /** * Allowlist for `updateRun`'s dynamic SET clause. Column names cannot be bound * as parameters, so they are matched against this table rather than * interpolated from caller input. `id` is deliberately absent: it is the * address of the row, not one of its fields. */ const UPDATABLE_COLUMNS: Record = { project: true, issue: true, repo: true, branch: true, worktree: true, state: true, attempt: true, turns: true, maxTurns: true, spendUsd: true, spendReservedUsd: true, provider429Count: true, resolvedModel: true, resolvedProvider: true, retryFallbacks: true, retryFallbackSucceeded: true, modelRecoveries: true, autoRetryCount: true, autoCompactionCount: true, sessionFile: true, workerPid: true, paneId: true, paneLabel: true, paneUnavailable: true, outputTokens: true, reasoningTokens: true, resumedFromRunId: true, lane: true, prUrl: true, headSha: true, mergeSha: true, baseRef: true, baseCheck: true, baseCheckAt: true, salvageSha: true, salvageError: true, salvageAckAt: true, quarantineDetail: true, startedAt: true, endedAt: true, lastError: true, settlementFlags: true, report: true, failureClass: true, recoveryAction: true, recoveredAt: true, model: true, graphTools: true, failureCharges: true, continuationCharges: true, terminalEvidence: true, lastProgressAt: true, }; /** Everything SQLite will accept from us. */ type SqlValue = string | number | null; /** The `runs` table exactly as SQLite hands it back: optional means NULL. */ interface RunRow { id: string; workerPid: number | null; paneId: string | null; paneLabel: string | null; paneUnavailable: string | null; outputTokens: number | null; reasoningTokens: number | null; project: string; issue: number; repo: string; branch: string; worktree: string; state: string; attempt: number; turns: number; maxTurns: number; spendUsd: number; /** NULL for a row admitted with no spend cap, or one written before #851. */ spendReservedUsd: number | null; provider429Count: number | null; resolvedModel: string | null; resolvedProvider: string | null; retryFallbacks: string | null; retryFallbackSucceeded: number | null; modelRecoveries: number | null; autoRetryCount: number | null; autoCompactionCount: number | null; sessionFile: string | null; resumedFromRunId: string | null; lane: string | null; prUrl: string | null; headSha: string | null; mergeSha: string | null; baseRef: string | null; baseCheck: string | null; baseCheckAt: number | null; salvageSha: string | null; salvageError: string | null; salvageAckAt: number | null; quarantineDetail: string | null; startedAt: number; endedAt: number | null; lastError: string | null; settlementFlags: string | null; report: string | null; terminalEvidence: string | null; failureClass: string | null; recoveryAction: string | null; recoveredAt: number | null; model: string | null; graphTools: string | null; failureCharges: number | null; continuationCharges: number | null; /** NULL for a row the progress pass has not observed, or one with no transcript yet (#1086). */ lastProgressAt: number | null; } /** The `base_health` table exactly as SQLite hands it back. */ interface BaseHealthRow { project: string; repo: string; branch: string; headSha: string; verdict: string; runsCount: number; detail: string | null; checkedAt: number; } function toBaseHealth(row: BaseHealthRow): BaseHealth { const health: BaseHealth = { repo: row.repo, branch: row.branch, headSha: row.headSha, verdict: row.verdict as BaseHealth["verdict"], runsCount: row.runsCount, checkedAt: row.checkedAt, }; if (row.detail !== null) health.detail = row.detail; return health; } /** The `base_freeze` table exactly as SQLite hands it back. */ interface BaseFreezeRow { project: string; repo: string; culpritSha: string; detail: string | null; setAt: number; clearedAt: number | null; clearedBy: string | null; clearedReason: string | null; } function toBaseFreeze(row: BaseFreezeRow): BaseFreeze { const freeze: BaseFreeze = { repo: row.repo, culpritSha: row.culpritSha, setAt: row.setAt, }; if (row.detail !== null) freeze.detail = row.detail; if (row.clearedAt !== null) { freeze.clearedAt = row.clearedAt; if (row.clearedBy !== null) freeze.clearedBy = row.clearedBy; if (row.clearedReason !== null) freeze.clearedReason = row.clearedReason; } return freeze; } /** The `release_composition` table exactly as SQLite hands it back (#850). */ interface ReleaseCompositionRow { project: string; campaign: string; /** JSON-encoded array of full PR URLs. */ allowedPrUrls: string; declaredBy: string | null; declaredAt: number; closedAt: number | null; closedBy: string | null; closedReason: string | null; } function toReleaseComposition(row: ReleaseCompositionRow): ReleaseComposition { const composition: ReleaseComposition = { campaign: row.campaign, allowedPrUrls: JSON.parse(row.allowedPrUrls) as string[], declaredAt: row.declaredAt, }; if (row.declaredBy !== null) composition.declaredBy = row.declaredBy; if (row.closedAt !== null) { composition.closedAt = row.closedAt; if (row.closedBy !== null) composition.closedBy = row.closedBy; if (row.closedReason !== null) composition.closedReason = row.closedReason; } return composition; } /** The `release_composition_overrides` table exactly as SQLite hands it back (#850). */ interface ReleaseCompositionOverrideRow { id: string; project: string; campaign: string; prUrl: string; at: number; by: string; reason: string; } interface FrictionRollupRow { project: string; day: string; kind: string; observations: number; occurrences: number; issues: string; samples: string; latestAt: number; } interface FrictionSurfaceRow { kind: string; at: number; } /** The `label_ops` table exactly as SQLite hands it back (#201). */ interface LabelOpRow { id: number; project: string; issue: number; op: "add" | "remove"; label: string; createdAt: number; attempts: number; nextAttemptAt: number; lastError: string | null; } /** NULL columns become absent properties, matching the other row converters. */ function toLabelOp(row: LabelOpRow): LabelOp { const op: LabelOp = { id: row.id, project: row.project, issue: row.issue, op: row.op, label: row.label, createdAt: row.createdAt, attempts: row.attempts, nextAttemptAt: row.nextAttemptAt, }; if (row.lastError !== null) op.lastError = row.lastError; return op; } /** The `reports` table exactly as SQLite hands it back. */ interface ReportRow { id: string; project: string; kind: string; body: string; state: string; attempts: number; attemptId: string | null; ambiguous: number; dedupeKey: string | null; createdAt: number; updatedAt: number; nextAttemptAt: number; messageId: number | null; messageIds: string | null; deliveredAt: number | null; sentParts: number; sentPartsHash: string | null; lastError: string | null; withdrawnAt: number | null; withdrawnBy: string | null; withdrawReason: string | null; } interface HandoffWithdrawalRow { id: string; project: string; target: string; actor: string; reason: string; at: number; summary: string; detail: string; } interface HeldNoticeLifecycleRow { id: string; project: string; summary: string; detail: string; digestedAt: number | null; digestReportId: string | null; withdrawnAt: number | null; } /** The `material_events` table exactly as SQLite hands it back (#274). */ interface MaterialEventRow { id: string; project: string; category: string; summary: string; evidence: string; occurredAt: number; recordedAt: number; digestReportId: string | null; } /** The `verb_ledger` table exactly as SQLite hands it back (#126). */ interface VerbLedgerRow { id: string; project: string; runId: string | null; issue: number | null; verb: string; role: string; args: string; decision: string; refusal: string | null; detail: string; sha: string | null; at: number; } /** The `merge_locks` table exactly as SQLite hands it back (#126). */ interface MergeLockRow { project: string; holder: string; prUrl: string; at: number; } /** The `review_revisions` table exactly as SQLite hands it back (#677). */ interface ReviewRevisionRow { id: string; project: string; runId: string; issue: number; prUrl: string; headSha: string; findings: string; round: number; reason: string; sessionFile: string | null; requestedAt: number; dispatchedAt: number | null; settledAt: number | null; outcome: string | null; /** Nullable rather than NOT NULL DEFAULT 0: rows written before #903 carry * NULL, which reads back as absent and counts as zero retries. */ infraRetries: number | null; /** The #1045 launch decision and provenance. All nullable, none backfilled: * a pre-#1045 row and a round nobody has decided yet are the same fact — * nobody wrote a decision down — and both must read back as unknown. */ launchMode: string | null; requestedModel: string | null; resolvedModel: string | null; originSessionRef: string | null; correctionSessionRef: string | null; launchDecidedAt: number | null; } /** The `model_escalations` table exactly as SQLite hands it back (#807). Every * column is NOT NULL, so the row is the record once `failureClass` is read * back as the enum the writer narrowed it from. */ interface ModelEscalationRow { model: string; failureClass: string; runId: string; at: number; } /** * The #1045 launch decision and provenance, projected off one review-revision * row. NULL becomes an absent property, so a round nobody has decided and a * row written before the columns existed both read back as {} — the explicit * unknown, never a fabricated `resume-original`. * * One helper, shared by the record converter and `reviewRevisionLaunch`, so * the shape a status snapshot renders can never disagree with the shape the * dispatcher reads. */ function toCorrectionProvenance(row: ReviewRevisionRow): ReviewCorrectionProvenance { const provenance: ReviewCorrectionProvenance = {}; if (row.launchMode !== null) provenance.launchMode = row.launchMode as ReviewLaunchMode; if (row.requestedModel !== null) provenance.requestedModel = row.requestedModel; if (row.resolvedModel !== null) provenance.resolvedModel = row.resolvedModel; if (row.originSessionRef !== null) provenance.originSessionRef = row.originSessionRef; if (row.correctionSessionRef !== null) provenance.correctionSessionRef = row.correctionSessionRef; return provenance; } /** * The one launch mode that has a correction session of its own (#1045). Bound * into the guarded UPDATE rather than written inline in SQL, so renaming the * union member is a compile error here instead of a WHERE clause that quietly * stops matching anything. */ const FRESH_CORRECTION: ReviewLaunchMode = "fresh-correction"; /** * NULL columns become absent properties, matching the other row converters: * a record read back out of the store deep-equals the one that went in. */ function toReviewRevision(row: ReviewRevisionRow): ReviewRevisionRecord { const record: ReviewRevisionRecord = { id: row.id, project: row.project, runId: row.runId, issue: row.issue, prUrl: row.prUrl, headSha: row.headSha, findings: row.findings, round: row.round, reason: row.reason as ReviewReason, requestedAt: row.requestedAt, // Every read path — pending, unsettled, per-PR, by id — hands the launch // provenance back on the row itself (#1045), so a caller never needs a // second query per round to know how the round was launched. ...toCorrectionProvenance(row), }; if (row.sessionFile !== null) record.sessionFile = row.sessionFile; if (row.dispatchedAt !== null) record.dispatchedAt = row.dispatchedAt; if (row.settledAt !== null) record.settledAt = row.settledAt; if (row.outcome !== null) record.outcome = row.outcome as ReviewRevisionOutcome; // Zero and NULL both read back as absent: a round nobody has retried and a // row written before the column existed are the same fact. if (row.infraRetries !== null && row.infraRetries > 0) record.infraRetries = row.infraRetries; if (row.launchDecidedAt !== null) record.launchDecidedAt = row.launchDecidedAt; return record; } /** * The `review_clearances` table exactly as SQLite hands it back (#913). Every * column is NOT NULL, so the row *is* the record — no converter needed. */ type ReviewClearanceRow = ReviewClearance; /** The `review_adjudications` table exactly as SQLite hands it back (#874). */ interface ReviewAdjudicationRow { id: string; project: string; issue: number; prUrl: string; headSha: string; role: string; provenance: string | null; state: string; evidence: string | null; disposition: string | null; requestedAt: number; dispatchedAt: number | null; settledAt: number | null; findings: string | null; } /** NULL columns become absent properties, like every other converter here, so * a record read back deep-equals the one that went in. Unparseable * provenance reads as absent rather than throwing: a corrupt JSON blob must * not make an otherwise-readable adjudication unreadable, and "no launch * recorded" is the honest degraded answer. */ function toReviewAdjudication(row: ReviewAdjudicationRow): ReviewAdjudicationRecord { const record: ReviewAdjudicationRecord = { id: row.id, project: row.project, issue: row.issue, prUrl: row.prUrl, headSha: row.headSha, role: row.role, state: row.state as ReviewAdjudicationState, requestedAt: row.requestedAt, }; if (row.provenance !== null) { try { record.provenance = JSON.parse(row.provenance) as ReviewAdjudicationProvenance; } catch { // left absent } } if (row.evidence !== null) record.evidence = row.evidence; if (row.disposition !== null) record.disposition = row.disposition; if (row.dispatchedAt !== null) record.dispatchedAt = row.dispatchedAt; if (row.settledAt !== null) record.settledAt = row.settledAt; if (row.findings !== null) record.findings = row.findings; return record; } /** The `intake_items` table exactly as SQLite hands it back (#299). */ interface IntakeRow { id: string; project: string; text: string; createdAt: number; state: string; issueUrl: string | null; groomedAt: number | null; source: string | null; } /** * NULL columns become absent properties, matching the other row converters: * a record read back out of the store deep-equals the one that went in. */ function toIntakeItem(row: IntakeRow): IntakeItem { const item: IntakeItem = { id: row.id, project: row.project, text: row.text, createdAt: row.createdAt, state: row.state as IntakeState, ...(row.issueUrl === null ? {} : { issueUrl: row.issueUrl }), ...(row.groomedAt === null ? {} : { groomedAt: row.groomedAt }), // Absent, not empty string: an operator's own idea has no provenance to // print, and "" would render as a machine-filed item with a blank source. ...(row.source === null ? {} : { source: row.source }), }; return item; } /** The `session_spend` table as SQLite hands it back (Phase 4). */ interface SessionSpendSqlRow { project: string; role: string; issue: number | null; model: string | null; resolvedModel: string | null; turns: number; spendUsd: number | null; at: number; } /** * Same NULL-becomes-absent posture as the converters above, and here it is * load-bearing rather than tidy: a NULL `spendUsd` must come back ABSENT, not * as `0`, because absent means the provider reported no cost and `0` means it * reported zero. Collapsing the two is exactly the ambiguity in the `runs` * table that forces `spend-telemetry.ts` to guess at a subscription verdict. */ function toSessionSpend(row: SessionSpendSqlRow): SessionSpendRow { return { project: row.project, role: row.role as SessionSpendRole, ...(row.issue === null ? {} : { issue: row.issue }), ...(row.model === null ? {} : { model: row.model }), ...(row.resolvedModel === null ? {} : { resolvedModel: row.resolvedModel }), turns: row.turns, ...(row.spendUsd === null ? {} : { spendUsd: row.spendUsd }), at: row.at, }; } /** The `grooming` table exactly as SQLite hands it back (#735). */ interface GroomingRow { project: string; issue: number; verdict: string; reason: string; evidence: string; recordedAt: number; promotedAt: number | null; promotedBy: string | null; } function toGrooming(row: GroomingRow): GroomingRecord { return { project: row.project, issue: row.issue, verdict: row.verdict as GroomingVerdict, reason: row.reason, evidence: row.evidence, recordedAt: row.recordedAt, // Both columns or neither: `markGroomingPromoted` writes them in one // statement, so a row carrying one without the other cannot arise. Read // off `promotedAt` alone rather than testing both, which would invent a // half-promoted state nothing can produce (#1041). ...(row.promotedAt === null ? {} : { promotedAt: row.promotedAt, promotedBy: (row.promotedBy ?? "operator") as PromotedBy }), }; } /** The `epic_approvals` table exactly as SQLite hands it back (#1041). */ interface EpicApprovalRow { issue: number; approvedAt: number; approvedBy: string; } /** The `orchestrator_incidents` table exactly as SQLite hands it back (#288). */ interface OrchestratorIncidentRow { project: string; mode: string; cause: string | null; since: number; diverted: number; } /** NULL columns become absent properties, matching the other row converters. */ function toOrchestratorIncident(row: OrchestratorIncidentRow): OrchestratorIncident { return { project: row.project, mode: row.mode as OrchestratorDownMode, ...(row.cause === null ? {} : { cause: row.cause }), since: row.since, diverted: row.diverted, }; } /** The `daemon_stops` table exactly as SQLite hands it back (#378). */ interface DaemonStopRow { id: string; at: number; controlPath: string; callerPid: number | null; callerUid: number | null; role: string | null; scope: string; project: string | null; daemonPid: number | null; runtimeDir: string | null; affected: string; reason: string; unattributed: number; } /** * NULL columns become absent properties, matching the other row converters: * a record read back out of the store deep-equals the one that went in. */ function toDaemonStop(row: DaemonStopRow): DaemonStop { return { id: row.id, at: row.at, controlPath: row.controlPath, ...(row.callerPid === null ? {} : { callerPid: row.callerPid }), ...(row.callerUid === null ? {} : { callerUid: row.callerUid }), ...(row.role === null ? {} : { role: row.role as SessionRole }), scope: row.scope as "global" | "project", ...(row.project === null ? {} : { project: row.project }), ...(row.daemonPid === null ? {} : { daemonPid: row.daemonPid }), ...(row.runtimeDir === null ? {} : { runtimeDir: row.runtimeDir }), affected: JSON.parse(row.affected) as { project: string; live: number }[], reason: row.reason, unattributed: row.unattributed === 1, }; } const SCHEMA = ` CREATE TABLE IF NOT EXISTS runs ( id TEXT PRIMARY KEY, project TEXT NOT NULL, issue INTEGER NOT NULL, repo TEXT NOT NULL, branch TEXT NOT NULL, worktree TEXT NOT NULL, state TEXT NOT NULL, attempt INTEGER NOT NULL, turns INTEGER NOT NULL, maxTurns INTEGER NOT NULL, spendUsd REAL NOT NULL, -- What admission reserved out of the daily budget for this run (#851). -- Nullable: an older row, or one admitted while the fleet had no cap. spendReservedUsd REAL, provider429Count INTEGER, resolvedModel TEXT, resolvedProvider TEXT, retryFallbacks TEXT, retryFallbackSucceeded INTEGER, modelRecoveries INTEGER, autoRetryCount INTEGER, autoCompactionCount INTEGER, sessionFile TEXT, -- The worker's exact process and workspace identity (#842). The pid is the -- session-host child the pane was reported for; the pane id and label are what -- Herdr knows it as. Nullable throughout: a historical row never had one, a -- run whose host has no Herdr never gets one, and NULL is the honest answer in -- both cases rather than a pane nobody can find. workerPid INTEGER, paneId TEXT, paneLabel TEXT, -- Why this run has no workspace representation (#841). Set when a pane could -- not be established or was lost, cleared when one is. Its presence beside a -- live run is what makes "degraded" a named state in status output rather than a -- silence the operator has to notice. paneUnavailable TEXT, -- Cumulative output and reasoning tokens (#518). Nullable: a historical row -- predates the accounting, and a provider that reports no token block leaves -- them absent rather than zero -- "reported nothing" and "produced nothing" -- invite opposite readings, and a zero would make the share silently wrong. outputTokens INTEGER, reasoningTokens INTEGER, resumedFromRunId TEXT, lane TEXT, prUrl TEXT, headSha TEXT, mergeSha TEXT, baseRef TEXT, baseCheck TEXT, baseCheckAt INTEGER, salvageSha TEXT, salvageError TEXT, salvageAckAt INTEGER, startedAt INTEGER NOT NULL, endedAt INTEGER, lastError TEXT, settlementFlags TEXT, -- The terminal evidence the settlement sweep recorded for this attempt -- (#986): exit path, first recoverable error line, audit flags. Nullable: -- rows written before the column, or not yet looked at by the sweep, carry -- NULL, which reads as "not recorded" — never "nothing was known". terminalEvidence TEXT, report TEXT, failureClass TEXT, recoveryAction TEXT, recoveredAt INTEGER, model TEXT, graphTools TEXT, failureCharges INTEGER, continuationCharges INTEGER, -- When the run's transcript was last observed to grow, as epoch ms (#1086). -- Written by the daemon's progress watch from the file's own mtime; NULL -- reads as "not observed yet", never "no progress". lastProgressAt INTEGER ); CREATE INDEX IF NOT EXISTS runs_project_issue ON runs (project, issue); CREATE INDEX IF NOT EXISTS runs_project_state ON runs (project, state); -- The historical infrastructure reconciliation's per-project review cursor -- (#638). A bounded pass examines settled ci-deterministic rows newest-first -- and persists the boundary of the last row it definitively decided, so older -- repairable rows are reached across later daemon starts instead of being -- starved by repeated rescans of the newest non-matches. classifierVersion is -- the signature-list fingerprint at the time the cursor advanced; when the -- classifier learns a new signature the stamp goes stale and the pass restarts -- from the newest row rather than skipping past newly recognised evidence. CREATE TABLE IF NOT EXISTS historical_infra_review ( project TEXT PRIMARY KEY, cursorStartedAt INTEGER NOT NULL, cursorRowid INTEGER NOT NULL, classifierVersion TEXT NOT NULL, updatedAt INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS base_health ( project TEXT NOT NULL, repo TEXT NOT NULL, branch TEXT NOT NULL, headSha TEXT NOT NULL, verdict TEXT NOT NULL, runsCount INTEGER NOT NULL, detail TEXT, checkedAt INTEGER NOT NULL, PRIMARY KEY (project, repo) ); -- A per-repo merge freeze (#283). While clearedAt IS NULL the repo is frozen: -- prMergeVerb refuses further merges to it with the base-red-freeze refusal. Set -- when a watched merge (or live base observation) turns the base red; the clear -- holds the mechanical green recovery and the operator's unfreeze verb, and acts -- as the durable override/clear audit ledger. CREATE TABLE IF NOT EXISTS base_freeze ( project TEXT NOT NULL, repo TEXT NOT NULL, culpritSha TEXT NOT NULL, detail TEXT, setAt INTEGER NOT NULL, clearedAt INTEGER, clearedBy TEXT, clearedReason TEXT, PRIMARY KEY (project, repo) ); -- The active release composition (#850): the campaign whose release a project -- is assembling plus the exact PRs allowed into it. While closedAt IS NULL, -- prMergeVerb refuses every other conductor_pr_merge with the -- outside-active-release refusal, so an early patch cannot accumulate -- unrelated merges; the clear columns hold the completion/cancellation audit -- on the row itself, and material_events carry the immutable lifecycle trail. CREATE TABLE IF NOT EXISTS release_composition ( project TEXT PRIMARY KEY, campaign TEXT NOT NULL, allowedPrUrls TEXT NOT NULL, declaredBy TEXT, declaredAt INTEGER NOT NULL, closedAt INTEGER, closedBy TEXT, closedReason TEXT ); -- Append-only operator overrides admitting exactly one PR into one campaign -- (#850). An override applies to its campaign only — it never outlives the -- release it was recorded for, and bypasses nothing but the composition check. CREATE TABLE IF NOT EXISTS release_composition_overrides ( id TEXT PRIMARY KEY, project TEXT NOT NULL, campaign TEXT NOT NULL, prUrl TEXT NOT NULL, at INTEGER NOT NULL, by TEXT NOT NULL, reason TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS release_composition_overrides_lookup ON release_composition_overrides (project, campaign, prUrl, at); CREATE TABLE IF NOT EXISTS notifications ( "key" TEXT PRIMARY KEY, at INTEGER NOT NULL ); -- A one-shot operator-selected ceiling for an issue's next claimed attempt. -- The run insert and consumption happen in one transaction. CREATE TABLE IF NOT EXISTS turn_overrides ( project TEXT NOT NULL, issue INTEGER NOT NULL CHECK (issue > 0), maxTurns INTEGER NOT NULL CHECK (maxTurns > 0), setAt INTEGER NOT NULL CHECK (setAt >= 0), PRIMARY KEY (project, issue) ); -- Append-only audit of every operator-selected ceiling. Consuming or replacing -- the pending value never erases what was requested. CREATE TABLE IF NOT EXISTS turn_override_ledger ( id TEXT PRIMARY KEY, project TEXT NOT NULL, issue INTEGER NOT NULL CHECK (issue > 0), maxTurns INTEGER NOT NULL CHECK (maxTurns > 0), setAt INTEGER NOT NULL CHECK (setAt >= 0) ); CREATE INDEX IF NOT EXISTS turn_override_ledger_project_at ON turn_override_ledger (project, setAt); -- The one-shot model escalation an issue chain has spent (#807). The primary -- key IS the cap: the insert either creates the chain's only escalation or -- changes nothing, so a settlement that crashes between claiming and queueing -- the continuation can never hand the same chain a second tier on restart, and -- a second spinning cap falls through to the existing decomposition verdict. -- Never deleted: the row is also what dispatch reads to keep the escalated -- chain on its stronger selector. CREATE TABLE IF NOT EXISTS model_escalations ( project TEXT NOT NULL, issue INTEGER NOT NULL CHECK (issue > 0), model TEXT NOT NULL, failureClass TEXT NOT NULL, runId TEXT NOT NULL, at INTEGER NOT NULL CHECK (at >= 0), PRIMARY KEY (project, issue) ); CREATE TABLE IF NOT EXISTS dispatch_summaries ( project TEXT PRIMARY KEY, summary TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS friction_rollups ( project TEXT NOT NULL, day TEXT NOT NULL, kind TEXT NOT NULL, observations INTEGER NOT NULL, occurrences INTEGER NOT NULL, issues TEXT NOT NULL, samples TEXT NOT NULL, latestAt INTEGER NOT NULL, PRIMARY KEY (project, day, kind) ); CREATE INDEX IF NOT EXISTS friction_rollups_project_day ON friction_rollups (project, day); CREATE TABLE IF NOT EXISTS friction_surfaces ( project TEXT NOT NULL, kind TEXT NOT NULL, at INTEGER NOT NULL, PRIMARY KEY (project, kind) ); -- The report outbox (#123). Deliberately NOT an extension of notifications: -- that table's primary key is its dedupe key, so it cannot hold two attempts of -- one report, and widening it would put escalation dedupe and report delivery -- on the same row. This one is keyed by report id and carries delivery state, -- the attempt in flight, and the message id Telegram actually returned. -- -- The partial unique index is the digest guard: a non-failed digest: key -- can exist at most once per project, so "has today's digest been handed over?" -- is a question the ledger answers rather than the model's memory of the last -- tick. A failed handoff can be replaced; material reports carry no key. CREATE TABLE IF NOT EXISTS reports ( id TEXT PRIMARY KEY, project TEXT NOT NULL, kind TEXT NOT NULL, body TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL, attemptId TEXT, ambiguous INTEGER NOT NULL, dedupeKey TEXT, createdAt INTEGER NOT NULL, updatedAt INTEGER NOT NULL, nextAttemptAt INTEGER NOT NULL, messageId INTEGER, messageIds TEXT, deliveredAt INTEGER, sentParts INTEGER NOT NULL DEFAULT 0, sentPartsHash TEXT, lastError TEXT, withdrawnAt INTEGER, withdrawnBy TEXT, withdrawReason TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS reports_dedupe ON reports (project, dedupeKey) WHERE dedupeKey IS NOT NULL AND state <> 'failed' AND state <> 'withdrawn'; CREATE INDEX IF NOT EXISTS reports_project_state ON reports (project, state, nextAttemptAt); CREATE TABLE IF NOT EXISTS handoff_withdrawals ( id TEXT PRIMARY KEY, project TEXT NOT NULL, target TEXT NOT NULL, actor TEXT NOT NULL, reason TEXT NOT NULL, at INTEGER NOT NULL, summary TEXT NOT NULL, detail TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS handoff_withdrawals_project_at ON handoff_withdrawals (project, at DESC); -- Ordinary material outcomes awaiting authorship in a deferred digest (#274). -- Association happens when a digest is accepted into the outbox, not when the -- session happens to remember the event or when Telegram later delivers it. CREATE TABLE IF NOT EXISTS material_events ( id TEXT PRIMARY KEY, project TEXT NOT NULL, category TEXT NOT NULL, summary TEXT NOT NULL, evidence TEXT NOT NULL, occurredAt INTEGER NOT NULL, recordedAt INTEGER NOT NULL, digestReportId TEXT REFERENCES reports(id) ); CREATE INDEX IF NOT EXISTS material_events_project_undigested ON material_events (project, occurredAt) WHERE digestReportId IS NULL; -- Interrupts a project's reporting.interruptOn policy deferred to the digest -- (#229): an escalation that must not page the operator's phone now is held -- here so the daily rollup can surface it. digestedAt set -> re-surfaced by a -- digest pass and no longer owed. A table rather than a flag on reports: these -- are escalations that never reached a send, distinct from reports the model -- wrote on purpose. CREATE TABLE IF NOT EXISTS held_notices ( id TEXT PRIMARY KEY, project TEXT NOT NULL, category TEXT NOT NULL, summary TEXT NOT NULL, detail TEXT NOT NULL, createdAt INTEGER NOT NULL, releaseOnAvailable INTEGER NOT NULL DEFAULT 0, urgent INTEGER NOT NULL DEFAULT 0, digestedAt INTEGER, digestReportId TEXT REFERENCES reports(id), withdrawnAt INTEGER, withdrawnBy TEXT, withdrawReason TEXT ); CREATE INDEX IF NOT EXISTS held_notices_project_undigested ON held_notices (project, digestedAt) WHERE digestedAt IS NULL; -- A due digest gets one bounded handoff lease per oldest availability notice -- and local day. The lease closes the snapshot-to-CLI race; expiry hands -- ownership back to the daemon catch-up instead of starving the notice. CREATE TABLE IF NOT EXISTS availability_digest_reservations ( project TEXT NOT NULL, cycleKey TEXT NOT NULL, firstNoticeId TEXT NOT NULL, expiresAt INTEGER NOT NULL, PRIMARY KEY (project, cycleKey, firstNoticeId) ); CREATE INDEX IF NOT EXISTS availability_digest_reservations_active ON availability_digest_reservations (project, expiresAt); -- The action ledger for the conductor-owned mutation verbs (#126). Every -- decided call lands here, refusals included: the question an escalation asks -- is "what did this run try", and a table that only records what succeeded -- cannot answer it. The 'args' column is the payload verbatim, so the row is -- evidence rather than a summary of evidence. CREATE TABLE IF NOT EXISTS verb_ledger ( id TEXT PRIMARY KEY, project TEXT NOT NULL, runId TEXT, issue INTEGER, verb TEXT NOT NULL, role TEXT NOT NULL, args TEXT NOT NULL, decision TEXT NOT NULL, refusal TEXT, detail TEXT NOT NULL, sha TEXT, at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS verb_ledger_project_at ON verb_ledger (project, at); CREATE INDEX IF NOT EXISTS verb_ledger_run ON verb_ledger (runId, at); -- One merge in flight per project (#126). A table rather than a mutex because -- the CLI, the daemon and every session are separate processes: an in-process -- lock serialises one process against itself and nothing else, which is the -- shape "one merge in flight" had on main -- none at all. The project is the -- primary key, so the claim is the row's existence and the race is SQLite's. CREATE TABLE IF NOT EXISTS merge_locks ( project TEXT PRIMARY KEY, holder TEXT NOT NULL, prUrl TEXT NOT NULL, at INTEGER NOT NULL ); -- One durable review-revision request (#677): an orchestrator returned a -- green, run-owned pull request to its worker with blocking findings. Every -- field the daemon needs to resume the exact session is persisted BEFORE the -- worker is woken -- the findings, the exact reviewed head, the round number, -- and the target run/session -- and the row doubles as the duplicate guard: a -- revision whose row exists pending is one nobody may queue again. CREATE TABLE IF NOT EXISTS review_revisions ( id TEXT PRIMARY KEY, project TEXT NOT NULL, runId TEXT NOT NULL, issue INTEGER NOT NULL, prUrl TEXT NOT NULL, headSha TEXT NOT NULL, findings TEXT NOT NULL, round INTEGER NOT NULL, reason TEXT NOT NULL, sessionFile TEXT, requestedAt INTEGER NOT NULL, dispatchedAt INTEGER, settledAt INTEGER, outcome TEXT, -- Infrastructure kills that returned this round to the pending set (#903). -- Nullable so an older database migrates by ALTER without a rewrite; NULL -- and 0 mean the same thing. infraRetries INTEGER, -- How this round was launched, and by which model and session (#1045). -- launchMode is 'resume-original' or 'fresh-correction' and is written -- ONCE per round, by the deciding dispatch, through a guarded UPDATE -- so a -- restart re-reads the decision instead of making a second one. All six are -- nullable and nothing is backfilled: a row from before #1045 recorded no -- decision, and inventing 'resume-original' for it would fabricate the one -- fact this provenance exists to make honest. launchMode TEXT, requestedModel TEXT, resolvedModel TEXT, originSessionRef TEXT, correctionSessionRef TEXT, launchDecidedAt INTEGER ); CREATE INDEX IF NOT EXISTS review_revisions_pending ON review_revisions (project, dispatchedAt, runId); -- One pending revision per run, enforced in the schema itself: the -- schema-level backstop for the enqueue transaction (#786). The transaction -- begins IMMEDIATE, so two connections serialize on the write lock before -- either reads — this index makes a second pending row for one run -- structurally impossible even if some other write path ever bypassed that -- transaction. CREATE UNIQUE INDEX IF NOT EXISTS review_revisions_one_pending_per_run ON review_revisions (runId) WHERE dispatchedAt IS NULL AND settledAt IS NULL; -- One recorded review clearance (#913): the explicit orchestrator disposition -- that the durable review findings standing at one exact head are settled. -- -- The #888 merge gate had no reachable clearance at an unchanged head. A -- re-review at the same head appends to (or opens) an unsettled row, so it -- blocks too, and nothing an orchestrator can call settles a round — a PR at -- the review-round ceiling was therefore permanently unmergeable at that head. -- This table is that missing disposition, kept deliberately separate from the -- merge call's own reason argument so the override is a recorded act with its -- own ledger row rather than a sentence in a merge. -- -- Append-only: a second clearance at the same head is another row, and its -- recorded timestamp is load-bearing -- the gate honours a clearance only for -- revision rows whose own last activity is at or before it, so a clearance can -- never bless a *later* finding at the same head. CREATE TABLE IF NOT EXISTS review_clearances ( id TEXT PRIMARY KEY, project TEXT NOT NULL, prUrl TEXT NOT NULL, headSha TEXT NOT NULL, at INTEGER NOT NULL, by TEXT NOT NULL, reason TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS review_clearances_lookup ON review_clearances (project, prUrl, headSha, at); -- One terminal review-ceiling adjudication (#874): the durable lifecycle that -- makes the final adjudication a recorded act rather than an in-memory fourth -- review round. -- -- The one-shot guarantee is the UNIQUE index below, not a flag a caller -- remembers to check: a repeated tick, two dispatchers, or a daemon restart all -- collide on (project, prUrl, headSha) and re-read the existing row. The exact -- head is part of that key on purpose — a verdict describes the diff it read, -- so it must never travel to a head nobody adjudicated. -- -- Deliberately NOT a row in review_revisions. Reusing round N+1 there would -- spend a worker review round the ceiling has already exhausted, and would -- resume the same session and model for a nominal fourth opinion, which is the -- failure #873 exists to end. The tables also answer different questions: a -- revision is "go and change this"; an adjudication is "decide about this". -- -- state carries the verdict: the terminal states ARE it (cleared, rejected, -- unavailable-model, stale-head, failed), so there is no second verdict column -- to disagree with them. evidence says why, and disposition (written by -- #876) says what was done about it. CREATE TABLE IF NOT EXISTS review_adjudications ( id TEXT PRIMARY KEY, project TEXT NOT NULL, issue INTEGER NOT NULL, prUrl TEXT NOT NULL, headSha TEXT NOT NULL, role TEXT NOT NULL, -- The resolved launch provenance (#875) as JSON, absent while pending: no -- model has resolved yet, and recording an intention as provenance would make -- status claim a launch that never happened. provenance TEXT, state TEXT NOT NULL, evidence TEXT, disposition TEXT, requestedAt INTEGER NOT NULL, dispatchedAt INTEGER, settledAt INTEGER, -- The findings that provoked the escalation (#932): what the orchestrator was -- submitting when it hit the ceiling. Nullable so an older database migrates -- by ALTER, and because an adjudication opened by any other path has none. -- Kept here rather than as a review_revisions round because a round would -- spend a ceiling that is already exhausted and would wake a worker; this is -- evidence for the adjudicator, not work for an implementer. findings TEXT ); -- The one-shot identity, structural rather than advisory: at most one -- adjudication per PR head, ever. CREATE UNIQUE INDEX IF NOT EXISTS review_adjudications_one_per_head ON review_adjudications (project, prUrl, headSha); -- Restart recovery and the status projection both read the non-terminal rows -- for one project; the PR lookup answers "is this head adjudicated". CREATE INDEX IF NOT EXISTS review_adjudications_open ON review_adjudications (project, settledAt); -- The install identities the host carried when the daemon last looked (#919). -- One row (the primary key is a constant): the question is what is installed -- NOW, and the install history is the upgrade journal's job. Recorded by the -- dispatch pass because reading it costs three subprocesses -- doing that at -- render time took the tick suite from 8.4s to 83.4s and would spawn three -- children every fifteen minutes forever. CREATE TABLE IF NOT EXISTS install_surfaces ( host TEXT PRIMARY KEY, at INTEGER NOT NULL, cliVersion TEXT NOT NULL, ompVersion TEXT, herdrSource TEXT, -- The one peer plugin a documented conductor contract runs on (#961). Carried -- on this row rather than in a table of its own because it is read by the same -- periodic pass, about the same host, for the same reason: an install that -- diverged from what is running or published, observed without an on-demand -- doctor run (#919). Every column is nullable — a pass that could not reach -- the registry records what it did read and leaves the rest unobserved, which -- the renderer must show as "unverified" and never as agreement. tgInstalled TEXT, tgDaemon TEXT, tgPublished TEXT ); -- Questions the orchestrator has put to its operator, and their answers (#136). -- -- The reason this is a table and not the session's memory: a question lived only -- in the model's context, so a compaction, a restart or a tick that ran long -- lost both the question and the fact that it was owed one. The session then -- either re-asked (the operator answers twice) or silently dropped it (the -- decision never lands, and nothing anywhere says a decision is outstanding). -- -- The 'condition' column is a machine-checkable precondition for the answer -- becoming actionable -- a PR merging, an issue closing, a version on npm -- -- stored as the raw string an operator or session wrote, parsed on read. Once -- met, 'conditionMetAt' is what turns a parked question into one the tick digest -- pushes at the session. -- -- 'conditionHead' binds a pr-checks-green / pr-review-ready verdict to the -- exact PR head it was observed at (#808). The checks surface answers per -- commit, so once the head changes the old verdict says nothing about the new -- head: the evaluator clears the met state on a head move and the row returns -- to pending. CREATE TABLE IF NOT EXISTS decisions ( id TEXT PRIMARY KEY, project TEXT NOT NULL, kind TEXT NOT NULL DEFAULT 'question', question TEXT NOT NULL, blocks TEXT, askedAt INTEGER NOT NULL, expiresAt INTEGER NOT NULL, condition TEXT, conditionMetAt INTEGER, conditionHead TEXT, state TEXT NOT NULL, resolvedAt INTEGER, resolution TEXT, -- One spec-out questionnaire (#947): N rows written in a single transaction, -- delivered as one message, resolved independently and in any order. Nullable -- because every ordinary single ask has neither: a historical row was never -- part of a group and must not read as a group of one, and specIssue is what -- makes a group's answers the spec's provenance rather than chat history. groupId TEXT, specIssue INTEGER ); CREATE INDEX IF NOT EXISTS decisions_project_state ON decisions (project, state); -- The groupId index is NOT here on purpose: this whole script runs before the -- additive migrations, and on a database written before #947 the column does not -- exist yet, so an index over it here fails the entire schema step. It is created -- beside the ALTER that adds the column, which every open path runs. -- One issue's current grooming verdict (#735), the durable answer to "has -- this been groomed, and what was decided" that the tick reads instead of -- re-deriving why a runway cannot move. Admission's lane/dependency holds land -- here as blocked (self-clearing on the next pass when the hold no longer -- applies); #679's scout loop writes promotable and considered into the -- same table. Keyed by project + issue, one row per issue, replaced in place. -- Deliberately separate from decisions (operator questions, 7-day expiry) and -- material_events (append-only digest outbox). CREATE TABLE IF NOT EXISTS grooming ( project TEXT NOT NULL, issue INTEGER NOT NULL CHECK (issue > 0), verdict TEXT NOT NULL, reason TEXT NOT NULL, evidence TEXT NOT NULL, recordedAt INTEGER NOT NULL CHECK (recordedAt >= 0), -- Promotion provenance (#1041): when this verdict was acted on by queueing -- the issue, and which of the three actors did it. Nullable because most -- verdicts are never promoted, and because a row written before mechanical -- promotion existed recorded nobody -- NULL is the only honest reading. -- Written once per verdict and cleared by the next upsert: a new verdict is a -- new judgement, and provenance outliving its judgement audits the wrong one. promotedAt INTEGER, promotedBy TEXT, PRIMARY KEY (project, issue) ); CREATE INDEX IF NOT EXISTS grooming_project_verdict ON grooming (project, verdict); -- The promoted index is NOT here, for the same reason decisions_group is not: -- this script runs before the additive migrations, so on a database written -- before #1041 the column does not exist yet and an index over it would fail -- the whole schema step. Created beside its ALTER below. -- The operator's standing approvals of epic scope (#1041). One row per approved -- epic, first write wins, revocable. -- -- Its own table rather than a resolved decision row, and the reason is that a -- gate has to read it: a decision's resolution column is free text written by a -- human for a human, and no parser turns "yes, but do the migration first" into -- a boolean. Decision rows also expire after DECISION_TTL_MS, and scope consent -- must not lapse because nobody looked at it for a week. Project-scoped like -- every other table here, so one project's approvals can never gate another's -- children. CREATE TABLE IF NOT EXISTS epic_approvals ( project TEXT NOT NULL, issue INTEGER NOT NULL CHECK (issue > 0), approvedAt INTEGER NOT NULL CHECK (approvedAt >= 0), approvedBy TEXT NOT NULL, PRIMARY KEY (project, issue) ); -- GitHub rate-limit refusals the tracker observed (#198). Written by the -- daemon's tracker hook, not polled, so status can show what GitHub actually -- refused next to the polled budget that might still look healthy. Only one -- column: a refusal carries no id or dedupe key — rows are pruned by age -- (24h) in the same write that inserts, kept long enough for status's 5m -- window and for an operator who wants to see the last day of refusals. CREATE TABLE IF NOT EXISTS gh_refusals ( at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS gh_refusals_at ON gh_refusals (at); -- Auto-restarts the daemon fired at a wedged orchestrator (#1049, Phase 4). -- The same windowed-event shape as gh_refusals above and for the same reason: -- the only question anyone asks is "how many in the last 6h", so a bare row -- per restart is the whole record and there is no cursor, no latch and no -- counter to corrupt. A crash between the restart and the next read costs at -- worst one row, never a mis-set high-water mark that silently disables the -- cap or spends it twice. -- -- Project-scoped, unlike gh_refusals: a rate-limit refusal is a property of -- the host's GitHub token, but a wedged orchestrator belongs to exactly one -- project, and one project burning its three restarts must never consume -- another project's budget. -- -- reason is NOT NULL because the whole value of the row past the count is the -- page it produces when the cap is reached: "3 restarts in 6h, all -- stall-marker" reads very differently from three different causes. CREATE TABLE IF NOT EXISTS orchestrator_restarts ( project TEXT NOT NULL, at INTEGER NOT NULL CHECK (at >= 0), reason TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS orchestrator_restarts_project_at ON orchestrator_restarts (project, at); -- Spend and turns for the sessions the DAEMON owns rather than dispatches -- (Phase 4 attribution). Today that is to-spec grooming and review -- adjudication: both are real model sessions the fleet pays for, and both -- were previously logged to stdout and then forgotten, so a fleet's stats -- reported only the worker half of its bill. -- -- Deliberately NOT the runs table. computeStats walks runs as issue journeys -- keyed on run.issue and folds FAILED_STATES into merged/settled counts, so a -- groom row parked in there would invent an extra journey for an issue that -- never had a worker, corrupt the settled counts, and skew $-per-merge by -- attributing grooming spend to a merge it did not produce. A small dedicated -- table is the honest shape. -- -- spendUsd is NULLABLE, and that is the point. In runs, an unmetered request -- and a genuinely free one are both stored as 0, which is why -- spend-telemetry.ts has to infer a "subscription" verdict from the SHARE of -- zeros (SPEND_MISSING_SHARE) instead of reading a fact. Here the two are -- distinguishable at the row: NULL means the provider reported no cost for -- this session (subscription billing, or a harness that emitted no usage -- event), 0.0 means it reported exactly zero. A reader must render NULL as -- unmetered, never as $0.00 — a fabricated zero reads as free. -- -- model is what the caller asked for and resolvedModel is what the harness -- actually ran; both nullable because a caller that did not observe either -- must not have one guessed for it. The per-model breakdown groups on -- resolvedModel, since that is the model that was billed. CREATE TABLE IF NOT EXISTS session_spend ( id INTEGER PRIMARY KEY AUTOINCREMENT, project TEXT NOT NULL, role TEXT NOT NULL CHECK (role IN ('groom', 'adjudicator')), issue INTEGER CHECK (issue IS NULL OR issue > 0), model TEXT, resolvedModel TEXT, turns INTEGER NOT NULL CHECK (turns >= 0), spendUsd REAL, at INTEGER NOT NULL CHECK (at >= 0) ); CREATE INDEX IF NOT EXISTS session_spend_project_at ON session_spend (project, at); -- The daemon's tracked gh call count, per UTC day and call source (#198). -- The single funnel in tracker/github.ts counts every spawn; the day/source -- pair is the partition, so today's column and today's row survive a restart. CREATE TABLE IF NOT EXISTS gh_calls ( day TEXT NOT NULL, source TEXT NOT NULL, calls INTEGER NOT NULL, PRIMARY KEY (day, source) ); -- The label projection outbox (#201). Every decided GitHub label change the -- dispatcher made is a row here until the tracker has applied it; the daemon -- retries with backoff, and eligibility reads pendingLabelOpsFor so a label -- GitHub refuses to move cannot strand the issue the daemon already decided -- to move on. The queue-label add and the state-label remove of one swap are -- two rows, applied strictly in id order, which is what makes a swap atomic: -- the remove cannot land before the add it was enqueued after. CREATE TABLE IF NOT EXISTS label_ops ( id INTEGER PRIMARY KEY AUTOINCREMENT, project TEXT NOT NULL, issue INTEGER NOT NULL, op TEXT NOT NULL CHECK (op IN ('add', 'remove')), label TEXT NOT NULL, createdAt INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, nextAttemptAt INTEGER NOT NULL DEFAULT 0, lastError TEXT ); CREATE INDEX IF NOT EXISTS label_ops_project_next ON label_ops (project, nextAttemptAt); -- Raw ideas captured by the intake command (#299), held durably until -- #300 resolves them into issues. A thought from 02:00 must survive a restart -- and a reinstall, so it lives here rather than in any process's memory. The -- state CHECK keeps the three-state vocabulary exact; issueUrl and groomedAt -- stay NULL until an idea becomes (or is dropped instead of) an issue. -- -- source is the provenance of a machine-filed item (Phase 4 signal mining): -- the stable key of the signal that produced it, e.g. -- "mined:settlement-weakening:omp/src/foo.ts". NULL means an operator typed -- this idea themselves, which is the only other way a row gets here. It is -- also the dedupe key: recordIntake derives a DETERMINISTIC id from -- project + source, so the hourly mining pass re-filing a signal it already -- filed is a no-op instead of an hourly duplicate. Declared here for fresh -- databases and added by a guarded ALTER for existing ones, with no backfill: -- every row written before mining existed was typed by the operator, and -- absent is exactly that. CREATE TABLE IF NOT EXISTS intake_items ( id TEXT PRIMARY KEY, project TEXT NOT NULL, text TEXT NOT NULL, createdAt INTEGER NOT NULL, state TEXT NOT NULL CHECK (state IN ('pending', 'groomed', 'dismissed')), issueUrl TEXT, groomedAt INTEGER, source TEXT ); CREATE INDEX IF NOT EXISTS intake_items_project_state ON intake_items (project, state, createdAt); -- The embedded orchestrator's lost-liveness incident (#288): durable so a -- daemon restarted while its orchestrator is still down rediscovers the open -- incident instead of forgetting it, and the diverted counter accumulates -- across the outage and across restarts. One row per project; removed on -- recovery. since is the dedupe anchor for both the down page and its -- closing recovery page. The mode CHECK keeps the two causes distinct. CREATE TABLE IF NOT EXISTS orchestrator_incidents ( project TEXT PRIMARY KEY, mode TEXT NOT NULL CHECK (mode IN ('start-failed', 'crashed')), cause TEXT, since INTEGER NOT NULL, diverted INTEGER NOT NULL DEFAULT 0 ); -- Conductor-owned Herdr worker workspaces (#1035 review). Herdr restores a -- session's workspaces and panes across a server restart but drops -- report-metadata tokens, so the live-only token cannot remain the only -- ownership authority: the restored surface would be invisible to discovery, -- a duplicate would be created beside it, and its stale panes could never be -- reconciled or removed. One row per workspace conductor created and still -- owns; forgotten when the workspace is removed or fails to create cleanly. CREATE TABLE IF NOT EXISTS herdr_worker_workspaces ( project TEXT NOT NULL, workspace_id TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (project, workspace_id) ); -- Every stop/restart of the shared daemon, and who asked for it and why -- (#378). Deliberately NOT partitioned by project: the daemon serves every -- configured project, so a record written by one project's CLI must be -- readable from another project's status the moment the daemon is down — -- that cross-project read is the whole point. Written before signalling on -- the mediated paths (cli stop/restart) and as an honest unattributed -- fallback when the daemon receives a signal with no mediated request. -- Explicit fields only: never raw argv, environment values or credentials. CREATE TABLE IF NOT EXISTS daemon_stops ( id TEXT PRIMARY KEY, at INTEGER NOT NULL, controlPath TEXT NOT NULL, callerPid INTEGER, callerUid INTEGER, role TEXT, scope TEXT NOT NULL, project TEXT, daemonPid INTEGER, runtimeDir TEXT, affected TEXT NOT NULL, reason TEXT NOT NULL, unattributed INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS daemon_stops_at ON daemon_stops (at); `; /** * `undefined` is not a legal binding, and a boolean is only accepted by some * bun:sqlite builds, so both are normalised before they reach the driver. */ function toSql(value: unknown): SqlValue { if (value === undefined || value === null) return null; if (typeof value === "boolean") return value ? 1 : 0; // `RunRecord.settlementFlags` and `RunRecord.lane` are the structured fields // on a run row, and each is JSON in one TEXT column rather than a table of // its own: the flags are 1:1 with the run, written once at settlement, and // never read except beside the row they belong to (#128); the lane is the // enforced declaration, written once at dispatch and read by the admission // gate beside the row that owns it (#744). A join per read would buy a query // nobody makes. if (typeof value === "object") return JSON.stringify(value); return value as SqlValue; } /** * Settlement flags out of their column, or undefined for anything that is not * an array of flag-shaped objects. * * Same posture as {@link toDispatchSummary}: this is advisory evidence, so a * row hand-edited or written by a future version drops its flags rather than * throwing and taking `status` down with it. */ /** * Within-run fallback pairs out of their JSON column, or undefined for * anything that is not an array of from/to string pairs. Same defensive * posture as {@link toSettlementFlags}: a row hand-edited or written by a * future version drops its fallbacks rather than throwing. */ function toRetryFallbacks(text: string): { from: string; to: string }[] | undefined { try { const value: unknown = JSON.parse(text); if (!Array.isArray(value) || value.length === 0) return undefined; const pairs = value.filter( (pair): pair is { from: string; to: string } => typeof pair === "object" && pair !== null && typeof (pair as { from?: unknown }).from === "string" && typeof (pair as { to?: unknown }).to === "string", ); return pairs.length === 0 ? undefined : pairs; } catch { return undefined; } } function toSettlementFlags(text: string): SettlementFlag[] | undefined { try { const value: unknown = JSON.parse(text); if (!Array.isArray(value) || value.length === 0) return undefined; const flags = value.filter( (flag): flag is SettlementFlag => typeof flag === "object" && flag !== null && typeof (flag as SettlementFlag).kind === "string" && typeof (flag as SettlementFlag).file === "string" && typeof (flag as SettlementFlag).detail === "string", ); return flags.length === 0 ? undefined : flags; } catch { return undefined; } } /** * The persisted file lane out of its JSON column, or undefined for anything * that is not a `FileLane`-shaped object. Same defensive posture as * {@link toSettlementFlags}: a row hand-edited or written by a future version * drops its lane rather than throwing, and a dropped lane reads as "no * declaration" — the fail-open admission, never a refusal (#744). */ function toLane(text: string): FileLane | undefined { try { const value: unknown = JSON.parse(text); const files = (value as { files?: unknown }).files; const source = (value as { source?: unknown }).source; const at = (value as { at?: unknown }).at; if ( typeof value !== "object" || value === null || !Array.isArray(files) || files.some((file) => typeof file !== "string") || typeof source !== "string" || (at !== "body" && typeof at !== "number") ) { return undefined; } return value as FileLane; } catch { return undefined; } } /** * A run's graph-tools session observation out of its JSON column, or * undefined for anything that is not a `{ present, at }` object. Same * defensive posture as {@link toSettlementFlags}: a row hand-edited or written * by a future version drops its observation rather than throwing, and a * dropped observation reads as "no observation recorded" — never "graph tools * absent", which is the `present: false` truth value (#726). */ function toGraphTools(text: string): GraphToolsObservation | undefined { try { const value: unknown = JSON.parse(text); const present = (value as { present?: unknown } | null | undefined)?.present; const at = (value as { at?: unknown } | null | undefined)?.at; if (typeof present !== "boolean" || typeof at !== "number") return undefined; return { present, at }; } catch { return undefined; } } /** * A NULL column becomes an absent property rather than an `undefined` one, so * a record read back out of the store deep-equals the one that went in. */ function toRecord(row: RunRow): RunRecord { const record: RunRecord = { id: row.id, project: row.project, issue: row.issue, repo: row.repo, branch: row.branch, worktree: row.worktree, state: row.state as RunState, attempt: row.attempt, turns: row.turns, maxTurns: row.maxTurns, spendUsd: row.spendUsd, startedAt: row.startedAt, }; if (row.spendReservedUsd !== null) record.spendReservedUsd = row.spendReservedUsd; if (row.sessionFile !== null) record.sessionFile = row.sessionFile; if (row.lastProgressAt !== null) record.lastProgressAt = row.lastProgressAt; if (row.workerPid !== null) record.workerPid = row.workerPid; if (row.paneId !== null) record.paneId = row.paneId; if (row.paneLabel !== null) record.paneLabel = row.paneLabel; if (row.paneUnavailable !== null) record.paneUnavailable = row.paneUnavailable; if (row.outputTokens !== null) record.outputTokens = row.outputTokens; if (row.reasoningTokens !== null) record.reasoningTokens = row.reasoningTokens; if (row.resumedFromRunId !== null) record.resumedFromRunId = row.resumedFromRunId; if (row.lane !== null) { const lane = toLane(row.lane); if (lane !== undefined) record.lane = lane; } if (row.prUrl !== null) record.prUrl = row.prUrl; if (row.headSha !== null) record.headSha = row.headSha; if (row.mergeSha !== null) record.mergeSha = row.mergeSha; if (row.baseRef !== null) record.baseRef = row.baseRef; if (row.baseCheck !== null) record.baseCheck = row.baseCheck as RunRecord["baseCheck"]; if (row.baseCheckAt !== null) record.baseCheckAt = row.baseCheckAt; if (row.salvageSha !== null) record.salvageSha = row.salvageSha; if (row.salvageError !== null) record.salvageError = row.salvageError; if (row.salvageAckAt !== null) record.salvageAckAt = row.salvageAckAt; if (row.quarantineDetail !== null) record.quarantineDetail = row.quarantineDetail; if (row.endedAt !== null) record.endedAt = row.endedAt; if (row.lastError !== null) record.lastError = row.lastError; if (row.settlementFlags !== null) { const flags = toSettlementFlags(row.settlementFlags); if (flags !== undefined) record.settlementFlags = flags; } if (row.report !== null) record.report = row.report; if (row.terminalEvidence !== null) record.terminalEvidence = row.terminalEvidence; if (row.provider429Count !== null) record.provider429Count = row.provider429Count; if (row.resolvedModel !== null) record.resolvedModel = row.resolvedModel; if (row.resolvedProvider !== null) record.resolvedProvider = row.resolvedProvider; if (row.retryFallbacks !== null) { const fallbacks = toRetryFallbacks(row.retryFallbacks); if (fallbacks !== undefined) record.retryFallbacks = fallbacks; } if (row.retryFallbackSucceeded !== null) record.retryFallbackSucceeded = row.retryFallbackSucceeded; if (row.modelRecoveries !== null) record.modelRecoveries = row.modelRecoveries; if (row.autoRetryCount !== null) record.autoRetryCount = row.autoRetryCount; if (row.autoCompactionCount !== null) record.autoCompactionCount = row.autoCompactionCount; if (row.failureClass !== null) record.failureClass = row.failureClass as FailureClass; if (row.recoveryAction !== null) record.recoveryAction = row.recoveryAction as RecoveryAction; if (row.recoveredAt !== null) record.recoveredAt = row.recoveredAt; if (row.failureCharges !== null && row.failureCharges > 0) record.failureCharges = row.failureCharges; if (row.continuationCharges !== null && row.continuationCharges > 0) { record.continuationCharges = row.continuationCharges; } if (row.model !== null) record.model = row.model; if (row.graphTools !== null) { const observed = toGraphTools(row.graphTools); if (observed !== undefined) record.graphTools = observed; } return record; } /** * NULL columns become absent properties, matching {@link toRecord}: a report * read back out of the store deep-equals the one that went in. */ function toReport(row: ReportRow): ReportRecord { const record: ReportRecord = { id: row.id, project: row.project, kind: row.kind as ReportKind, body: row.body, state: row.state as ReportDeliveryState, attempts: row.attempts, ambiguous: row.ambiguous !== 0, createdAt: row.createdAt, updatedAt: row.updatedAt, nextAttemptAt: row.nextAttemptAt, }; if (row.attemptId !== null) record.attemptId = row.attemptId; if (row.dedupeKey !== null) record.dedupeKey = row.dedupeKey; if (row.messageId !== null) record.messageId = row.messageId; if (row.messageIds !== null) { // Written as JSON by this store; a row that fails to parse reads as if // no ids were recorded, which is the honest reading of unreadable bytes. try { const ids: unknown = JSON.parse(row.messageIds); if (Array.isArray(ids) && ids.every((id) => typeof id === "number")) record.messageIds = ids; } catch { // treated as absent } } if (row.sentParts !== 0) record.sentParts = row.sentParts; if (row.sentPartsHash !== null) record.sentPartsHash = row.sentPartsHash; if (row.deliveredAt !== null) record.deliveredAt = row.deliveredAt; if (row.lastError !== null) record.lastError = row.lastError; if (row.withdrawnAt !== null) record.withdrawnAt = row.withdrawnAt; if (row.withdrawnBy !== null) record.withdrawnBy = row.withdrawnBy; if (row.withdrawReason !== null) record.withdrawReason = row.withdrawReason; return record; } function toHandoffWithdrawal(row: HandoffWithdrawalRow): HandoffWithdrawal { return { id: row.id, project: row.project, summary: row.summary, detail: row.detail, target: row.target as HandoffWithdrawal["target"], actor: row.actor, reason: row.reason, at: row.at, }; } function toMaterialEvent(row: MaterialEventRow): MaterialEvent { return { id: row.id, project: row.project, category: row.category, summary: row.summary, evidence: row.evidence, occurredAt: row.occurredAt, recordedAt: row.recordedAt, ...(row.digestReportId === null ? {} : { digestReportId: row.digestReportId }), }; } interface DecisionRow { id: string; project: string; kind: string; question: string; blocks: string | null; askedAt: number; expiresAt: number; condition: string | null; conditionMetAt: number | null; conditionHead: string | null; state: string; resolvedAt: number | null; resolution: string | null; groupId: string | null; specIssue: number | null; } /** Same NULL-to-absent contract as {@link toReport}. */ function toDecision(row: DecisionRow): DecisionRecord { const record: DecisionRecord = { id: row.id, project: row.project, // Only `watch` is ever a watch: anything else — and especially the historic // NULL the migration backfilled as `question` — reads as a human question. // A real question may carry a condition; that must not relabel it. kind: row.kind === "watch" ? "watch" : "question", question: row.question, askedAt: row.askedAt, expiresAt: row.expiresAt, state: row.state as DecisionState, }; if (row.blocks !== null) record.blocks = row.blocks; if (row.condition !== null) record.condition = row.condition; if (row.conditionMetAt !== null) record.conditionMetAt = row.conditionMetAt; if (row.conditionHead !== null) record.conditionHead = row.conditionHead; if (row.resolvedAt !== null) record.resolvedAt = row.resolvedAt; if (row.resolution !== null) record.resolution = row.resolution; if (row.groupId !== null) record.groupId = row.groupId; if (row.specIssue !== null) record.specIssue = row.specIssue; return record; } /** * Long enough that a collision is not a thing anyone has to handle, short * enough that an operator can compare it against the id printed in a Telegram * message by eye — which is the only reconciliation this transport allows. */ function newReportId(): string { return crypto.randomUUID().replaceAll("-", "").slice(0, 12); } /** * The id of a machine-filed intake item (Phase 4): a pure function of the * project and the signal's source key, so the hourly mining pass re-deriving * it lands on the row it already wrote. * * Same width and alphabet as {@link newReportId} on purpose. The id is the * handle an operator types into `intake dismiss`, it is printed in a column * next to other ids, and nothing downstream should have to care whether a row * came from a machine — the `source` column says that, in the one place that * asks. 48 bits over a per-project namespace of at most a few thousand mined * signals is not a collision anyone has to handle. * * The project is part of the digest, not just a prefix, so two projects mining * the same file path never share a row. */ function minedIntakeId(project: string, source: string): string { return createHash("sha256").update(`${project}\u0000${source}`).digest("hex").slice(0, 12); } /** * A ledger row, with the same NULL-becomes-absent posture as {@link toReport}. * * `args` is stored as JSON and read back defensively: a row whose payload no * longer parses is still a row that says which verb was called and how it was * decided, and losing the whole entry to a malformed blob would erase exactly * the call somebody is trying to audit. */ function toVerbLedgerEntry(row: VerbLedgerRow): VerbLedgerEntry { let args: Record = {}; try { const parsed: unknown = JSON.parse(row.args); if (parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)) { args = parsed as Record; } } catch { args = { "(unreadable)": row.args }; } const entry: VerbLedgerEntry = { id: row.id, project: row.project, verb: row.verb as VerbName, role: row.role as SessionRole, args, decision: row.decision as VerbDecision, detail: row.detail, at: row.at, }; if (row.runId !== null) entry.runId = row.runId; if (row.issue !== null) entry.issue = row.issue; if (row.refusal !== null) entry.refusal = row.refusal as VerbRefusal; if (row.sha !== null) entry.sha = row.sha; return entry; } function toDispatchSummary(text: string): DispatchSummary | undefined { try { const value = JSON.parse(text) as DispatchSummary; const counts = [value.completedAt, value.ready, value.routed, value.admitted]; if ( !counts.every((count) => Number.isSafeInteger(count) && count >= 0) || typeof value.degraded !== "boolean" || !Array.isArray(value.holds) || value.holds.some( (hold) => typeof hold !== "object" || hold === null || typeof hold.reason !== "string" || !Number.isSafeInteger(hold.count) || hold.count < 0 || !Array.isArray(hold.issues) || hold.issues.length > 5 || hold.count < hold.issues.length || !hold.issues.every((issue) => Number.isSafeInteger(issue) && issue > 0) || (hold.details !== undefined && (!Array.isArray(hold.details) || hold.details.length > 5 || !hold.details.every((detail) => typeof detail === "string" && detail.length > 0))), ) ) { return undefined; } return value; } catch { return undefined; } } function parseNumberList(text: string): number[] { try { const value = JSON.parse(text) as unknown; return Array.isArray(value) ? value.filter((item): item is number => Number.isSafeInteger(item) && item > 0) : []; } catch { return []; } } function parseTextList(text: string): string[] { try { const value = JSON.parse(text) as unknown; return Array.isArray(value) ? value.filter((item): item is string => typeof item === "string" && item.length > 0) : []; } catch { return []; } } function boundedUnique(current: readonly T[], additions: readonly T[], limit: number): T[] { return [...new Set([...current, ...additions])].slice(0, limit); } function isFrictionKind(value: string): value is FrictionKind { if (value.startsWith("admission:")) { return isFrictionHoldReason(value.slice("admission:".length)); } return ( value === "feedback:escalation-should-digest" || value === "feedback:report-noise" || value === "feedback:report-surprise" ); } /** Single database for every project; every table partitions by project name. */ export function dbPath(): string { return join(stateDir(), "conductor.db"); } /** * Live worker count across the run store, opening the database WITHOUT * creating or upgrading it. The setup apply's quiescence preflight must count * workers before the acknowledged barrier, and database preparation is itself * one of the mutations that barrier must precede — so the observation that * gates it cannot be the thing that creates or migrates the store (#618). The * on-disk database is opened read-only (never created, never "prepared"); a * missing file reports zero because no run has ever been persisted, which the * caller only trusts when no daemon is running either. An unreadable or * schema-broken file throws rather than reading as zero — fail closed, never * absence. */ export function liveWorkersReadOnly(project?: string): number { const path = dbPath(); if (!existsSync(path)) return 0; let db: Database | undefined; try { db = new Database(path, { create: false, readonly: true }); db.exec("PRAGMA busy_timeout = 5000;"); const placeholders = LIVE_STATES.map(() => "?").join(", "); // `undefined` is the host-global selector: every project's live rows count, // never only the rows of a project literally named "" — a host-global // observation that scoped to `project = ''` hid every real named row and // read zero beside a running fleet (#651 review #3). const row = project === undefined ? db .query<{ n: number }, SqlValue[]>( `SELECT COUNT(*) AS n FROM runs WHERE state IN (${placeholders})`, ) .get(...LIVE_STATES) : db .query<{ n: number }, SqlValue[]>( `SELECT COUNT(*) AS n FROM runs WHERE project = ? AND state IN (${placeholders})`, ) .get(project, ...LIVE_STATES); return row?.n ?? 0; } catch (err) { throw new Error( `cannot read ${path} to prove the fleet is quiescent: ${err instanceof Error ? err.message : String(err)}`, ); } finally { db?.close(); } } /** The stem every restorable `conductor.db` snapshot is published under. */ export const DB_SNAPSHOT_STEM = "conductor.db.bak"; /** * A single-file, transaction-consistent copy of the store at `source` into * `target` — the shared core of {@link snapshotDb} and the setup apply's * rollback capture. * * Uses SQLite's own `VACUUM INTO`, never a plain file copy: the store opens * with `journal_mode=WAL`, and a `copyFileSync` can capture a torn database * while committed rows still sit uncheckpointed in the live WAL — a file that * exists, looks fresh and passes `integrity_check` yet is missing the newest * commits. `VACUUM INTO` reads the source as a whole (main file plus WAL) and * writes a single-file snapshot, so the result opens cleanly in a fresh * connection and contains every committed row. * * Runs against its own read of the source, deliberately not the caller's * writer connection, so a snapshot can be taken while a writer holds an * uncheckpointed WAL. `:memory:` has no on-disk state to snapshot and is * refused. */ export function vacuumInto(source: string, target: string): void { if (source === ":memory:") throw new Error("cannot snapshot an in-memory store"); if (!existsSync(source)) { throw new Error(`cannot snapshot ${source}: the store does not exist`); } let connection: Database | undefined; try { connection = new Database(source); connection.exec("PRAGMA busy_timeout = 5000;"); // `VACUUM INTO` takes the target as an SQL literal, not a bound parameter. connection.exec(`VACUUM INTO '${target.replaceAll("'", "''")}'`); } finally { connection?.close(); } } /** * A fresh, restorable copy of the store at `source`, published under `destDir` * with the same temp-then-atomic-publish convention as the config backups. * * The stem carries `at` (default now) so the filename's timestamp is the * snapshot's logical instant — the cadence passes its tick time, keeping the * name, the day marker and the retention ordering in one timeline. * * Returns the published snapshot path. */ export function snapshotDb(source: string, destDir: string, at: number = Date.now()): string { if (source === ":memory:") throw new Error("cannot snapshot an in-memory store"); if (!existsSync(source)) { throw new Error(`cannot snapshot ${source}: the store does not exist`); } mkdirSync(destDir, { recursive: true }); const temporary = join(destDir, `.${DB_SNAPSHOT_STEM}-${process.pid}.${randomUUID()}.tmp`); try { vacuumInto(source, temporary); } catch (err) { // Never leave a half-written snapshot on disk. rmSync(temporary, { force: true }); throw err; } return publishTempFile(temporary, destDir, `${DB_SNAPSHOT_STEM}-${backupTimestamp(at)}`); } /** * How many conductor.db snapshots the daemon's daily cadence keeps — a week * of ledger history at one snapshot per day. Deliberately a fixed count, not * an age or space policy: the store's ledger is append-only and the issue * forbids inventing a fleet-wide retention policy, so pruning is * configured-count only (#289). */ export const DB_SNAPSHOT_RETENTION = 7; /** * Prune a snapshot directory to the newest `keep` conductor.db snapshots. * * Snapshot stems embed the publish timestamp (`YYYY-MM-DDTHH-MM-SS…`), so * lexicographic order is chronological; only files named with the shared * `DB_SNAPSHOT_STEM` prefix are considered, never unrelated backups next to * them. Idempotent: a directory already at or under the bound removes * nothing. Returns the number of files removed. */ export function pruneDbSnapshots(dir: string, keep: number): number { const names = readdirSync(dir) .filter((name) => name.startsWith(DB_SNAPSHOT_STEM)) .sort(); const stale = names.length > keep ? names.slice(0, names.length - keep) : []; for (const name of stale) unlinkSync(join(dir, name)); return stale.length; } /** The UTC calendar day (`YYYY-MM-DD`) a moment falls on — the partition key * for the daemon's observed github call counter (#198). */ export function utcDay(now: number = Date.now()): string { return new Date(now).toISOString().slice(0, 10); } /** * Open (creating if needed) the run store at `dbPath`; `:memory:` is honoured * for tests. Safe to call on a fresh path — the schema is applied on open, so * there is no separate migration step to forget. */ export function openStore(dbPath: string): Store { if (dbPath !== ":memory:" && !dbPath.startsWith("file:")) { mkdirSync(dirname(dbPath), { recursive: true }); } const db = new Database(dbPath, { create: true }); // Set the wait before *any* pragma that may need SQLite's writer lock. // journal_mode itself can return SQLITE_BUSY during another connection's // close/checkpoint; setting this afterwards caused load-only CLI flakes // before the schema path ever had a chance to use the timeout (#396). db.exec("PRAGMA busy_timeout = 5000;"); // WAL lets the plugin read status while the daemon is mid-write; the busy // timeout covers the single writer lock they still contend for. db.exec("PRAGMA journal_mode = WAL;"); db.exec("PRAGMA foreign_keys = ON;"); db.exec(SCHEMA); // Failed or explicitly withdrawn handoffs no longer own a digest dedupe key: // both leave their underlying ledger rows owed for a replacement report. const reportDedupeIndex = db .query<{ sql: string | null }, []>( `SELECT sql FROM sqlite_master WHERE type = 'index' AND name = 'reports_dedupe'`, ) .get(); if (!(reportDedupeIndex?.sql ?? "").includes("state <> 'withdrawn'")) { db.exec( `BEGIN IMMEDIATE; DROP INDEX IF EXISTS reports_dedupe; CREATE UNIQUE INDEX reports_dedupe ON reports (project, dedupeKey) WHERE dedupeKey IS NOT NULL AND state <> 'failed' AND state <> 'withdrawn'; COMMIT;`, ); } // Additive migrations are idempotent and preserve every existing row. // Historical rows predate per-run ceilings, so the package default is the // only truthful recoverable value; every new run persists its actual cap. const columns = db.query<{ name: string }, []>("PRAGMA table_info(runs)").all(); if (!columns.some((column) => column.name === "maxTurns")) { db.exec( `ALTER TABLE runs ADD COLUMN maxTurns INTEGER NOT NULL DEFAULT ${DEFAULT_CAPS.workerMaxTurns}`, ); } // v0.3.18 and earlier created `runs` without the worker-observed PR head. if (!columns.some((column) => column.name === "headSha")) { db.exec("ALTER TABLE runs ADD COLUMN headSha TEXT"); } // Historical merged rows were never watched after settlement, so NULL is // "not observed", never permission to claim their base branch was green. for (const [name, type] of [ ["mergeSha", "TEXT"], ["baseRef", "TEXT"], ["baseCheck", "TEXT"], ["baseCheckAt", "INTEGER"], ] as const) { if (!columns.some((column) => column.name === name)) { db.exec(`ALTER TABLE runs ADD COLUMN ${name} ${type}`); } } // v0.3.22 and earlier removed a blocked run's dirty tree without saving it // (#118), so no row before this release has anywhere to record where the // work went. A NULL `salvageSha` on a historical row therefore means "never // asked", which reads the same as "clean tree" — the honest reading for a // release that never salvaged one. for (const [name, type] of [ ["salvageSha", "TEXT"], ["salvageError", "TEXT"], ["salvageAckAt", "INTEGER"], ] as const) { if (!columns.some((column) => column.name === name)) { db.exec(`ALTER TABLE runs ADD COLUMN ${name} ${type}`); } } // Rows written before #737 were never asked whether their tree was // quarantined, and a NULL means exactly that: the pre-quarantine passes // reaped the same trees they always had, so nothing before this release // needs backfilling — the column starts null and only a post-release pass // that actually refuses a broken store writes it. if (!columns.some((column) => column.name === "quarantineDetail")) { db.exec("ALTER TABLE runs ADD COLUMN quarantineDetail TEXT"); } // The worker's process/workspace identity (#842). Rows written before it exists // carry none, which is the truth: nothing reported a pane for them, so there is // no pane for a restart to reconcile. Nothing is backfilled — a pid guessed // after the fact is the one thing worse than no pid, because a reused pid reads // as a live worker. for (const [name, type] of [ ["workerPid", "INTEGER"], ["paneId", "TEXT"], ["paneLabel", "TEXT"], ["paneUnavailable", "TEXT"], ["outputTokens", "INTEGER"], ["reasoningTokens", "INTEGER"], ] as const) { if (!columns.some((column) => column.name === name)) { db.exec(`ALTER TABLE runs ADD COLUMN ${name} ${type}`); } } // Rows written before the settlement audit existed (#128) were never audited, // so a NULL here means "no audit ran" and reads the same as "found nothing". // That conflation is deliberate and harmless: the flags are advisory, and no // decision anywhere is taken on their absence. if (!columns.some((column) => column.name === "settlementFlags")) { db.exec("ALTER TABLE runs ADD COLUMN settlementFlags TEXT"); } // The settlement report was first persisted in #199; rows written before it // simply have no report, which is the honest reading of a kill that never // produced one. No backfill: the bytes only ever existed in a worker's // memory. if (!columns.some((column) => column.name === "report")) { db.exec("ALTER TABLE runs ADD COLUMN report TEXT"); } // The terminal evidence the settlement sweep records for every non-clean // row (#986). Rows written before the column were never looked at by the // recorder, so NULL is the honest reading — no backfill, because the // recorder derives evidence from facts only the settling pass holds. if (!columns.some((column) => column.name === "terminalEvidence")) { db.exec("ALTER TABLE runs ADD COLUMN terminalEvidence TEXT"); } // The progress watch's last-observed transcript write (#1086). Rows written // before the column were never observed, so NULL is the honest reading — no // backfill, because the instant only the file's own mtime holds. if (!columns.some((column) => column.name === "lastProgressAt")) { db.exec("ALTER TABLE runs ADD COLUMN lastProgressAt INTEGER"); } // Every row written before #132 is unclassified, and NULL is the honest // reading of that: the budget counters below deliberately still count an // unclassified terminal row exactly as this release's predecessor did, so an // upgrade never silently re-opens a budget an operator had already spent. for (const [name, type] of [ ["failureClass", "TEXT"], ["recoveryAction", "TEXT"], ["recoveredAt", "INTEGER"], ] as const) { if (!columns.some((column) => column.name === name)) { db.exec(`ALTER TABLE runs ADD COLUMN ${name} ${type}`); } } // The telegram plugin's three versions (#961), on the existing host row. A // store written before this release has no observation of them, and NULL is // the honest reading: the periodic pass that reads them had not shipped, so // the renderer says "unverified" until the next pass rather than implying the // versions agreed. const surfaceColumns = db.query<{ name: string }, []>("PRAGMA table_info(install_surfaces)").all(); for (const name of ["tgInstalled", "tgDaemon", "tgPublished"] as const) { if (!surfaceColumns.some((column) => column.name === name)) { db.exec(`ALTER TABLE install_surfaces ADD COLUMN ${name} TEXT`); } } // The model a run dispatched on was first written by the modelFallbacks // failover chain (#286). Rows predating the column are NULL, which is the // honest reading: only a chain-configured project records a model, and the // harness's own downgrade stays on the worker result, never in this column. if (!columns.some((column) => column.name === "model")) { db.exec("ALTER TABLE runs ADD COLUMN model TEXT"); } // The code-graph session observation (#726): rows written before the column // existed were never observed, so NULL is the honest reading of "no // observation recorded" — never "graph tools absent", which is the // `present: false` truth value of a run that did record one. if (!columns.some((column) => column.name === "graphTools")) { db.exec("ALTER TABLE runs ADD COLUMN graphTools TEXT"); } // Resume provenance (#567): rows written before #564 landed were never // resumed, so NULL is the honest reading of "no resume happened" — never // "resumed from unknown". No backfill: the identity existed only in the // dispatch decision, and deriving it now would guess at history. if (!columns.some((column) => column.name === "resumedFromRunId")) { db.exec("ALTER TABLE runs ADD COLUMN resumedFromRunId TEXT"); } // The declared file lane (#744): rows written before the column existed were // admitted before the declaration was persisted, so NULL is the honest // reading of "no declaration survives on this row" — the same fail-open as // a run admitted without a lane today. No backfill: re-parsing issue bodies // now would re-derive a snapshot admission no longer enforces, which is // exactly the #288 shape the gate exists to avoid. if (!columns.some((column) => column.name === "lane")) { db.exec("ALTER TABLE runs ADD COLUMN lane TEXT"); } // The count of in-session provider 429s a run recorded was first written by // the worker (#573). Rows predating the column are NULL, which is the honest // reading: the count was not recorded, so the classifier must not treat the // absence as "no rate limiting happened". if (!columns.some((column) => column.name === "provider429Count")) { db.exec("ALTER TABLE runs ADD COLUMN provider429Count INTEGER"); } // The within-run harness reliability surface (#539/#581): the resolved // model/provider and the fallback/retry/compaction counters. Rows predating // the columns are NULL, the honest reading of "the collection side had not // landed yet" — never "the run was healthy". for (const [name, type] of [ ["resolvedModel", "TEXT"], ["resolvedProvider", "TEXT"], ["retryFallbacks", "TEXT"], ["retryFallbackSucceeded", "INTEGER"], ["modelRecoveries", "INTEGER"], ["autoRetryCount", "INTEGER"], ["autoCompactionCount", "INTEGER"], ] as const) { if (!columns.some((column) => column.name === name)) { db.exec(`ALTER TABLE runs ADD COLUMN ${name} ${type}`); } } // The preserved budget charges (#795 review round 2): rows written before the // columns existed were never returned for revision from a terminal state, so // NULL is the honest reading of "no review claim preserved a charge on this // row", and the counters treat it as zero. for (const [name, type] of [ ["failureCharges", "INTEGER"], ["continuationCharges", "INTEGER"], ] as const) { if (!columns.some((column) => column.name === name)) { db.exec(`ALTER TABLE runs ADD COLUMN ${name} ${type}`); } } // Existing held-notice rows predate exact digest ownership (#274). A NULL // report id is truthful for already-digested history and means "still owed" // only while digestedAt is also NULL. const heldNoticeColumns = db.query<{ name: string }, []>("PRAGMA table_info(held_notices)").all(); if (!heldNoticeColumns.some((column) => column.name === "digestReportId")) { db.exec("ALTER TABLE held_notices ADD COLUMN digestReportId TEXT"); } // #273 distinguishes category-deferred notices from otherwise-interruptible // notices held only because the operator was outside their working window. // Historical rows were category-deferred, so false is the only safe default. if (!heldNoticeColumns.some((column) => column.name === "releaseOnAvailable")) { db.exec("ALTER TABLE held_notices ADD COLUMN releaseOnAvailable INTEGER NOT NULL DEFAULT 0"); } // Urgent recovery notices bypass category batching only; they still wait for // the configured availability window. Historical notices were not urgent. for (const [name, type] of [ ["withdrawnAt", "INTEGER"], ["withdrawnBy", "TEXT"], ["withdrawReason", "TEXT"], ] as const) { if (!heldNoticeColumns.some((column) => column.name === name)) { db.exec(`ALTER TABLE held_notices ADD COLUMN ${name} ${type}`); } } if (!heldNoticeColumns.some((column) => column.name === "urgent")) { db.exec("ALTER TABLE held_notices ADD COLUMN urgent INTEGER NOT NULL DEFAULT 0"); } // Multi-part delivery (#566): reports created before splitting existed carry // a single `messageId` and no per-part state, which is the honest reading of // a row that predates the watermark. `sentParts` defaults to 0 — "nothing // confirmed" — so a retry of a historical row starts from the first part. const reportColumns = db.query<{ name: string }, []>("PRAGMA table_info(reports)").all(); if (!reportColumns.some((column) => column.name === "messageIds")) { db.exec("ALTER TABLE reports ADD COLUMN messageIds TEXT"); } if (!reportColumns.some((column) => column.name === "sentParts")) { db.exec("ALTER TABLE reports ADD COLUMN sentParts INTEGER NOT NULL DEFAULT 0"); } if (!reportColumns.some((column) => column.name === "sentPartsHash")) { db.exec("ALTER TABLE reports ADD COLUMN sentPartsHash TEXT"); } for (const [name, type] of [ ["withdrawnAt", "INTEGER"], ["withdrawnBy", "TEXT"], ["withdrawReason", "TEXT"], ] as const) { if (!reportColumns.some((column) => column.name === name)) { db.exec(`ALTER TABLE reports ADD COLUMN ${name} ${type}`); } } // The kind split (#459): a row is either a question a human must answer or a // watch the orchestrator set for itself. Databases written before the split // never marked it, and every such row was a question — `watch` has no // historical form, because the old `decision open --resolves-when` was the // orchestrator's own misjudgement, not a durable kind. Defaulting the // historical rows to `question` is the honest reading. const decisionColumns = db.query<{ name: string }, []>("PRAGMA table_info(decisions)").all(); if (!decisionColumns.some((column) => column.name === "kind")) { db.exec("ALTER TABLE decisions ADD COLUMN kind TEXT NOT NULL DEFAULT 'question'"); } // The head binding (#808): rows written before it existed — met or not — // carry no binding and read back without one. Nothing is backfilled: fabricating // a head for an observation made before heads were recorded would be guessing // at history. The evaluator re-binds such a row on the next pass from a live // read, never from a migration guess. if (!decisionColumns.some((column) => column.name === "conditionHead")) { db.exec("ALTER TABLE decisions ADD COLUMN conditionHead TEXT"); } // The spec-out questionnaire (#947). Both columns are nullable and NULL is the // truthful value for every historical row: it was a single ask, not one item // of a group, and it specced nothing. The index is created here too, so an // upgraded database reads groups as fast as a fresh one. for (const [name, type] of [ ["groupId", "TEXT"], ["specIssue", "INTEGER"], ] as const) { if (!decisionColumns.some((column) => column.name === name)) { db.exec(`ALTER TABLE decisions ADD COLUMN ${name} ${type}`); } } db.exec("CREATE INDEX IF NOT EXISTS decisions_group ON decisions (groupId)"); // Promotion provenance on a grooming verdict (#1041). Purely additive, both // nullable, nothing backfilled: a verdict recorded before mechanical // promotion existed was queued — if it was queued at all — by a hand nobody // wrote down, and inventing `promotedBy: 'operator'` for it would put a // fabricated line in the very audit these columns exist to make honest. const groomingColumns = db.query<{ name: string }, []>("PRAGMA table_info(grooming)").all(); for (const [name, type] of [ ["promotedAt", "INTEGER"], ["promotedBy", "TEXT"], ] as const) { if (!groomingColumns.some((column) => column.name === name)) { db.exec(`ALTER TABLE grooming ADD COLUMN ${name} ${type}`); } } // Beside the ALTER, not in SCHEMA, so an upgraded database reads the audit // window as fast as a fresh one without the schema step depending on a column // it may still have to add. db.exec("CREATE INDEX IF NOT EXISTS grooming_promoted ON grooming (project, promotedAt)"); // Already-met watches, written before the deadline existed (#966). Unlike the // head binding above this IS backfilled, and the difference is that nothing is // being guessed: `conditionMetAt` is a recorded observation, and the deadline // is arithmetic on it. // // It has to be a migration rather than a re-observation, because // `markDecisionConditionMet` refuses a row whose `conditionMetAt` is already // set — deliberately, so the digest flag cannot look new every tick. Without // this, the three watches that motivated the fix would have kept // "[CONDITION MET — act on this now]" forever and the fix would have applied // only to conditions that fire in the future. // // Narrow on purpose: open watches only, only where the deadline is still the // infinite sentinel. A resolved row is history, a question's deadline runs // from `askedAt`, and a watch that somehow already carries a finite deadline // is left exactly as it is. db.run( `UPDATE decisions SET expiresAt = conditionMetAt + ? WHERE kind = 'watch' AND state = 'open' AND conditionMetAt IS NOT NULL AND expiresAt = ?`, [DECISION_TTL_MS, DECISION_WATCH_NEVER_EXPIRES], ); // The per-run spend reservation (#851). Rows written before it existed hold // no reservation, which is the truth: admission never took one for them. if (!columns.some((column) => column.name === "spendReservedUsd")) { db.exec("ALTER TABLE runs ADD COLUMN spendReservedUsd REAL"); } // Infra retries on a review round (#903). Historical rows were never // retried — the path did not exist, every infra kill settled `failed` — so // NULL is the honest reading and counts as zero. const reviewRevisionColumns = db.query<{ name: string }, []>("PRAGMA table_info(review_revisions)").all(); if (!reviewRevisionColumns.some((column) => column.name === "infraRetries")) { db.exec("ALTER TABLE review_revisions ADD COLUMN infraRetries INTEGER"); } // The launch decision and its model/session provenance (#1045). Purely // additive — one guarded ALTER per column, no backfill, no table rebuild — // because the identity of an existing round is exactly what this lane must // not disturb: its issue, run, attempt and continuation charges, round // number, branch, PR URL, reviewed head, findings and settlement history all // stay byte-identical across the upgrade. // // Nothing is backfilled, and that is the contract, not laziness. Every round // that ran before #1045 resumed its original session — but nobody wrote that // down, and a stamped 'resume-original' would be indistinguishable from a // decision the daemon actually made. These columns exist to explain #1035 // afterwards; seeding them with a guess is precisely the lie they are meant // to prevent. Legacy rows therefore project every field as absent, and every // reader must render an explicit unknown. for (const [name, type] of [ ["launchMode", "TEXT"], ["requestedModel", "TEXT"], ["resolvedModel", "TEXT"], ["originSessionRef", "TEXT"], ["correctionSessionRef", "TEXT"], ["launchDecidedAt", "INTEGER"], ] as const) { if (!reviewRevisionColumns.some((column) => column.name === name)) { db.exec(`ALTER TABLE review_revisions ADD COLUMN ${name} ${type}`); } } // Mined-signal provenance on an intake item (Phase 4). Purely additive, no // backfill: every row that predates signal mining was typed by an operator, // so an absent source is not a gap to be filled but the accurate reading. // // Nothing is rewritten, and in particular no id is: a legacy row keeps its // random id, and `intake dismiss ` on an id an operator wrote down last // week still resolves the same row after the upgrade. const intakeColumns = db.query<{ name: string }, []>("PRAGMA table_info(intake_items)").all(); if (!intakeColumns.some((column) => column.name === "source")) { db.exec("ALTER TABLE intake_items ADD COLUMN source TEXT"); } // The escalation findings on an adjudication (#932). Rows written by #874's // lifecycle carried none — the ceiling still refused then — so NULL is the // honest reading and renders as "no findings recorded". const adjudicationColumns = db.query<{ name: string }, []>("PRAGMA table_info(review_adjudications)").all(); if (!adjudicationColumns.some((column) => column.name === "findings")) { db.exec("ALTER TABLE review_adjudications ADD COLUMN findings TEXT"); } // One-time repair, and the reason it lives here rather than in the classifier: // a turn-0 environment fault that was ALREADY classified `unknown` and // escalated is never revisited. `runsNeedingClassification` only offers a row // whose class is NULL, or one whose recovery has not run yet — an // `unknown`/`escalate` row with `recoveredAt` set satisfies neither, so no // upgrade and no later tick would ever correct it. // // That mattered on a real fleet. The per-run `$HOME` redirect this release // deletes hid the harness's own credentials, so every dispatch died before it // read its issue and each one still spent an implementation attempt. Deleting // the redirect stops the bleeding; it does not give back attempts already // charged, and an issue sitting at its cap for a fault that was never in its // code stays undispatchable forever. // // Narrow by construction, and narrow through the CLASSIFIER's own predicate // rather than a second copy of the rule: {@link neverStarted}. A class of // exactly `unknown` — a row a human or a later release classified as anything // else is left alone — plus zero turns and either a recognised start error or // nothing recorded at all. A turn-0 row carrying an unrecognised error (a // failed mirror fetch, say) is untouched. Idempotent for free: afterwards these // rows read `env-start-failure`, which the WHERE no longer matches. const misread = db .query< { id: string; lastError: string | null; prUrl: string | null; headSha: string | null; salvageSha: string | null }, [] >( `SELECT id, lastError, prUrl, headSha, salvageSha FROM runs WHERE failureClass = 'unknown' AND turns = 0`, ) .all() .filter( (row) => neverStarted({ turns: 0, ...(row.lastError === null ? {} : { lastError: row.lastError }), ...(row.prUrl === null ? {} : { prUrl: row.prUrl }), ...(row.headSha === null ? {} : { headSha: row.headSha }), ...(row.salvageSha === null ? {} : { salvageSha: row.salvageSha }), }) !== undefined, ); if (misread.length > 0) { const reclassify = db.query( `UPDATE runs SET failureClass = 'env-start-failure' WHERE id = ?`, ); for (const row of misread) reclassify.run(row.id); } // The same repair, for the case the one above deliberately leaves alone: its // comment names it exactly ("a turn-0 row carrying an unrecognised error (a // failed mirror fetch, say) is untouched"). #177 gave that fault a name, but // only for rows written after it existed. Every earlier one was logged // `unknown`, whose recovery is `escalate` — so `runsNeedingClassification` // never revisits it and `countFailures` counts it forever. With // `maxAttemptsPerIssue` defaulting to 2, a single mirror ref-lock is half an // issue's entire budget spent on a fault that was never in its code. // // Narrow the same way: exactly `unknown`, state `failed`, and the // CLASSIFIER's own predicate rather than a second copy of the rule. // Deliberately only `failureClass` — not `recoveryAction`/`recoveredAt`. // Relieving the budget is the point; re-animating a months-old run into the // requeue sweep is not, and clearing the recovery columns would do exactly // that. Idempotent for free: these rows then read `dispatch-infra`, which the // WHERE no longer matches. const mirrorBroke = db .query< { id: string; lastError: string | null; prUrl: string | null; headSha: string | null; salvageSha: string | null }, [] >( `SELECT id, lastError, prUrl, headSha, salvageSha FROM runs WHERE failureClass = 'unknown' AND turns = 0 AND state = 'failed'`, ) .all() .filter( (row) => dispatchInfra({ turns: 0, ...(row.lastError === null ? {} : { lastError: row.lastError }), ...(row.prUrl === null ? {} : { prUrl: row.prUrl }), ...(row.headSha === null ? {} : { headSha: row.headSha }), ...(row.salvageSha === null ? {} : { salvageSha: row.salvageSha }), }) !== undefined, ); if (mirrorBroke.length > 0) { const reclassify = db.query( `UPDATE runs SET failureClass = 'dispatch-infra' WHERE id = ?`, ); for (const row of mirrorBroke) reclassify.run(row.id); } const insertRun = db.query( `INSERT INTO runs ( id, project, issue, repo, branch, worktree, state, attempt, turns, maxTurns, spendUsd, spendReservedUsd, sessionFile, resumedFromRunId, lane, prUrl, headSha, mergeSha, baseRef, baseCheck, baseCheckAt, salvageSha, salvageError, salvageAckAt, startedAt, endedAt, lastError, settlementFlags, report, graphTools, lastProgressAt ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectRun = db.query(`SELECT * FROM runs WHERE id = ?`); const selectGraphToolsObs = db.query<{ graphTools: string }, [string]>( `SELECT graphTools FROM runs WHERE project = ? AND graphTools IS NOT NULL`, ); const selectActive = db.query( `SELECT * FROM runs WHERE project = ? AND state IN (${ACTIVE_PLACEHOLDERS}) ORDER BY startedAt ASC`, ); const selectLive = db.query( `SELECT * FROM runs WHERE project = ? AND state IN (${LIVE_PLACEHOLDERS}) ORDER BY startedAt ASC`, ); // Runs holding an ACTIVE MUTATION LEASE (#899, #925): a live worker, or a // review revision the daemon has handed to one and that has not settled. // The second half is belt-and-braces — `claimRunForReview` moves the row to // `running` — but the claim and the launch are not the same instant, and a // lease the reader briefly forgets is a lane two writers can share. A // *queued* revision holds nothing: no worker has been woken for it. // // One query rather than a composition at each call site, because two // readers now decide "may this file be written" from it — admission's // file-lane gate and `conductor_pr_recover`'s fail-closed lane check — and // a lease rule that disagrees between them is the bug both are guarding. const selectLeased = db.query( `SELECT * FROM runs WHERE project = ? AND (state IN (${LIVE_PLACEHOLDERS}) OR id IN ( SELECT runId FROM review_revisions WHERE project = ? AND dispatchedAt IS NOT NULL AND settledAt IS NULL)) ORDER BY startedAt ASC`, ); const selectRetained = db.query( `SELECT * FROM runs WHERE project = ? AND state IN ('failed', 'killed', 'orphaned') AND worktree <> '' ORDER BY endedAt ASC, startedAt ASC`, ); const selectRecentRuns = db.query( `SELECT * FROM runs WHERE rowid IN ( SELECT MAX(rowid) FROM runs WHERE project = ? GROUP BY issue ) AND (state <> 'merged' OR COALESCE(endedAt, startedAt) >= ?) ORDER BY startedAt DESC`, ); // Every attempt that recorded one PR URL, newest first — deliberately NOT the // `recentRuns` newest-attempt-per-issue view. Mediated PR verbs resolve a // PR's owner through this: a requeued continuation hides its predecessor's // PR from `recentRuns` the moment it is created, so ownership must be able // to read past the issue's newest row (#434). Merged rows keep the same // recent-history bound as `recentRuns`. const selectRunsForPr = db.query( `SELECT * FROM runs WHERE project = ? AND prUrl = ? AND (state <> 'merged' OR COALESCE(endedAt, startedAt) >= ?) ORDER BY startedAt DESC, rowid DESC`, ); // The stats window (#282): rows settling at or after `since` — merged rows // included, so a chain whose PR merged inside the window is an outcome — // plus, for those merged issues, their WHOLE attempt chain even when it // started before the window. A continuation that began weeks ago and merged // today must report one lead time from its first claim, not from the window // start; ordering by issue/startedAt makes the first row of each chain its // first queue-label claim. const selectStatsRuns = db.query( `SELECT * FROM runs WHERE project = ? AND ( COALESCE(endedAt, startedAt) >= ? OR issue IN ( SELECT issue FROM runs WHERE project = ? AND state = 'merged' AND COALESCE(endedAt, startedAt) >= ? ) ) ORDER BY issue ASC, startedAt ASC, rowid ASC`, ); const selectPendingBaseChecks = db.query( `SELECT * FROM runs WHERE project = ? AND state = 'merged' AND baseCheck = 'pending' ORDER BY COALESCE(endedAt, startedAt) ASC, rowid ASC LIMIT ?`, ); const upsertBaseHealthRow = db.query< unknown, [string, string, string, string, string, number, SqlValue, number] >( `INSERT OR REPLACE INTO base_health (project, repo, branch, headSha, verdict, runsCount, detail, checkedAt) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectBaseHealth = db.query( `SELECT * FROM base_health WHERE project = ? ORDER BY repo ASC`, ); const selectMergedRepoBranches = db.query< { repo: string; baseRef: string | null }, [string, number] >( `SELECT repo, baseRef FROM runs WHERE project = ? AND state = 'merged' AND COALESCE(endedAt, startedAt) >= ? GROUP BY repo, baseRef ORDER BY repo ASC, baseRef ASC`, ); // Re-open (or refresh) an active freeze. On conflict we reset the clear // columns — a red re-observation after an override must re-arm the freeze — // and take the newest culprit/evidence/setAt as the current truth. const setBaseFreezeRow = db.query< unknown, [string, string, string, SqlValue, number] >( `INSERT INTO base_freeze (project, repo, culpritSha, detail, setAt, clearedAt, clearedBy, clearedReason) VALUES (?, ?, ?, ?, ?, NULL, NULL, NULL) ON CONFLICT(project, repo) DO UPDATE SET culpritSha = excluded.culpritSha, detail = excluded.detail, setAt = excluded.setAt, clearedAt = NULL, clearedBy = NULL, clearedReason = NULL`, ); const wasBaseFrozen = db.query<{ n: number }, [string, string]>( `SELECT COUNT(*) AS n FROM base_freeze WHERE project = ? AND repo = ? AND clearedAt IS NULL`, ); const selectBaseFreeze = db.query( `SELECT * FROM base_freeze WHERE project = ? AND repo = ?`, ); const selectFreezes = db.query( `SELECT * FROM base_freeze WHERE project = ? ORDER BY (clearedAt IS NULL) DESC, setAt DESC, repo ASC`, ); const clearBaseFreezeRow = db.query< unknown, [number, string, string, string, string] >( `UPDATE base_freeze SET clearedAt = ?, clearedBy = ?, clearedReason = ? WHERE project = ? AND repo = ? AND clearedAt IS NULL`, ); const selectActiveReleaseComposition = db.query( `SELECT * FROM release_composition WHERE project = ? AND closedAt IS NULL`, ); const declareReleaseCompositionRow = db.query< unknown, [string, string, string, SqlValue, number] >( `INSERT INTO release_composition (project, campaign, allowedPrUrls, declaredBy, declaredAt) VALUES (?, ?, ?, ?, ?) ON CONFLICT(project) DO UPDATE SET campaign = excluded.campaign, allowedPrUrls = excluded.allowedPrUrls, declaredBy = excluded.declaredBy, declaredAt = excluded.declaredAt, closedAt = NULL, closedBy = NULL, closedReason = NULL`, ); const retireReleaseCompositionRow = db.query( `UPDATE release_composition SET closedAt = ?, closedBy = ?, closedReason = ? WHERE project = ? AND closedAt IS NULL`, ); const insertReleaseCompositionOverride = db.query< unknown, [string, string, string, string, number, string, string] >( `INSERT INTO release_composition_overrides (id, project, campaign, prUrl, at, by, reason) VALUES (?, ?, ?, ?, ?, ?, ?)`, ); const selectReleaseCompositionOverride = db.query< ReleaseCompositionOverrideRow, [string, string, string] >( `SELECT * FROM release_composition_overrides WHERE project = ? AND campaign = ? AND prUrl = ? ORDER BY at DESC LIMIT 1`, ); const selectReleaseCompositionOverrides = db.query< ReleaseCompositionOverrideRow, [string, string] >( `SELECT * FROM release_composition_overrides WHERE project = ? AND campaign = ? ORDER BY at DESC`, ); // Declaring is a check-then-insert, and two processes (CLI + daemon) can // race: without the transaction the second declarer would silently replace // a still-active composition — exactly the overwrite this guard exists to // prevent. Inside one write transaction the re-check sees the row the first // declarer just wrote. Returns the conflicting active row, or null when the // declaration took. const declareReleaseCompositionTx = db.transaction( ( project: string, campaign: string, allowedPrUrls: string, declaredBy: string | null, declaredAt: number, ): ReleaseCompositionRow | null => { const existing = selectActiveReleaseComposition.get(project); if (existing !== null) return existing; declareReleaseCompositionRow.run(project, campaign, allowedPrUrls, declaredBy, declaredAt); return null; }, ); const countAttempts = db.query<{ n: number }, [string, number]>( `SELECT COUNT(*) AS n FROM runs WHERE project = ? AND issue = ?`, ); // The classes excluded here are the ones whose cause is not the work (#132): a // runner that never produced a verdict, a row whose PR had already merged, a // kill that came from a daemon restart rather than from a turn ceiling. On this // fleet #350 spent its whole continuation budget on causes like those and then // sat in admission hold for six dispatch cycles. An unclassified row (NULL) is // counted exactly as before, so upgrading changes no existing budget. // The classes a terminal row must NOT carry to count as a charged failed // attempt or continuation — the budget counters' exclusions, shared verbatim // with the review claim that preserves a charge (#795): the claim evaluates // the same predicate against the pre-claim row, so a preserved charge can // never disagree with what the counters would have counted. const CHARGEABLE_FAILURE_EXCLUSIONS = "'ci-infra', 'settlement-stuck', 'env-start-failure', 'dispatch-infra', 'provider-credit', 'provider-transient', 'provider-capacity', 'returned-for-revision', 'model-empty-stop'"; const CHARGEABLE_CONTINUATION_EXCLUSIONS = "'admin-kill', 'settlement-stuck', 'env-start-failure', 'dispatch-infra', 'provider-credit', 'provider-transient', 'provider-capacity'"; // A row whose review revision preserved a charge (#795 review round 2) keeps // counting after the claim leaves it `pushed-green`. The preserved columns // are per-category INTEGER totals, summed here — not a single enum over // rows — so repeated same-category terminal events across consecutive review // rounds each keep their charge: a killed run contributes one continuation, // its review claim moves that charge into `continuationCharges`, and a // second cap on the resumed round is counted by the state branch on top of // it. The two sums can never double-count one event: a terminal event is // counted by the state branch only while the row is still terminal, and the // claim that flips it to `running` is the same atomic statement that adds it // to the preserved column. const countFailures = db.query<{ n: number }, SqlValue[]>( `SELECT (SELECT COALESCE(SUM(failureCharges), 0) FROM runs WHERE project = ? AND issue = ?) + (SELECT COUNT(*) FROM runs WHERE project = ? AND issue = ? AND state = 'failed' AND (failureClass IS NULL OR failureClass NOT IN (${CHARGEABLE_FAILURE_EXCLUSIONS}))) AS n`, ); const countContinuations = db.query<{ n: number }, SqlValue[]>( `SELECT (SELECT COALESCE(SUM(continuationCharges), 0) FROM runs WHERE project = ? AND issue = ?) + (SELECT COUNT(*) FROM runs WHERE project = ? AND issue = ? AND ( (state IN ('killed', 'orphaned', 'blocked') AND (failureClass IS NULL OR failureClass NOT IN (${CHARGEABLE_CONTINUATION_EXCLUSIONS}))) -- Two failed-state classes charge a continuation instead of an -- attempt (#795, #1001): a reviewer's decision, and a provider -- that answered with empty turns until the harness gave up. -- Neither is the work failing, and both resume from the head the -- attempt already pushed - but each still consumes a bounded -- budget, so a provider stuck in that state escalates to a human -- rather than looping forever. OR (state = 'failed' AND failureClass IN ('returned-for-revision', 'model-empty-stop')) )) AS n`, ); // How many times one issue reached a given class. Recovery uses it to bound a // retry loop whose cause is persistent (e.g. a mirror that will not refresh): // after `DISPATCH_INFRA_MAX_STRIKES` the daemon escalates instead of looping. const countByClass = db.query<{ n: number }, [string, number, FailureClass]>( `SELECT COUNT(*) AS n FROM runs WHERE project = ? AND issue = ? AND failureClass = ?`, ); // Newest first, and bounded: every row this returns costs `gh` calls to gather // facts for, so a fleet with a long unclassified history classifies over // several ticks rather than spending one tick's budget on all of it. // // Classified-but-unrecovered rows are included, not just unclassified ones: a // recovery whose tracker write failed has to come back, or the log line saying // "retrying next tick" is a lie and the row is stranded with a class and no // action. `hold` is excluded because it is *recorded only* by design — its // `recoveredAt` stays NULL forever, and re-offering it would spin the sweep. // // One deliberate exception re-enters the sweep after classification: a // blocked row with a PR. It is the observation half of the // `awaiting-observation` recovery (#1068) — the sweep re-reads the PR every // pass, and the classifier's merged-PR branch names `settlement-stuck` the // way it does for any other stuck row, so a blocked run whose PR later merges // (the #1062 shape: merged at 15:02, row still `blocked`/`question`) settles // instead of listing for Duty 1 triage forever. A genuine question row is // re-offered too and its escalation is not repeated: the notifications ledger // dedupes on the class+run summary of the stable evidence. Rows without a PR // have nothing to observe and are only re-offered by the ordinary clauses. const selectUnclassified = db.query( `SELECT * FROM runs WHERE project = ? AND state IN ('blocked', 'killed', 'failed', 'orphaned', 'pushed-green') AND ( failureClass IS NULL OR (recoveredAt IS NULL AND recoveryAction IN ('settle', 'continue', 'requeue', 'rerun-checks')) OR (state = 'blocked' AND prUrl IS NOT NULL) ) ORDER BY startedAt DESC, rowid DESC LIMIT ?`, ); // The settled rows a classifier correction can no longer reach (#638). A // `ci-deterministic`/`escalate` row never re-enters `selectUnclassified` // (its class is set and its recovery is not retryable), so evidence the // classifier learned after the row was written — the codeload setup 429 of // #177 — stays a charged attempt forever. This WHERE is the historical // reconciliation's exact predicate: nothing but rows currently reading // `ci-deterministic`, never a class a human re-read, and the class itself is // only ever produced for `state = 'failed'`. Bounded like every other sweep // read, because each candidate costs tracker calls to re-fetch its evidence. // `after` is the persisted review cursor: candidates are only rows strictly // older than it, so a bounded pass progresses through a long misclassified // history instead of rescanning the newest non-matches forever. A NULL cursor // (first pass, or one rendered stale by a classifier-signature change) means // "start from the newest". `SELECT rowid, *` exposes the hidden tie-breaker // so the pass can record an exact resume boundary. const selectHistoricalInfraCandidates = db.query( `SELECT rowid, * FROM runs WHERE project = ? AND failureClass = 'ci-deterministic' AND state = 'failed' AND (? IS NULL OR (startedAt < ? OR (startedAt = ? AND rowid < ?))) ORDER BY startedAt DESC, rowid DESC LIMIT ?`, ); const selectHistoricalInfraCursor = db.query< { cursorStartedAt: number; cursorRowid: number; classifierVersion: string }, [string] >( `SELECT cursorStartedAt, cursorRowid, classifierVersion FROM historical_infra_review WHERE project = ?`, ); const upsertHistoricalInfraCursor = db.query< unknown, [string, number, number, string, number] >( `INSERT INTO historical_infra_review (project, cursorStartedAt, cursorRowid, classifierVersion, updatedAt) VALUES (?, ?, ?, ?, ?) ON CONFLICT(project) DO UPDATE SET cursorStartedAt = excluded.cursorStartedAt, cursorRowid = excluded.cursorRowid, classifierVersion = excluded.classifierVersion, updatedAt = excluded.updatedAt`, ); // The one mutation the historical reconciliation may make. Guarded by the // current class, so a second pass is a no-op and a race can never reclassify // a row a human has re-read. Deliberately only `failureClass` — never // `recoveryAction` or `recoveredAt`: relieving the budget is the point, // re-animating a months-old run into the requeue sweep is not (#638). const reclassifyToInfra = db.query( `UPDATE runs SET failureClass = 'ci-infra' WHERE id = ? AND failureClass = 'ci-deterministic'`, ); // Two different questions, deliberately not one number (#132). // // `recoveredAt IS NULL` alone does NOT mean "awaiting recovery": two actions // are **recorded only** by design and never stamp it. `hold` is documented as // such right above `selectUnclassified`; `none` is what the settle sweep // writes beside `returned-for-revision` when a reviewer closes pushed work // without merging, because the remedy there is the queue label the sweep // deliberately leaves on, not an action any sweep takes. // // Counting them together is the #109 defect restated: on this fleet the line // read `returned-for-revision 44` while every genuinely actionable class sat // at zero, so a monotonically growing tally of *review decisions* crowded out // the classes an operator can act on. The split is on `recoveryAction`, which // the sweep already treats this way, so the surface now agrees with the model // instead of contradicting it. // Newest terminal runs, reduced to the two columns the spend-telemetry // judgement needs (#970). Terminal only: an in-flight run has not finished // reporting cost, so its 0.00 is not a telemetry statement. const selectSpendSamples = db.query<{ turns: number; spendUsd: number }, [string, number]>( `SELECT turns, spendUsd FROM runs WHERE project = ? AND endedAt IS NOT NULL ORDER BY endedAt DESC, rowid DESC LIMIT ?`, ); // Newest runs that can yield a per-turn pace (#1063): rows with turns and // elapsed time recorded. Same over-sample discipline as the spend query // above — the caller filters by model attribution, so this stays unfiltered // beyond what no row can answer without. const selectLatencySamples = db.query( `SELECT * FROM runs WHERE project = ? AND turns > 0 AND endedAt IS NOT NULL ORDER BY endedAt DESC, rowid DESC LIMIT ?`, ); const selectFailureClassCounts = db.query<{ cls: string; n: number }, [string]>( `SELECT failureClass AS cls, COUNT(*) AS n FROM runs WHERE project = ? AND failureClass IS NOT NULL AND recoveredAt IS NULL AND state IN ('blocked', 'killed', 'failed', 'orphaned', 'pushed-green') AND recoveryAction NOT IN ('none', 'hold', 'observe') GROUP BY failureClass ORDER BY n DESC, failureClass ASC`, ); const selectRecordedOnlyClassCounts = db.query<{ cls: string; n: number }, [string]>( `SELECT failureClass AS cls, COUNT(*) AS n FROM runs WHERE project = ? AND failureClass IS NOT NULL AND recoveredAt IS NULL AND state IN ('blocked', 'killed', 'failed', 'orphaned', 'pushed-green') AND recoveryAction IN ('none', 'hold', 'observe') GROUP BY failureClass ORDER BY n DESC, failureClass ASC`, ); // The surviving `unknown` rows (#986): classified but never explained, each // carrying its recorded terminal evidence for `status` to print. Bounded — // a backlog must not turn one status render into an unbounded scan. const selectUnknownEvidence = db.query( `SELECT * FROM runs WHERE project = ? AND failureClass = 'unknown' ORDER BY startedAt DESC, rowid DESC LIMIT ?`, ); const selectRecoveredSince = db.query( `SELECT * FROM runs WHERE project = ? AND recoveredAt IS NOT NULL AND recoveredAt >= ? ORDER BY recoveredAt DESC, rowid DESC`, ); // Salvage state that still describes an issue's present, so an operator is // shown a preserved WIP tip exactly while it is the thing a re-claim would // build on — and an unsalvaged tree for as long as it is the only copy. // Newest-attempt-only for the same reason the board is: a later attempt has // already consumed or superseded whatever an older one left behind. const selectSalvaged = db.query( `SELECT * FROM runs WHERE rowid IN ( SELECT MAX(rowid) FROM runs WHERE project = ? GROUP BY issue ) AND (salvageSha IS NOT NULL OR salvageError IS NOT NULL) ORDER BY issue ASC`, ); // Quarantined retained trees (#737), newest attempt per issue — the same // shape as `selectSalvaged` because the same change can land twice: a row // long since reaped must not keep its issue looking quarantined, and an // older quarantined attempt must not claim the issue while a newer, // healthy one replaced it. const selectQuarantined = db.query( `SELECT * FROM runs WHERE rowid IN ( SELECT MAX(rowid) FROM runs WHERE project = ? GROUP BY issue ) AND quarantineDetail IS NOT NULL ORDER BY issue ASC`, ); // Newest attempt for one issue. `startedAt` is millisecond-resolution and two // attempts could in principle share one, so rowid breaks the tie by insertion // order — a `tail` that attached to the older of two same-millisecond attempts // would follow a transcript nobody is writing to any more. const selectLatestRun = db.query( `SELECT * FROM runs WHERE project = ? AND issue = ? ORDER BY startedAt DESC, rowid DESC LIMIT 1`, ); // Every attempt for one issue, oldest first. An exhaustion escalation groups // these by `failureClass` to report how the continuation budget was spent // (#439); the numeric `*For` counters cannot reconstruct that on their own. const selectRunsForIssue = db.query( `SELECT * FROM runs WHERE project = ? AND issue = ? ORDER BY startedAt ASC, rowid ASC`, ); const upsertTurnOverride = db.query( `INSERT INTO turn_overrides (project, issue, maxTurns, setAt) VALUES (?, ?, ?, ?) ON CONFLICT(project, issue) DO UPDATE SET maxTurns = excluded.maxTurns, setAt = excluded.setAt`, ); const insertTurnOverrideLedger = db.query< unknown, [string, string, number, number, number] >( `INSERT INTO turn_override_ledger (id, project, issue, maxTurns, setAt) VALUES (?, ?, ?, ?, ?)`, ); const selectTurnOverride = db.query<{ maxTurns: number }, [string, number]>( `SELECT maxTurns FROM turn_overrides WHERE project = ? AND issue = ?`, ); const deleteTurnOverride = db.query( `DELETE FROM turn_overrides WHERE project = ? AND issue = ?`, ); const selectTurnOverrides = db.query( `SELECT project, issue, maxTurns, setAt FROM turn_overrides WHERE project = ? ORDER BY issue ASC`, ); const selectTurnOverrideLedger = db.query< TurnOverride, [string, number, number, number] >( `SELECT project, issue, maxTurns, setAt FROM turn_override_ledger WHERE project = ? AND (? = 0 OR issue = ?) ORDER BY setAt DESC, rowid DESC LIMIT ?`, ); // #807. `INSERT OR IGNORE` is the whole one-shot rule: the second spinning // cap on a chain changes no rows, so the decomposition verdict is reached // without any read-then-write window. Composed into `escalateModelTx`, which // is what makes the marker and the continuation it authorises one fact. const claimModelEscalation = db.query< unknown, [string, number, string, string, string, number] >( `INSERT OR IGNORE INTO model_escalations (project, issue, model, failureClass, runId, at) VALUES (?, ?, ?, ?, ?, ?)`, ); const selectModelEscalation = db.query( `SELECT model, failureClass, runId, at FROM model_escalations WHERE project = ? AND issue = ?`, ); const stampRecovered = db.query( `UPDATE runs SET recoveredAt = ? WHERE id = ?`, ); // The settlement report row is kept for the audit trail; nothing pools it // across attempts any more (#488). const countStartedSince = db.query<{ n: number }, [string, number]>( `SELECT COUNT(*) AS n FROM runs WHERE project = ? AND startedAt >= ?`, ); const sumSpendSince = db.query<{ total: number }, [string, number]>( `SELECT COALESCE(SUM(spendUsd), 0) AS total FROM runs WHERE project = ? AND startedAt >= ?`, ); // Reservations held by runs that are still live (#851). Summed rather than // tracked: a settle releases the unspent remainder by leaving LIVE_STATES, // with nothing to remember to write, and a daemon restart re-reads the same // reservations off the same rows. A run that spent MORE than it reserved is // counted at what it reserved — its overspend is already in `spendSince`. const sumLiveReservations = db.query<{ total: number }, SqlValue[]>( `SELECT COALESCE(SUM(spendReservedUsd), 0) AS total FROM runs WHERE project = ? AND state IN (${LIVE_PLACEHOLDERS}) AND spendReservedUsd IS NOT NULL`, ); const upsertInstallSurfaces = db.query< unknown, [number, string, string | null, string | null, string | null, string | null, string | null] >( `INSERT INTO install_surfaces (host, at, cliVersion, ompVersion, herdrSource, tgInstalled, tgDaemon, tgPublished) VALUES ('host', ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(host) DO UPDATE SET at = excluded.at, cliVersion = excluded.cliVersion, ompVersion = excluded.ompVersion, herdrSource = excluded.herdrSource, tgInstalled = excluded.tgInstalled, tgDaemon = excluded.tgDaemon, tgPublished = excluded.tgPublished`, ); const selectInstallSurfaces = db.query< { at: number; cliVersion: string; ompVersion: string | null; herdrSource: string | null; tgInstalled: string | null; tgDaemon: string | null; tgPublished: string | null; }, [] >( `SELECT at, cliVersion, ompVersion, herdrSource, tgInstalled, tgDaemon, tgPublished FROM install_surfaces WHERE host = 'host'`, ); const countNotified = db.query<{ n: number }, [string]>( `SELECT COUNT(*) AS n FROM notifications WHERE "key" = ?`, ); // First notification wins, so the row records when a human was actually // paged rather than when the newest retry re-reported the same event. const insertNotified = db.query( `INSERT OR IGNORE INTO notifications ("key", at) VALUES (?, ?)`, ); // Orchestrator-down incident (#288). One row per project; INSERT OR IGNORE is // the dedupe so a flapping orchestrator opens (and pages) once per incident. const selectOrchestratorIncident = db.query( `SELECT project, mode, cause, since, diverted FROM orchestrator_incidents WHERE project = ?`, ); const insertOrchestratorIncident = db.query( `INSERT OR IGNORE INTO orchestrator_incidents (project, mode, cause, since) VALUES (?, ?, ?, ?)`, ); const bumpOrchestratorIncident = db.query( `UPDATE orchestrator_incidents SET diverted = diverted + ? WHERE project = ?`, ); const deleteOrchestratorIncident = db.query( `DELETE FROM orchestrator_incidents WHERE project = ? RETURNING project, mode, cause, since, diverted`, ); // Worker-workspace ownership rows (#1035 review). See the table comment: // the durable half of discovery, so a Herdr restart that drops metadata // tokens cannot orphan the surface conductor created. const selectHerdrWorkerWorkspaces = db.query<{ workspace_id: string }, [string]>( `SELECT workspace_id FROM herdr_worker_workspaces WHERE project = ? ORDER BY created_at ASC, workspace_id ASC`, ); const insertHerdrWorkerWorkspace = db.query( `INSERT OR IGNORE INTO herdr_worker_workspaces (project, workspace_id, created_at) VALUES (?, ?, ?)`, ); const deleteHerdrWorkerWorkspace = db.query( `DELETE FROM herdr_worker_workspaces WHERE project = ? AND workspace_id = ?`, ); const upsertDispatch = db.query( `INSERT INTO dispatch_summaries (project, summary) VALUES (?, ?) ON CONFLICT(project) DO UPDATE SET summary = excluded.summary`, ); const selectDispatch = db.query<{ summary: string }, [string]>( `SELECT summary FROM dispatch_summaries WHERE project = ?`, ); // One projection list for every grooming read, so a new column can never be // visible through one accessor and NULL through another. const GROOMING_COLUMNS = "project, issue, verdict, reason, evidence, recordedAt, promotedAt, promotedBy"; const selectGrooming = db.query( `SELECT ${GROOMING_COLUMNS} FROM grooming WHERE project = ? AND issue = ?`, ); const selectGroomingAll = db.query( `SELECT ${GROOMING_COLUMNS} FROM grooming WHERE project = ? ORDER BY issue ASC`, ); const selectGroomingByVerdict = db.query( `SELECT ${GROOMING_COLUMNS} FROM grooming WHERE project = ? AND verdict = ? ORDER BY issue ASC`, ); // Newest promotion first, and `promotedAt DESC` alone would be a non- // deterministic order for a batch promoted inside one millisecond, which the // daemon's promotion loop does routinely — the issue number breaks the tie so // an audit digest renders the same list twice. const selectPromotionsSince = db.query( `SELECT ${GROOMING_COLUMNS} FROM grooming WHERE project = ? AND promotedAt IS NOT NULL AND promotedAt >= ? ORDER BY promotedAt DESC, issue DESC`, ); // The upsert deliberately does NOT carry the promotion columns forward: on // conflict they are reset to NULL, because a re-recorded verdict is a new // judgement and provenance that outlives its judgement audits the wrong // decision (#1041). Spelled as explicit `= NULL` rather than left out, so the // clearing is a stated choice a reader can weigh rather than an omission. const upsertGroomingRow = db.query( `INSERT INTO grooming (project, issue, verdict, reason, evidence, recordedAt) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(project, issue) DO UPDATE SET verdict = excluded.verdict, reason = excluded.reason, evidence = excluded.evidence, recordedAt = excluded.recordedAt, promotedAt = NULL, promotedBy = NULL`, ); // Stamp-once, and the `IS NULL` guard is the whole contract: `changes === 0` // means either no such verdict or one already promoted, which is exactly the // set of cases a caller must not queue the issue for. That makes this write // the promotion latch, so it runs before the label op (#1041). const markGroomingPromotedRow = db.query( `UPDATE grooming SET promotedAt = ?, promotedBy = ? WHERE project = ? AND issue = ? AND promotedAt IS NULL`, ); // A promoted row is inert to admission's blocked bookkeeping, in both // directions (#1041): the sweep below skips it and the per-hold upsert never // rewrites it. Promotion handed the issue to dispatch, and a lane/dependency // hold on a queued issue is flow control of the moment, not a re-grooming. // Without the exclusion the audit line would die inside one 5-minute // admission pass — rewritten `blocked`, then swept — while the tick that must // review the promotion runs every ~30 minutes. const clearGroomedBlocked = db.query( `DELETE FROM grooming WHERE project = ? AND verdict = 'blocked' AND reason IN ('file-lane', 'depends-on') AND promotedAt IS NULL`, ); const selectGroomingPromotedAt = db.query<{ promotedAt: number | null }, [string, number]>( `SELECT promotedAt FROM grooming WHERE project = ? AND issue = ?`, ); // One transaction, so a pass can never leave the blocked inventory half // written: the stale rows are cleared (the hold no longer applies) and the // current holds re-recorded atomically. const reconcileGroomingTx = db.transaction( (project: string, holds: readonly { issue: number; reason: AdmissionHoldReason; detail?: string }[]): void => { clearGroomedBlocked.run(project); for (const hold of holds) { if (BLOCKED_GROOMING_HOLDS[hold.reason] !== true) continue; if (selectGroomingPromotedAt.get(project, hold.issue)?.promotedAt != null) continue; upsertGroomingRow.run(project, hold.issue, "blocked", hold.reason, hold.detail ?? hold.reason, Date.now()); } }, ); // A groomed verdict describes an OPEN issue, and nothing retired one when its // issue closed (#964). `clearGroomedBlocked` above only ever cleared the two // per-pass admission holds, so every other row — `promotable`, `considered`, // `needs-product-decision`, an in-flight batch marker — accumulated forever. // Measured on this fleet 2026-08-23: 64 of 65 rows named a closed issue, and // all 26 `promotable` ones did, so `status` advertised 26 candidates ready to // promote on a repo with exactly one open issue. // // Deleted one row at a time inside a transaction rather than with a generated // `NOT IN (...)`: the closed set is computed by the caller from a complete // open-issue snapshot, and an over-long IN list is exactly where a partial // snapshot would quietly delete verdicts it never verified. const deleteGroomingRow = db.query( `DELETE FROM grooming WHERE project = ? AND issue = ?`, ); const retireGroomingTx = db.transaction((project: string, issues: readonly number[]): number => { let removed = 0; for (const issue of issues) removed += deleteGroomingRow.run(project, issue).changes; return removed; }); // Epic scope approvals (#1041). `DO NOTHING` rather than `DO UPDATE` is the // idempotence: the durable fact is when consent was first given, so a second // `epic approve` must not silently restamp the timestamp and make an old // approval look fresh in an audit. const insertEpicApproval = db.query( `INSERT INTO epic_approvals (project, issue, approvedAt, approvedBy) VALUES (?, ?, ?, ?) ON CONFLICT(project, issue) DO NOTHING`, ); const deleteEpicApproval = db.query( `DELETE FROM epic_approvals WHERE project = ? AND issue = ?`, ); const selectEpicApproval = db.query( `SELECT issue, approvedAt, approvedBy FROM epic_approvals WHERE project = ? AND issue = ?`, ); const selectEpicApprovals = db.query( `SELECT issue, approvedAt, approvedBy FROM epic_approvals WHERE project = ? ORDER BY issue ASC`, ); const selectFrictionRollup = db.query( `SELECT * FROM friction_rollups WHERE project = ? AND day = ? AND kind = ?`, ); const upsertFrictionRollup = db.query< unknown, [string, string, string, number, number, string, string, number] >( `INSERT INTO friction_rollups (project, day, kind, observations, occurrences, issues, samples, latestAt) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(project, day, kind) DO UPDATE SET observations = excluded.observations, occurrences = excluded.occurrences, issues = excluded.issues, samples = excluded.samples, latestAt = excluded.latestAt`, ); const selectFrictionSince = db.query( `SELECT * FROM friction_rollups WHERE project = ? AND day >= ? ORDER BY latestAt DESC`, ); const selectFrictionSurfaces = db.query( `SELECT kind, at FROM friction_surfaces WHERE project = ?`, ); const upsertFrictionSurface = db.query( `INSERT INTO friction_surfaces (project, kind, at) VALUES (?, ?, ?) ON CONFLICT(project, kind) DO UPDATE SET at = excluded.at`, ); // Observed GitHub rate-limit refusals (#198). Insert + prune share one // transaction so a burst never grows the table without bound. const insertGhRefusal = db.query( `INSERT INTO gh_refusals (at) VALUES (?)`, ); const pruneGhRefusals = db.query( `DELETE FROM gh_refusals WHERE at < ?`, ); const selectGhRefusalsSince = db.query<{ n: number; latest: number | null }, [number]>( `SELECT COUNT(*) AS n, MAX(at) AS latest FROM gh_refusals WHERE at >= ?`, ); const recordGhRefusalTx = db.transaction((at: number): void => { insertGhRefusal.run(at); pruneGhRefusals.run(at - 24 * 60 * 60 * 1000); }); // Auto-restarts of a wedged orchestrator (Phase 4). Insert + prune share one // transaction, exactly as above, so a project that keeps wedging can never // grow this table without bound. The retention (7d) is deliberately longer // than the 6h cap window: the cap needs the last few hours, but an operator // reading the page wants to know whether this is the first bad night or the // fourth. const insertOrchestratorRestart = db.query( `INSERT INTO orchestrator_restarts (project, at, reason) VALUES (?, ?, ?)`, ); const pruneOrchestratorRestarts = db.query( `DELETE FROM orchestrator_restarts WHERE at < ?`, ); const selectOrchestratorRestartsSince = db.query<{ at: number; reason: string }, [string, number]>( `SELECT at, reason FROM orchestrator_restarts WHERE project = ? AND at >= ? ORDER BY at DESC, rowid DESC`, ); const recordOrchestratorRestartTx = db.transaction( (project: string, at: number, reason: string): void => { insertOrchestratorRestart.run(project, at, reason); pruneOrchestratorRestarts.run(at - 7 * 24 * 60 * 60 * 1000); }, ); // Daemon-owned session spend (Phase 4). Append-only: these sessions have no // lifecycle, they either ran or they did not. const insertSessionSpend = db.query( `INSERT INTO session_spend (project, role, issue, model, resolvedModel, turns, spendUsd, at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectSessionSpendSince = db.query( `SELECT project, role, issue, model, resolvedModel, turns, spendUsd, at FROM session_spend WHERE project = ? AND at >= ? ORDER BY at DESC, id DESC`, ); // The daemon's tracked github call counts (#198): one UPSERT per spawn. const bumpGhCall = db.query( `INSERT INTO gh_calls (day, source, calls) VALUES (?, ?, 1) ON CONFLICT(day, source) DO UPDATE SET calls = calls + 1`, ); const selectGhCalls = db.query<{ source: string; calls: number }, [string]>( `SELECT source, calls FROM gh_calls WHERE day = ? ORDER BY source`, ); const sumGhCallsBetween = db.query<{ total: number }, [string, string]>( `SELECT COALESCE(SUM(calls), 0) AS total FROM gh_calls WHERE day BETWEEN ? AND ?`, ); // The report outbox (#123). Every terminal transition names the attempt it is // settling, because a request can come back *after* the stale-`sending` sweep // has already reclaimed its row — and a late answer must not overwrite a // newer attempt's outcome. That predicate is also what stops one report being // concurrently in flight twice: only one claim can move a `pending` row. const insertReport = db.query( `INSERT OR IGNORE INTO reports ( id, project, kind, body, state, attempts, attemptId, ambiguous, dedupeKey, createdAt, updatedAt, nextAttemptAt, messageId, deliveredAt, lastError ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectReport = db.query(`SELECT * FROM reports WHERE id = ?`); const selectReportByDedupe = db.query( `SELECT * FROM reports WHERE project = ? AND dedupeKey = ? AND state NOT IN ('failed', 'withdrawn') ORDER BY createdAt DESC, rowid DESC LIMIT 1`, ); const selectDueReports = db.query( `SELECT * FROM reports WHERE project = ? AND state = 'pending' AND nextAttemptAt <= ? ORDER BY createdAt ASC, rowid ASC LIMIT ?`, ); const withdrawPendingReportRow = db.query< unknown, [number, string, string, number, string, string] >( `UPDATE reports SET state = 'withdrawn', withdrawnAt = ?, withdrawnBy = ?, withdrawReason = ?, updatedAt = ?, attemptId = NULL WHERE id = ? AND project = ? AND state = 'pending'`, ); const selectReportNoticeIds = db.query<{ id: string }, [string, string]>( `SELECT id FROM held_notices WHERE digestReportId = ? AND project = ?`, ); const unassignReportMaterialEvents = db.query( `UPDATE material_events SET digestReportId = NULL WHERE digestReportId = ? AND project = ?`, ); const markReportMaterialEventsPossibleRepeat = db.query( `UPDATE material_events SET summary = CASE WHEN summary NOT LIKE 'POSSIBLE REPEAT%' THEN 'POSSIBLE REPEAT of report ' || ? || ': ' || summary ELSE summary END WHERE digestReportId = ? AND project = ?`, ); const markReportHeldNoticesPossibleRepeat = db.query( `UPDATE held_notices SET summary = CASE WHEN summary NOT LIKE 'POSSIBLE REPEAT%' THEN 'POSSIBLE REPEAT of report ' || ? || ': ' || summary ELSE summary END, detail = CASE WHEN detail NOT LIKE 'This handoff may already%' THEN 'This handoff may already have reached Telegram before its outcome was lost.\n\n' || detail ELSE detail END WHERE digestReportId = ? AND project = ? AND withdrawnAt IS NULL`, ); const unassignReportHeldNotices = db.query( `UPDATE held_notices SET digestReportId = NULL, digestedAt = NULL WHERE digestReportId = ? AND project = ? AND withdrawnAt IS NULL`, ); const selectHeldNoticeLifecycle = db.query( `SELECT id, project, summary, detail, digestedAt, digestReportId, withdrawnAt FROM held_notices WHERE id = ?`, ); const withdrawHeldNoticeRow = db.query< unknown, [number, string, string, string, string] >( `UPDATE held_notices SET withdrawnAt = ?, withdrawnBy = ?, withdrawReason = ? WHERE id = ? AND project = ? AND withdrawnAt IS NULL AND ( (digestReportId IS NULL AND digestedAt IS NULL) OR EXISTS ( SELECT 1 FROM reports WHERE reports.id = held_notices.digestReportId AND reports.state IN ('failed', 'withdrawn') ) )`, ); const insertHandoffWithdrawal = db.query< unknown, [string, string, string, string, string, number, string, string] >( `INSERT INTO handoff_withdrawals (id, project, target, actor, reason, at, summary, detail) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectHandoffWithdrawal = db.query( `SELECT * FROM handoff_withdrawals WHERE id = ?`, ); const selectHandoffWithdrawals = db.query( `SELECT * FROM handoff_withdrawals WHERE project = ? ORDER BY at DESC, rowid DESC LIMIT ?`, ); const deletePendingMaterialReport = db.query( `DELETE FROM reports WHERE id = ? AND project = ? AND kind = 'material' AND state = 'pending'`, ); const deletePendingAvailabilityReport = db.query( `DELETE FROM reports WHERE id = ? AND project = ? AND kind = 'digest' AND state = 'pending' AND dedupeKey LIKE 'availability/%'`, ); const selectPendingAvailabilityReports = db.query< { id: string; ambiguous: number }, [string] >( `SELECT id, ambiguous FROM reports WHERE project = ? AND kind = 'digest' AND state = 'pending' AND dedupeKey LIKE 'availability/%' ORDER BY createdAt ASC, rowid ASC`, ); const releaseAvailabilityNotices = db.query< unknown, [number, string, string, string, string, string] >( `UPDATE held_notices SET digestReportId = NULL, digestedAt = NULL, summary = CASE WHEN ? = 1 AND summary NOT LIKE 'POSSIBLE REPEAT%' THEN 'POSSIBLE REPEAT of catch-up ' || ? || ': ' || summary ELSE summary END WHERE project = ? AND digestReportId = ? AND withdrawnAt IS NULL AND EXISTS ( SELECT 1 FROM reports WHERE id = ? AND project = ? AND kind = 'digest' AND state = 'pending' AND dedupeKey LIKE 'availability/%' )`, ); const selectAvailabilityReservationCandidate = db.query<{ id: string }, [string]>( `SELECT held_notices.id FROM held_notices WHERE held_notices.project = ? AND held_notices.releaseOnAvailable = 1 AND held_notices.withdrawnAt IS NULL AND ( (held_notices.digestReportId IS NULL AND held_notices.digestedAt IS NULL) OR EXISTS ( SELECT 1 FROM reports WHERE reports.id = held_notices.digestReportId AND ( reports.state IN ('failed', 'withdrawn') OR ( reports.state = 'pending' AND reports.kind = 'digest' AND reports.dedupeKey LIKE 'availability/%' ) ) ) ) ORDER BY held_notices.createdAt ASC, held_notices.rowid ASC LIMIT 1`, ); const selectAvailabilityReservation = db.query< { expiresAt: number }, [string, string, string] >( `SELECT expiresAt FROM availability_digest_reservations WHERE project = ? AND cycleKey = ? AND firstNoticeId = ?`, ); const insertAvailabilityReservation = db.query< unknown, [string, string, string, number] >( `INSERT OR IGNORE INTO availability_digest_reservations (project, cycleKey, firstNoticeId, expiresAt) VALUES (?, ?, ?, ?)`, ); const countActiveAvailabilityReservations = db.query< { count: number }, [string, number] >( `SELECT COUNT(*) AS count FROM availability_digest_reservations WHERE project = ? AND expiresAt > ?`, ); const deleteAvailabilityReservationForNotice = db.query< unknown, [string, string] >( `DELETE FROM availability_digest_reservations WHERE project = ? AND firstNoticeId = ?`, ); // Held notices (#229) and ordinary material events (#274) stay separate // sources. Both are associated with one accepted digest by exact row id. const insertHeldNotice = db.query( `INSERT INTO held_notices (id, project, category, summary, detail, createdAt, releaseOnAvailable, urgent, digestedAt) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL) ON CONFLICT(id) DO NOTHING`, ); const insertHeldNoticeDraft = (notice: HeldNoticeDraft): boolean => insertHeldNotice.run( notice.id ?? crypto.randomUUID(), notice.project, notice.category, notice.summary, notice.detail, notice.createdAt, notice.releaseOnAvailable === true ? 1 : 0, notice.urgent === true ? 1 : 0, ).changes === 1; const deferPendingReportToNoticeTx = db.transaction( (id: string, notice: HeldNoticeDraft): boolean => { if (deletePendingMaterialReport.run(id, notice.project).changes !== 1) return false; if (!insertHeldNoticeDraft(notice)) { throw new Error(`held notice ${notice.id ?? "(generated)"} already exists`); } return true; }, ); const releasePendingAvailabilityReportRecord = ( id: string, project: string, possibleRepeat = false, ): boolean => { if ( releaseAvailabilityNotices.run(possibleRepeat ? 1 : 0, id, project, id, id, project) .changes === 0 ) { return false; } if (deletePendingAvailabilityReport.run(id, project).changes !== 1) { throw new Error(`availability catch-up ${id} changed during release`); } return true; }; const releasePendingAvailabilityReportTx = db.transaction( releasePendingAvailabilityReportRecord, ); const reserveAvailabilityDigestTx = db.transaction( (project: string, cycleKey: string, at: number, expiresAt: number): boolean => { const candidate = selectAvailabilityReservationCandidate.get(project); if (candidate === null) return false; if (selectAvailabilityReservation.get(project, cycleKey, candidate.id) !== null) { return false; } if ( insertAvailabilityReservation.run(project, cycleKey, candidate.id, expiresAt).changes === 0 ) { return false; } for (const row of selectPendingAvailabilityReports.all(project)) { releasePendingAvailabilityReportRecord(row.id, project, row.ambiguous !== 0); } return true; }, ); type HeldNoticeRow = { id: string; category: string; summary: string; detail: string; createdAt: number; releaseOnAvailable: number; urgent: number; }; const undigestedNoticeWhere = `project = ? AND withdrawnAt IS NULL AND ( (digestReportId IS NULL AND digestedAt IS NULL) OR EXISTS (SELECT 1 FROM reports WHERE reports.id = held_notices.digestReportId AND reports.state IN ('failed', 'withdrawn')) )`; const selectUndigestedNotices = db.query( `SELECT id, category, summary, detail, createdAt, releaseOnAvailable, urgent FROM held_notices WHERE ${undigestedNoticeWhere} ORDER BY createdAt ASC, rowid ASC LIMIT ?`, ); const selectAvailabilityHeldNotices = db.query( `SELECT id, category, summary, detail, createdAt, releaseOnAvailable, urgent FROM held_notices WHERE ${undigestedNoticeWhere} AND releaseOnAvailable = 1 ORDER BY createdAt ASC, rowid ASC LIMIT ?`, ); const selectHeldNoticeBacklog = db.query< { count: number; availabilityCount: number | null; oldestAt: number | null; digestOldestAt: number | null; availabilityOldestAt: number | null; }, [string] >( `SELECT COUNT(*) AS count, SUM(CASE WHEN releaseOnAvailable = 1 THEN 1 ELSE 0 END) AS availabilityCount, MIN(createdAt) AS oldestAt, MIN(CASE WHEN releaseOnAvailable = 0 THEN createdAt END) AS digestOldestAt, MIN(CASE WHEN releaseOnAvailable = 1 THEN createdAt END) AS availabilityOldestAt FROM held_notices WHERE project = ? AND withdrawnAt IS NULL AND ( (digestReportId IS NULL AND digestedAt IS NULL) OR EXISTS (SELECT 1 FROM reports WHERE reports.id = held_notices.digestReportId AND reports.state IN ('failed', 'withdrawn')) )`, ); const assignHeldNoticeToDigest = db.query( `UPDATE held_notices SET digestedAt = ?, digestReportId = ? WHERE project = ? AND id = ? AND withdrawnAt IS NULL AND ( (digestReportId IS NULL AND digestedAt IS NULL) OR EXISTS (SELECT 1 FROM reports WHERE reports.id = held_notices.digestReportId AND reports.state IN ('failed', 'withdrawn')) )`, ); const insertMaterialEvent = db.query( `INSERT INTO material_events (id, project, category, summary, evidence, occurredAt, recordedAt, digestReportId) VALUES (?, ?, ?, ?, ?, ?, ?, NULL)`, ); const selectMaterialEvent = db.query( `SELECT * FROM material_events WHERE id = ?`, ); const selectUndigestedMaterialEvents = db.query( `SELECT * FROM material_events WHERE project = ? AND ( digestReportId IS NULL OR EXISTS (SELECT 1 FROM reports WHERE reports.id = material_events.digestReportId AND reports.state IN ('failed', 'withdrawn')) ) ORDER BY occurredAt ASC, recordedAt ASC, rowid ASC LIMIT ?`, ); const selectMaterialEventBacklog = db.query<{ count: number; oldestAt: number | null }, [string]>( `SELECT COUNT(*) AS count, MIN(occurredAt) AS oldestAt FROM material_events WHERE project = ? AND ( digestReportId IS NULL OR EXISTS (SELECT 1 FROM reports WHERE reports.id = material_events.digestReportId AND reports.state IN ('failed', 'withdrawn')) )`, ); const assignMaterialEventToDigest = db.query( `UPDATE material_events SET digestReportId = ? WHERE project = ? AND id = ? AND ( digestReportId IS NULL OR EXISTS (SELECT 1 FROM reports WHERE reports.id = material_events.digestReportId AND reports.state IN ('failed', 'withdrawn')) )`, ); // INSERT OR IGNORE, not plain INSERT (Phase 4). For an operator's own idea // the id is random and the clause never fires; for a mined signal the id is // derived from project + source, so the second filing of a signal already on // file collides with its own row and does nothing. Crucially "does nothing" // includes not touching `state`: an UPSERT here would resurrect an item the // operator dismissed on every mining pass, which is the failure mode this // shape exists to prevent. const insertIntake = db.query( `INSERT OR IGNORE INTO intake_items (id, project, text, createdAt, state, source) VALUES (?, ?, ?, ?, 'pending', ?)`, ); // Read back what is actually on file after the insert, which for a collision // is the row that was already there. The caller gets the truth (including a // `dismissed` state and the original `createdAt`) rather than the draft it // hoped to write. const selectIntakeById = db.query( `SELECT * FROM intake_items WHERE id = ?`, ); const selectPendingIntake = db.query( `SELECT * FROM intake_items WHERE project = ? AND state = 'pending' ORDER BY createdAt ASC, rowid ASC`, ); const groomIntake = db.query( `UPDATE intake_items SET state = 'groomed', issueUrl = ?, groomedAt = ? WHERE id = ? AND state = 'pending'`, ); const dismissIntake = db.query( `UPDATE intake_items SET state = 'dismissed', issueUrl = NULL, groomedAt = NULL WHERE id = ? AND state = 'pending'`, ); // The newest digest dedupe key a project has actually run toward — an // `updatedAt`-newest row whose key is a digest prefix and that did not end in // failure. The CLI refuses an off-schedule digest by comparing this against // today's key under `digestDue` (#229). const selectLastDigestKey = db.query<{ key: string | null }, [string]>( `SELECT dedupeKey AS key FROM reports WHERE project = ? AND dedupeKey LIKE 'digest:%' AND state NOT IN ('failed', 'withdrawn') ORDER BY createdAt DESC LIMIT 1`, ); const claimReportRow = db.query( `UPDATE reports SET state = 'sending', attemptId = ?, attempts = attempts + 1, updatedAt = ? WHERE id = ? AND state = 'pending' AND nextAttemptAt <= ?`, ); const deliverReportRow = db.query( `UPDATE reports SET state = 'delivered', attemptId = NULL, messageId = ?, messageIds = ?, deliveredAt = ?, updatedAt = ?, lastError = NULL WHERE id = ? AND state = 'sending' AND attemptId = ?`, ); // Mid-split progress, written before the next part ships: the first // `sentParts` parts of the fingerprinted message are confirmed accepted, so // a crash or a definitive failure can only ever re-send the part that was in // flight (#566). Guarded by attempt id like every other transition. const markPartsSentRow = db.query( `UPDATE reports SET sentParts = ?, sentPartsHash = ?, updatedAt = ? WHERE id = ? AND state = 'sending' AND attemptId = ?`, ); const requeueReportRow = db.query( `UPDATE reports SET state = 'pending', attemptId = NULL, nextAttemptAt = ?, lastError = ?, updatedAt = ? WHERE id = ? AND state = 'sending' AND attemptId = ?`, ); // Stays `sending` and keeps its attempt id: the request is over but its // outcome is not known, which is precisely what `sending` means. `updatedAt` // is left alone so the stale window keeps running from when the request left. const uncertainReportRow = db.query( `UPDATE reports SET ambiguous = 1, lastError = ? WHERE id = ? AND state = 'sending' AND attemptId = ?`, ); const failReportRow = db.query( `UPDATE reports SET state = 'failed', attemptId = NULL, lastError = ?, updatedAt = ? WHERE id = ? AND state = 'sending' AND attemptId = ?`, ); const selectStaleSending = db.query( `SELECT * FROM reports WHERE project = ? AND state = 'sending' AND updatedAt <= ? ORDER BY createdAt ASC, rowid ASC`, ); const recoverSendingRows = db.query( `UPDATE reports SET state = 'pending', attemptId = NULL, ambiguous = 1, nextAttemptAt = ?, updatedAt = ? WHERE project = ? AND state = 'sending' AND updatedAt <= ?`, ); const selectOpenReports = db.query( `SELECT * FROM reports WHERE project = ? AND state IN ('pending', 'sending', 'failed') ORDER BY createdAt ASC, rowid ASC`, ); const enqueueReportRecord = (draft: ReportDraft): ReportEnqueue => { const report: ReportRecord = { id: newReportId(), project: draft.project, kind: draft.kind, body: draft.body, state: "pending", attempts: 0, ambiguous: false, ...(draft.dedupeKey === undefined ? {} : { dedupeKey: draft.dedupeKey }), createdAt: draft.at, updatedAt: draft.at, // Due immediately. No report has earned a backoff before its first send. nextAttemptAt: draft.at, }; const inserted = insertReport.run( report.id, report.project, report.kind, report.body, report.state, report.attempts, null, 0, toSql(draft.dedupeKey), report.createdAt, report.updatedAt, report.nextAttemptAt, null, null, null, ); if (inserted.changes > 0) return { report, deduped: false }; // `INSERT OR IGNORE` swallowed a conflict. A digest-key conflict is the // expected path; a random 48-bit id collision has no matching key and fails // closed below instead of returning an unrelated report. const existing = draft.dedupeKey === undefined ? null : selectReportByDedupe.get(draft.project, draft.dedupeKey); if (existing === null) { throw new Error(`report ${report.id} could not be enqueued for ${draft.project}`); } return { report: toReport(existing), deduped: true }; }; const enqueueDigestReportRecord = ( draft: ReportDraft, materialEventIds: readonly string[], heldNoticeIds: readonly string[], ): ReportEnqueue => { if (draft.kind !== "digest") { throw new Error("digest handoff needs kind digest"); } const enqueued = enqueueReportRecord(draft); if (enqueued.deduped) return enqueued; for (const id of new Set(materialEventIds)) { if (assignMaterialEventToDigest.run(enqueued.report.id, draft.project, id).changes !== 1) { throw new Error(`material event ${id} is not owed by project ${draft.project}`); } } for (const id of new Set(heldNoticeIds)) { if (assignHeldNoticeToDigest.run(draft.at, enqueued.report.id, draft.project, id).changes !== 1) { throw new Error(`held notice ${id} is not owed by project ${draft.project}`); } if (!draft.dedupeKey?.startsWith("availability/")) { deleteAvailabilityReservationForNotice.run(draft.project, id); } } return enqueued; }; const withdrawHandoffTx = db.transaction( ( id: string, project: string, actor: string, reason: string, at: number, ): HandoffWithdrawalResult => { const report = selectReport.get(id); if (report !== null) { if (report.project !== project) return { kind: "not-found" }; if (report.state !== "pending") { return { kind: "refused", target: "report", state: report.state }; } const noticeIds = selectReportNoticeIds.all(id, project).map((row) => row.id); if (withdrawPendingReportRow.run(at, actor, reason, at, id, project).changes !== 1) { const current = selectReport.get(id); return { kind: "refused", target: "report", state: current?.state ?? "not-found", }; } if (report.ambiguous !== 0) { markReportMaterialEventsPossibleRepeat.run(id, id, project); markReportHeldNoticesPossibleRepeat.run(id, id, project); } unassignReportMaterialEvents.run(id, project); unassignReportHeldNotices.run(id, project); for (const noticeId of noticeIds) { deleteAvailabilityReservationForNotice.run(project, noticeId); } const summary = report.body.split("\n", 1)[0]!.slice(0, 240); const withdrawal: HandoffWithdrawal = { id, project, target: "report", actor, reason, at, summary, detail: report.body, }; insertHandoffWithdrawal.run( id, project, "report", actor, reason, at, summary, report.body, ); return { kind: "withdrawn", withdrawal }; } const prior = selectHandoffWithdrawal.get(id); if (prior !== null) { return prior.project === project ? { kind: "refused", target: prior.target as HandoffWithdrawal["target"], state: "withdrawn" } : { kind: "not-found" }; } const notice = selectHeldNoticeLifecycle.get(id); if (notice === null || notice.project !== project) return { kind: "not-found" }; if (notice.withdrawnAt !== null) { return { kind: "refused", target: "notice", state: "withdrawn" }; } if (notice.digestReportId !== null) { const owner = selectReport.get(notice.digestReportId); if (owner === null || (owner.state !== "failed" && owner.state !== "withdrawn")) { return { kind: "refused", target: "notice", state: "assigned" }; } } else if (notice.digestedAt !== null) { return { kind: "refused", target: "notice", state: "digested" }; } if (withdrawHeldNoticeRow.run(at, actor, reason, id, project).changes !== 1) { return { kind: "refused", target: "notice", state: "changed" }; } deleteAvailabilityReservationForNotice.run(project, id); const withdrawal: HandoffWithdrawal = { id, project, target: "notice", actor, reason, at, summary: notice.summary, detail: notice.detail, }; insertHandoffWithdrawal.run( id, project, "notice", actor, reason, at, notice.summary, notice.detail, ); return { kind: "withdrawn", withdrawal }; }, ); const enqueueDigestReportTx = db.transaction(enqueueDigestReportRecord); const enqueueAvailabilityReportTx = db.transaction( (draft: ReportDraft, heldNoticeIds: readonly string[]): ReportEnqueue | undefined => { if (draft.kind !== "digest" || !draft.dedupeKey?.startsWith("availability/")) { throw new Error("availability handoff needs an availability digest key"); } if ( (countActiveAvailabilityReservations.get(draft.project, draft.at)?.count ?? 0) > 0 ) { return undefined; } return enqueueDigestReportRecord(draft, [], heldNoticeIds); }, ); const insertDecision = db.query< never, [string, string, string, string, SqlValue, number, number, SqlValue, string, SqlValue, SqlValue] >( `INSERT INTO decisions (id, project, kind, question, blocks, askedAt, expiresAt, condition, state, groupId, specIssue) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectDecisionGroup = db.query( `SELECT * FROM decisions WHERE groupId = ? ORDER BY askedAt ASC, rowid ASC`, ); const selectDecision = db.query(`SELECT * FROM decisions WHERE id = ?`); const selectOpenDecisions = db.query( `SELECT * FROM decisions WHERE project = ? AND state = 'open' ORDER BY askedAt ASC, rowid ASC`, ); const selectDueDecisions = db.query( `SELECT * FROM decisions WHERE project = ? AND state = 'open' AND expiresAt <= ? ORDER BY askedAt ASC, rowid ASC`, ); // `state = 'open'` in the WHERE is the double-resolve guard: the second call // changes no rows and returns false rather than overwriting the first answer. const resolveDecisionRow = db.query( `UPDATE decisions SET state = ?, resolution = ?, resolvedAt = ? WHERE id = ? AND state = 'open'`, ); const expireDecisionRows = db.query( `UPDATE decisions SET state = 'expired', resolvedAt = ?, resolution = 'expired unanswered' WHERE project = ? AND state = 'open' AND expiresAt <= ?`, ); // Set once per head: a condition that flickers must not keep moving its own // timestamp, or the digest's "act on this now" would reset to looking new // every tick. A `pr-checks-green` / `pr-review-ready` row whose PR head // moved is cleared first and re-marks here against the new head (`head` // binds the verdict to the exact commit it was observed at, #808). // A watch's deadline starts when its condition fires, not when it was opened // (#966). // // `DECISION_WATCH_NEVER_EXPIRES` is right for the waiting phase, and its // reasoning is quoted where it is defined: a condition that has not fired is // not a question nobody answered, and expiring it could drop a release. But // the instant `conditionMetAt` is stamped the row stops waiting — it becomes // an actionable item, which is exactly what a question's TTL bounds. Measured // on this fleet 2026-08-23: three watches had carried // "[CONDITION MET — act on this now]" for 10-12 hours over work that was // entirely closed, and would have carried it forever. // // Only watches are touched: a question's `expiresAt` already runs from // `askedAt` and must keep doing so, so the deadline is written only where it // is currently infinite. `kind = 'watch'` is the guard rather than a value // comparison, because a question with a condition is still a question. const markConditionMetRow = db.query( `UPDATE decisions SET conditionMetAt = ?, conditionHead = ?, expiresAt = CASE WHEN kind = 'watch' THEN ? ELSE expiresAt END WHERE id = ? AND state = 'open' AND conditionMetAt IS NULL`, ); // The inverse of the mark, for the two conditions whose observation goes // stale: a `pr-checks-green` / `pr-review-ready` binding is about a commit, // so a head change drops both the timestamp and the binding and the row // returns to pending (#808). Returns false when no met state existed to // drop, so re-runs are a no-op. // // The deadline goes back with it (#966): a PR that went red again is waiting, // not ignored, and leaving a countdown on it would expire a watch whose // condition is once again unfired. const clearConditionMetRow = db.query( `UPDATE decisions SET conditionMetAt = NULL, conditionHead = NULL, expiresAt = CASE WHEN kind = 'watch' THEN ? ELSE expiresAt END WHERE id = ? AND state = 'open' AND conditionMetAt IS NOT NULL`, ); const insertVerbLedger = db.query( `INSERT INTO verb_ledger (id, project, runId, issue, verb, role, args, decision, refusal, detail, sha, at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); const selectVerbLedger = db.query( `SELECT * FROM verb_ledger WHERE project = ? AND (? = '' OR runId = ?) AND (? = 0 OR issue = ?) ORDER BY at DESC, rowid DESC LIMIT ?`, ); const insertDaemonStop = db.query( `INSERT INTO daemon_stops (id, at, controlPath, callerPid, callerUid, role, scope, project, daemonPid, runtimeDir, affected, reason, unattributed) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); // `rowid` breaks the tie when two requests land in the same millisecond, so // "newest first" never silently hides one stop behind another. const selectLatestDaemonStop = db.query( `SELECT * FROM daemon_stops ORDER BY at DESC, rowid DESC LIMIT 1`, ); // Breaking a stale claim and taking it must be one transaction, or two // daemons both see the stale row, both delete it, and both insert. const breakStaleMergeLock = db.query( `DELETE FROM merge_locks WHERE project = ? AND at <= ?`, ); const claimMergeLock = db.query( `INSERT OR IGNORE INTO merge_locks (project, holder, prUrl, at) VALUES (?, ?, ?, ?)`, ); const dropMergeLock = db.query( `DELETE FROM merge_locks WHERE project = ? AND holder = ?`, ); const selectMergeLock = db.query( `SELECT * FROM merge_locks WHERE project = ?`, ); const takeMergeLock = db.transaction( (project: string, holder: string, prUrl: string, now: number, staleAt: number): boolean => { breakStaleMergeLock.run(project, staleAt); return claimMergeLock.run(project, holder, prUrl, now).changes > 0; }, ); // The durable review-revision request outbox (#677): the orchestrator's verb // writes the row (findings, head, round, target session) and the daemon's // next dispatch pass wakes it. The duplicate in-flight guard is the insert: // a pending row for the same run refuses the second request atomically, and // once dispatched the run row's own state (pushed-green → running) is the // in-flight marker. const insertReviewRevision = db.query( `INSERT INTO review_revisions (id, project, runId, issue, prUrl, headSha, findings, round, reason, sessionFile, requestedAt, dispatchedAt, settledAt, outcome) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL)`, ); const selectPendingReviewRevisions = db.query( `SELECT * FROM review_revisions WHERE project = ? AND dispatchedAt IS NULL AND settledAt IS NULL ORDER BY requestedAt ASC, rowid ASC`, ); const selectPendingReviewForRun = db.query( `SELECT * FROM review_revisions WHERE project = ? AND runId = ? AND dispatchedAt IS NULL AND settledAt IS NULL LIMIT 1`, ); const selectLatestReviewRound = db.query<{ n: number | null }, [string, string]>( `SELECT MAX(round) AS n FROM review_revisions WHERE project = ? AND runId = ?`, ); const markReviewRevisionDispatchedRow = db.query( `UPDATE review_revisions SET dispatchedAt = ? WHERE id = ?`, ); const settleReviewRevisionRow = db.query( `UPDATE review_revisions SET outcome = ?, settledAt = ? WHERE id = ?`, ); const selectUnsettledReviewRevisions = db.query( `SELECT * FROM review_revisions WHERE project = ? AND settledAt IS NULL ORDER BY requestedAt ASC, rowid ASC`, ); const requeueReviewRevisionRow = db.query( `UPDATE review_revisions SET dispatchedAt = NULL WHERE id = ?`, ); // The bounded infra retry (#903). One statement decides and writes: the // WHERE carries the bound, so two daemons cannot both spend the last retry // and a settled row can never be revived. `changes` is the decision. const retryReviewRevisionRow = db.query( `UPDATE review_revisions SET dispatchedAt = NULL, infraRetries = COALESCE(infraRetries, 0) + 1 WHERE id = ? AND settledAt IS NULL AND COALESCE(infraRetries, 0) < ?`, ); const selectReviewRevisionById = db.query( `SELECT * FROM review_revisions WHERE id = ?`, ); // The write-once launch decision (#1045). ONE guarded UPDATE decides and // writes, exactly as the infra retry above does, because a read-then-write // is precisely how two dispatch passes both "find no decision" and both // launch. `changes` is the answer: // // - launchMode IS NULL -> nobody has decided; this call wins. // - the guard triple matches -> the SAME decision again (a retry, a // restart deciding identically). The row // is re-affirmed rather than refused, so // a dispatch retry is idempotent. // - anything else -> 0 changes, and the caller must read the // recorded decision and honour it. // // `IS` rather than `=` on the guard: NULL = NULL is NULL in SQL, so `=` // would refuse a re-affirmation of a decision that carried no requested // model or no origin lineage — the shape a legacy-session round has. // // launchDecidedAt is COALESCE'd, never overwritten: a re-affirmation must // leave the moment of the original decision alone, or a retry loop would // make an old decision look freshly made in every audit that reads it. // // The SET touches decision columns only. The issue, run, round, branch, PR // URL, reviewed head, findings, charges and settlement history are the // round's identity and are never written by this lane. const recordReviewRevisionLaunchRow = db.query< unknown, [string, string | null, string | null, number, string, string, string, string | null, string | null] >( `UPDATE review_revisions SET launchMode = ?, requestedModel = ?, originSessionRef = ?, launchDecidedAt = COALESCE(launchDecidedAt, ?) WHERE id = ? AND project = ? AND (launchMode IS NULL OR (launchMode IS ? AND requestedModel IS ? AND originSessionRef IS ?))`, ); // The fresh correction's own session lineage (#1045), attachable only to a // round that was decided `fresh-correction`: a resumed round IS its origin // session, so a separate correction lineage on one would name a session that // never existed. The mode is bound rather than inlined so the vocabulary // lives in exactly one place — {@link ReviewLaunchMode}. // // Deliberately not write-once: an infra-killed round relaunches under the // same decision, and the newest fresh session is then the truth about what // is running. `resolvedModel` is COALESCE'd from the argument side, so a // later call that does not know the resolution cannot erase a known one. const recordReviewCorrectionSessionRow = db.query( `UPDATE review_revisions SET correctionSessionRef = ?, resolvedModel = COALESCE(?, resolvedModel) WHERE id = ? AND project = ? AND launchMode = ?`, ); const selectReviewRevisionForProject = db.query( `SELECT * FROM review_revisions WHERE id = ? AND project = ?`, ); // The exact-head merge gate (#888): every row that stands between a PR and // a merge at one head — anything not yet settled (queued, or dispatched and // crashed mid-review) plus anything settled `failed` there. `skipped`, // `revised` and `pending` outcomes never block: a skip is an assessment // that never ran (the escalation for one says "merge it as it stands"), and // a revised/pending outcome belongs to a worker that pushed past the // reviewed head. Head comparison is case-insensitive; prUrl is matched // exactly — every writer records it from the same tracker payloads. // // A recorded clearance (#913) is the one disposition available at an // *unchanged* head, and it applies strictly backwards: the NOT EXISTS // requires the clearance to be at or after the row's own last activity // (`settledAt` when the round finished, else `requestedAt`). Without that // comparison a single clearance would bless every future finding at that // head — a strictly worse hole than the deadlock it fixes — so the // timestamp is the load-bearing half of this clause, not decoration. The // `outcome = 'failed'` term stays exactly as it was: a clearance is how // failed evidence is dispositioned, never by widening what counts as // blocking. const selectMergeBlockingReviews = db.query( `SELECT * FROM review_revisions WHERE project = ? AND prUrl = ? AND lower(headSha) = lower(?) AND (settledAt IS NULL OR outcome = 'failed') AND NOT EXISTS ( SELECT 1 FROM review_clearances c WHERE c.project = review_revisions.project AND c.prUrl = review_revisions.prUrl AND lower(c.headSha) = lower(review_revisions.headSha) AND c.at >= COALESCE(review_revisions.settledAt, review_revisions.requestedAt) ) ORDER BY requestedAt ASC, rowid ASC`, ); const insertReviewClearance = db.query( `INSERT INTO review_clearances (id, project, prUrl, headSha, at, by, reason) VALUES (?, ?, ?, ?, ?, ?, ?)`, ); const selectReviewClearances = db.query( `SELECT * FROM review_clearances WHERE project = ? AND prUrl = ? AND lower(headSha) = lower(?) ORDER BY at DESC, rowid DESC`, ); // The review-ceiling adjudication lifecycle (#874). Head comparison is // case-insensitive everywhere, exactly as the merge gate and the clearance // lookup are: a 40-hex SHA read from two GitHub surfaces can differ in case // and mean the same commit. const insertReviewAdjudication = db.query( `INSERT INTO review_adjudications (id, project, issue, prUrl, headSha, role, provenance, state, evidence, disposition, requestedAt, dispatchedAt, settledAt, findings) VALUES (?, ?, ?, ?, ?, ?, NULL, 'pending', NULL, NULL, ?, NULL, NULL, ?)`, ); const selectAdjudicationForHead = db.query( `SELECT * FROM review_adjudications WHERE project = ? AND prUrl = ? AND lower(headSha) = lower(?)`, ); // The non-terminal row for one PR, whatever head it is on. This is what makes // a moved head refuse rather than open a second live adjudication. const selectOpenAdjudicationForPr = db.query( `SELECT * FROM review_adjudications WHERE project = ? AND prUrl = ? AND settledAt IS NULL ORDER BY requestedAt ASC, rowid ASC LIMIT 1`, ); const selectAdjudicationById = db.query( `SELECT * FROM review_adjudications WHERE id = ?`, ); const selectOpenAdjudications = db.query( `SELECT * FROM review_adjudications WHERE project = ? AND settledAt IS NULL ORDER BY requestedAt ASC, rowid ASC`, ); // Every review round recorded for one PR, oldest first (#932): the history the // adjudicator is shown. Keyed by PR rather than by run so a continuation that // inherited the PR does not hide its predecessor's findings. const selectRevisionsForPr = db.query( `SELECT * FROM review_revisions WHERE project = ? AND prUrl = ? ORDER BY round ASC, requestedAt ASC, rowid ASC`, ); const selectAdjudicationsForPr = db.query( `SELECT * FROM review_adjudications WHERE project = ? AND prUrl = ? ORDER BY requestedAt DESC, rowid DESC`, ); // Dispatch: pending -> running, recording the launch provenance. The WHERE // carries the transition, so two dispatchers racing one row cannot both // launch — the loser changes nothing and reads `false`. const markAdjudicationRunningRow = db.query( `UPDATE review_adjudications SET state = 'running', provenance = ?, dispatchedAt = ? WHERE id = ? AND state = 'pending' AND settledAt IS NULL`, ); // Settle: only from a non-terminal state, and only once. An invalid // transition changes nothing and preserves whatever the row already said, // which is the diagnostic evidence a later reader needs. const settleAdjudicationRow = db.query( `UPDATE review_adjudications SET state = ?, evidence = COALESCE(?, evidence), settledAt = ? WHERE id = ? AND settledAt IS NULL`, ); const recordAdjudicationDispositionRow = db.query( `UPDATE review_adjudications SET disposition = ? WHERE id = ?`, ); // Admission is a transaction for the same reason the revision enqueue is: the // CLI and the daemon are separate processes with separate connections, so a // deferred read-then-write would let both see "no row" and both insert. It // begins IMMEDIATE, taking the write lock before the read, and the UNIQUE // index on (project, prUrl, headSha) is the structural backstop underneath. const openAdjudicationTx = db.transaction( ( id: string, project: string, issue: number, prUrl: string, headSha: string, role: string, requestedAt: number, findings: string | undefined, ): ReviewAdjudicationAdmission => { const sameHead = selectAdjudicationForHead.get(project, prUrl, headSha); if (sameHead !== null && sameHead !== undefined) { return { kind: "existing", record: toReviewAdjudication(sameHead) }; } const open = selectOpenAdjudicationForPr.get(project, prUrl); if (open !== null && open !== undefined) { return { kind: "refused", block: "in-flight-different-head", record: toReviewAdjudication(open), }; } insertReviewAdjudication.run(id, project, issue, prUrl, headSha, role, requestedAt, findings ?? null); return { kind: "created", record: toReviewAdjudication(selectAdjudicationById.get(id)!) }; }, ); // The states the review verb admits and the dispatch pass may therefore // claim (#795, #1101): a settled green run, a capped/failed run that pushed // one (`failed` / `killed`), and — since #1101 — a `stopped` run that had // already pushed when the operator stopped it. The set is closed on purpose // — a live row (`running` / `claimed`) is already doing its own work, a // `pushed-pending` PR is not green yet, and a `blocked` / `orphaned` / // `merged` row is not work the orchestrator returned for revision. // // The terminal leg does two things in the same atomic statement (#795 review // rounds 1-2): it PRESERVES the budget charge the row's terminal event was // charging — added to a per-category INTEGER column, so repeated // same-category terminal events across consecutive review rounds each keep // their charge and `maxAttemptsPerIssue` / `maxContinuationsPerIssue` keep // bounding the lifecycle — and it CLEARS the lifecycle recovery metadata // (`failureClass` / `recoveryAction` / `recoveredAt`), so the resumed // execution's own terminal failure is classified and recovered fresh instead // of being excluded from the classification sweep by the original terminal's // recovered state. The state branch stops counting the row the instant this // statement flips it to `running`, and the preserved column starts counting // it in the same statement: one event, one count, never both. const claimRunForReviewTerminal = db.query( `UPDATE runs SET state = 'running', endedAt = NULL, failureClass = NULL, recoveryAction = NULL, recoveredAt = NULL, failureCharges = COALESCE(failureCharges, 0) + CASE WHEN state = 'failed' AND (failureClass IS NULL OR failureClass NOT IN (${CHARGEABLE_FAILURE_EXCLUSIONS})) THEN 1 ELSE 0 END, continuationCharges = COALESCE(continuationCharges, 0) + CASE WHEN state = 'failed' AND failureClass IN ('returned-for-revision', 'model-empty-stop') THEN 1 WHEN state = 'killed' AND (failureClass IS NULL OR failureClass NOT IN (${CHARGEABLE_CONTINUATION_EXCLUSIONS})) THEN 1 ELSE 0 END WHERE id = ? AND state IN ('failed', 'killed')`, ); // The settled-green leg: a pushed row carries no charge and no recovery // metadata, so the claim stays exactly what it always was. const claimRunForReviewGreen = db.query( `UPDATE runs SET state = 'running', endedAt = NULL WHERE id = ? AND state = 'pushed-green'`, ); // The stopped leg (#1101): reclaiming an operator-stopped run spends no // budget — an operator stop never charged one — and `prUrl IS NOT NULL` // keeps a stop before any push unclaimable: only a run whose row records // the reviewed PR may be woken for its revision. const claimRunForReviewStopped = db.query( `UPDATE runs SET state = 'running', endedAt = NULL WHERE id = ? AND state = 'stopped' AND prUrl IS NOT NULL`, ); // Exactly one leg can match (the row is in exactly one state at the moment // of the claim), a repeated claim matches none, and the three statements // plus the read of `changes` are one transaction — the dispatch pass can // never see a half-claimed row. const claimRunForReviewTx = db.transaction((runId: string): boolean => { if (claimRunForReviewTerminal.run(runId).changes > 0) return true; if (claimRunForReviewGreen.run(runId).changes > 0) return true; return claimRunForReviewStopped.run(runId).changes > 0; }); // Appending is a single-statement concatenation so even a write from a // second connection cannot lose one side of the merge: SQLite holds the // row's write lock for the whole UPDATE and re-reads the current value, so // the second append lands on the first's result rather than on the read it // took before the first committed. const appendReviewFindingsRow = db.query( `UPDATE review_revisions SET findings = findings || ? WHERE id = ?`, ); // The decision and the write share one transaction (#677, #786): a same-run, // same-head finding folds into the revision still waiting to dispatch // (appended — one round, one row, one worker attempt, and the resumed // session reads both findings), while a revision already pending at a // different head — a review of a moved PR — is refused rather than amended. // Two concurrent same-head requests therefore cannot both create a row or // overwrite each other: whichever commits first, the other sees its pending // row inside this transaction and appends to it. The verb turns the `refused` // outcome into the in-flight refusal. // // The transaction is begun IMMEDIATE — the write lock is taken BEFORE the // pending-row read. The CLI and the daemon are separate processes with // separate SQLite connections; under a deferred read-then-write both could // read "no pending row" before either wrote, and the second INSERT would // then fail with SQLITE_BUSY_SNAPSHOT instead of re-reading the winner's // committed row — losing the second finding as an untyped verb error. With // IMMEDIATE, the losing connection blocks on the write lock (bounded by the // 5s busy timeout) and, once the winner commits, reads the fresh state and // appends to the winner's row. The unique pending-per-run index remains as // the schema-level backstop. const enqueueReviewRevisionTx = db.transaction( (draft: ReviewRevisionDraft): ReviewRevisionEnqueue => { const pending = selectPendingReviewForRun.get(draft.project, draft.runId); if (pending !== null) { if (pending.headSha !== draft.headSha) { return { kind: "refused", block: "pending-different-head" }; } const merged = `${pending.findings}\n\n${draft.findings}`; appendReviewFindingsRow.run(`\n\n${draft.findings}`, pending.id); return { kind: "appended", record: { ...toReviewRevision(pending), findings: merged } }; } const record: ReviewRevisionRecord = { ...draft, id: crypto.randomUUID() }; insertReviewRevision.run( record.id, record.project, record.runId, record.issue, record.prUrl, record.headSha, record.findings, record.round, record.reason, record.sessionFile ?? null, record.requestedAt, ); return { kind: "created", record }; }, ); // The label projection outbox (#201). A decided label change is a row until // the tracker has applied it; the projector drains in id order, and one // issue's ops are atomic as a run — a failed op parks the rest of its issue // for the next pass rather than letting a later op land out of order. const insertLabelOp = db.query( `INSERT INTO label_ops (project, issue, op, label, createdAt) VALUES (?, ?, ?, ?, ?)`, ); const selectPendingLabelOps = db.query( `SELECT o.* FROM label_ops AS o WHERE o.project = ? AND o.nextAttemptAt <= ? AND NOT EXISTS ( SELECT 1 FROM label_ops AS e WHERE e.project = o.project AND e.issue = o.issue AND e.id < o.id ) ORDER BY o.id`, ); const selectPendingLabelOpsFor = db.query( `SELECT * FROM label_ops WHERE project = ? AND issue = ? ORDER BY id`, ); const selectOldestPendingLabelOp = db.query( `SELECT * FROM label_ops WHERE project = ? ORDER BY createdAt ASC, id ASC LIMIT 1`, ); const countLabelOps = db.query<{ n: number }, [string]>( `SELECT COUNT(*) AS n FROM label_ops WHERE project = ?`, ); const deleteLabelOp = db.query(`DELETE FROM label_ops WHERE id = ?`); const bumpLabelOpAttempt = db.query( `UPDATE label_ops SET attempts = attempts + 1, lastError = ?, nextAttemptAt = ? WHERE id = ?`, ); // Rate-limit refusals defer WITHOUT counting an attempt: a shared outage is // not the op's fault, so it must not burn the backoff escalation (#208). const parkLabelOp = db.query( `UPDATE label_ops SET lastError = ?, nextAttemptAt = ? WHERE id = ?`, ); // Coalesce: if the LATEST pending op for the same project+issue+label is // already the same operation, adding another is a duplicate that would sit // parked behind the oldest pending one, growing unboundedly (a reconcile // re-offering a removal the tracker keeps refusing). Only the latest matters: // a `remove → add → remove` sequence must keep the final remove even though // the first remove is still owed, because the add in between is a real // transition the final remove answers. Opposite transitions are never // coalesced. The check and insert share the transaction so they cannot race // (#201). const selectLatestLabelOp = db.query<{ op: "add" | "remove" }, [string, number, string]>( `SELECT op FROM label_ops WHERE project = ? AND issue = ? AND label = ? ORDER BY id DESC LIMIT 1`, ); /** The coalescing outbox insert, without a transaction of its own: the caller * owns one, so a decision that must be durable *with* its label transition * (the #807 escalation) can commit both or neither. */ const enqueueLabelOpRows = ( rows: readonly { project: string; issue: number; op: "add" | "remove"; label: string; createdAt: number; }[], ): void => { for (const row of rows) { const latest = selectLatestLabelOp.get(row.project, row.issue, row.label); if (latest !== undefined && latest !== null && latest.op === row.op) continue; insertLabelOp.run(toSql(row.project), row.issue, row.op, row.label, row.createdAt); } }; const enqueueLabelOpsTx = db.transaction(enqueueLabelOpRows); /** * The whole #807 escalation as one durable decision: the chain's one-shot * marker, the failed → queue label swap that spends it, and the settling * run's recovery stamp. * * They commit together or not at all, which is the only way the promise * survives a restart. As three separate writes led by the marker, a daemon * that died right after it would come back to a chain reading as * already-escalated with no queue label owed: the continuation silently lost, * and the next cap on that chain judged a second one. * * A run id with no row is a caller bug, not a state: throwing rolls the * marker back rather than recording an escalation nothing can ever spend. */ const escalateModelTx = db.transaction( ( project: string, issue: number, model: string, failureClass: string, runId: string, swapFrom: string, queueLabel: string, at: number, ): boolean => { if (claimModelEscalation.run(project, issue, model, failureClass, runId, at).changes === 0) { return false; } enqueueLabelOpRows([ { project, issue, op: "remove", label: swapFrom, createdAt: at }, { project, issue, op: "add", label: queueLabel, createdAt: at }, ]); if (stampRecovered.run(at, runId).changes === 0) { throw new Error(`escalateModel: run ${runId} does not exist`); } return true; }, ); const recordFriction = (project: string, observation: FrictionObservation): void => { if ( !Number.isSafeInteger(observation.occurrences) || observation.occurrences < 1 || !Number.isSafeInteger(observation.at) || observation.at < 0 ) { return; } const day = new Date(observation.at).toISOString().slice(0, 10); const prior = selectFrictionRollup.get(project, day, observation.kind); const issues = boundedUnique( prior === null ? [] : parseNumberList(prior.issues), [ ...(observation.issues ?? []), ...(observation.issue === undefined || observation.issue <= 0 ? [] : [observation.issue]), ], FRICTION_ISSUE_SAMPLES, ); const sample = observation.sample?.replace(/\s+/g, " ").trim().slice(0, FRICTION_TEXT_LIMIT); const samples = boundedUnique( prior === null ? [] : parseTextList(prior.samples), sample === undefined || sample.length === 0 ? [] : [sample], FRICTION_TEXT_SAMPLES, ); upsertFrictionRollup.run( project, day, observation.kind, (prior?.observations ?? 0) + 1, (prior?.occurrences ?? 0) + observation.occurrences, JSON.stringify(issues), JSON.stringify(samples), Math.max(prior?.latestAt ?? 0, observation.at), ); }; const recordGhRefusal = (at: number): void => { // Same guard as `recordFriction`: a bad clock must not corrupt the store. if (!Number.isSafeInteger(at) || at < 0) return; recordGhRefusalTx(at); }; const insertRunRecord = (r: Omit): RunRecord => { const record: RunRecord = { ...r, id: crypto.randomUUID() }; insertRun.run( record.id, record.project, record.issue, record.repo, record.branch, record.worktree, record.state, record.attempt, record.turns, record.maxTurns, record.spendUsd, toSql(record.spendReservedUsd), toSql(record.sessionFile), toSql(record.resumedFromRunId), toSql(record.lane), toSql(record.prUrl), toSql(record.headSha), toSql(record.mergeSha), toSql(record.baseRef), toSql(record.baseCheck), toSql(record.baseCheckAt), toSql(record.salvageSha), toSql(record.salvageError), toSql(record.salvageAckAt), record.startedAt, toSql(record.endedAt), toSql(record.lastError), toSql(record.settlementFlags), toSql(record.report), toSql(record.graphTools), toSql(record.lastProgressAt), ); return record; }; /** The one row writer both the single ask and a questionnaire item use. */ const writeDecision = (draft: DecisionDraft): DecisionRecord => { const kind = draft.kind ?? "question"; const record: DecisionRecord = { id: crypto.randomUUID(), project: draft.project, kind, question: draft.question, askedAt: draft.at, // The seven-day deadline is for a question a human never answered. A // watch has no such deadline: it is the orchestrator's own bookkeeping, // and expiring it silently could drop a release the day its condition // finally fires, so it is never due. expiresAt: kind === "watch" ? DECISION_WATCH_NEVER_EXPIRES : draft.at + DECISION_TTL_MS, state: "open", }; if (draft.blocks !== undefined) record.blocks = draft.blocks; if (draft.condition !== undefined) record.condition = draft.condition; if (draft.groupId !== undefined) record.groupId = draft.groupId; if (draft.specIssue !== undefined) record.specIssue = draft.specIssue; insertDecision.run( record.id, record.project, record.kind, record.question, toSql(record.blocks), record.askedAt, record.expiresAt, toSql(record.condition), record.state, toSql(record.groupId), toSql(record.specIssue), ); return record; }; /** * One spec-out questionnaire, written in one transaction (#947). * * The atomicity is the whole guarantee: every item is durable BEFORE the * single delivery, so a crash halfway can never put questions in front of an * operator that no row is waiting on. The group id is minted here rather than * supplied, because a caller that can name the group can also collide with * another one, and an empty batch is refused — a questionnaire of nothing is a * caller bug, not an empty delivery. */ const createDecisionGroupTx = db.transaction( (drafts: readonly DecisionDraft[], specIssue: number): DecisionRecord[] => { if (drafts.length === 0) throw new Error("a questionnaire needs at least one item"); if (!Number.isInteger(specIssue) || specIssue <= 0) { throw new Error(`a questionnaire must name the issue it specs, got ${String(specIssue)}`); } const groupId = crypto.randomUUID(); return drafts.map((draft) => writeDecision({ ...draft, groupId, specIssue })); }, ); const createRunTx = db.transaction((r: Omit): RunRecord => { if (r.state !== "claimed") return insertRunRecord(r); const turnOverride = selectTurnOverride.get(r.project, r.issue)?.maxTurns; const record = insertRunRecord({ ...r, maxTurns: Math.max(r.maxTurns, turnOverride ?? r.maxTurns), }); if (turnOverride !== undefined) deleteTurnOverride.run(r.project, r.issue); return record; }); const setTurnOverrideTx = db.transaction( (project: string, issue: number, maxTurns: number, setAt: number): void => { upsertTurnOverride.run(project, issue, maxTurns, setAt); insertTurnOverrideLedger.run(crypto.randomUUID(), project, issue, maxTurns, setAt); }, ); return { createRun(r: Omit): RunRecord { return createRunTx(r); }, updateRun(id: string, patch: RunPatch): void { const assignments: string[] = []; const values: SqlValue[] = []; for (const [column, value] of Object.entries(patch)) { // An explicitly-`undefined` key is the same as an absent one: a patch // built by spreading a partial result must not widen into columns the // caller did not mean to write (#468). Clearing a column still works by // naming it with an explicit `null`, which `toSql` maps to SQL NULL. if (!UPDATABLE_COLUMNS[column] || value === undefined) continue; assignments.push(`${column} = ?`); values.push(toSql(value)); } if (assignments.length === 0) return; values.push(id); // An unknown id matches no row: zero changes, no error, by design — the // dispatcher patches runs a restart may already have reaped. db.query( `UPDATE runs SET ${assignments.join(", ")} WHERE id = ?`, ).run(...values); }, getRun(id: string): RunRecord | undefined { const row = selectRun.get(id); return row ? toRecord(row) : undefined; }, graphToolsObservationCounts(project: string): { recorded: number; present: number; absent: number } { // Counted in JS through the same defensive converter `toRecord` uses, so // a hand-edited row can never move a truth-value count (#726). let present = 0; let absent = 0; for (const row of selectGraphToolsObs.all(project)) { const observed = toGraphTools(row.graphTools); if (observed === undefined) continue; if (observed.present) present += 1; else absent += 1; } return { recorded: present + absent, present, absent }; }, activeRuns(project: string): RunRecord[] { return selectActive.all(project, ...ACTIVE_STATES).map(toRecord); }, liveRuns(project: string): RunRecord[] { return selectLive.all(project, ...LIVE_STATES).map(toRecord); }, leasedRuns(project: string): RunRecord[] { return selectLeased.all(project, ...LIVE_STATES, project).map(toRecord); }, retainedRuns(project: string): RunRecord[] { return selectRetained.all(project).map(toRecord); }, recentRuns(project: string, mergedSinceEpochMs: number): RunRecord[] { return selectRecentRuns.all(project, mergedSinceEpochMs).map(toRecord); }, runsForProjectPr(project: string, prUrl: string, mergedSinceEpochMs: number): RunRecord[] { return selectRunsForPr.all(project, prUrl, mergedSinceEpochMs).map(toRecord); }, enqueueReviewRevision(draft: ReviewRevisionDraft): ReviewRevisionEnqueue { // `immediate` = BEGIN IMMEDIATE: the write lock is acquired before the // pending-row read, so a second connection's transaction cannot observe // a snapshot older than the first connection's commit (#786). return enqueueReviewRevisionTx.immediate(draft); }, pendingReviewForRun(project: string, runId: string): ReviewRevisionRecord | undefined { const row = selectPendingReviewForRun.get(project, runId); return row === null ? undefined : toReviewRevision(row); }, pendingReviewRevisions(project: string): ReviewRevisionRecord[] { return selectPendingReviewRevisions.all(project).map(toReviewRevision); }, latestReviewRound(project: string, runId: string): number { return selectLatestReviewRound.get(project, runId)?.n ?? 0; }, markReviewRevisionDispatched(id: string, at: number): void { markReviewRevisionDispatchedRow.run(at, id); }, recordReviewRevisionLaunch( project: string, id: string, decision: { launchMode: ReviewLaunchMode; requestedModel?: string; originSessionRef?: string; at: number; }, ): boolean { // The decision and the write are one statement (#1045): `changes` is the // verdict, so a second dispatch pass cannot observe "undecided" and // launch its own. Re-affirming the identical decision reports true — // a dispatch retry after a killed launch must be idempotent, not fatal. const requestedModel = decision.requestedModel ?? null; const originSessionRef = decision.originSessionRef ?? null; return ( recordReviewRevisionLaunchRow.run( decision.launchMode, requestedModel, originSessionRef, decision.at, id, project, decision.launchMode, requestedModel, originSessionRef, ).changes > 0 ); }, recordReviewCorrectionSession(project: string, id: string, ref: string, resolvedModel?: string): boolean { // Guarded on the recorded mode, in the statement: a round decided // `resume-original` has no correction session to name, and answering // false here is how the caller learns it read the wrong decision. return ( recordReviewCorrectionSessionRow.run(ref, resolvedModel ?? null, id, project, FRESH_CORRECTION).changes > 0 ); }, reviewRevisionLaunch(project: string, id: string): ReviewCorrectionProvenance | undefined { const row = selectReviewRevisionForProject.get(id, project); // A missing round is `undefined`; a round nobody has decided is an empty // provenance. The distinction matters: the first is "no such round", the // second is "this round exists and nothing was written down". return row === null || row === undefined ? undefined : toCorrectionProvenance(row); }, settleReviewRevision(id: string, outcome: ReviewRevisionOutcome, at: number): void { settleReviewRevisionRow.run(outcome, at, id); }, unsettledReviewRevisions(project: string): ReviewRevisionRecord[] { return selectUnsettledReviewRevisions.all(project).map(toReviewRevision); }, requeueReviewRevision(id: string): void { requeueReviewRevisionRow.run(id); }, retryReviewRevisionAfterInfra(id: string, max: number): ReviewRevisionRetry { const requeued = retryReviewRevisionRow.run(id, max).changes > 0; const retries = selectReviewRevisionById.get(id)?.infraRetries ?? 0; return requeued ? { kind: "requeued", retries } : { kind: "exhausted", retries }; }, mergeBlockingReviews(project: string, prUrl: string, headSha: string): ReviewRevisionRecord[] { return selectMergeBlockingReviews.all(project, prUrl, headSha).map(toReviewRevision); }, recordReviewClearance(clearance: { project: string; prUrl: string; headSha: string; by: string; reason: string; at?: number; }): ReviewClearance { const record: ReviewClearance = { id: newReportId(), project: clearance.project, prUrl: clearance.prUrl, headSha: clearance.headSha, at: clearance.at ?? Date.now(), by: clearance.by, reason: clearance.reason, }; insertReviewClearance.run( record.id, record.project, record.prUrl, record.headSha, record.at, record.by, record.reason, ); return record; }, reviewClearances(project: string, prUrl: string, headSha: string): ReviewClearance[] { return selectReviewClearances.all(project, prUrl, headSha).map((row) => ({ ...row })); }, openReviewAdjudication(draft: { project: string; issue: number; prUrl: string; headSha: string; role: string; requestedAt?: number; findings?: string; }): ReviewAdjudicationAdmission { // `immediate` = BEGIN IMMEDIATE: the write lock is taken before the // one-shot read, so two processes cannot both see "no adjudication" and // both insert one (#874). return openAdjudicationTx.immediate( newReportId(), draft.project, draft.issue, draft.prUrl, draft.headSha, draft.role, draft.requestedAt ?? Date.now(), draft.findings, ); }, markReviewAdjudicationRunning( id: string, provenance: ReviewAdjudicationProvenance, at: number = Date.now(), ): boolean { return markAdjudicationRunningRow.run(JSON.stringify(provenance), at, id).changes > 0; }, settleReviewAdjudication( id: string, state: ReviewAdjudicationState, evidence?: string, at: number = Date.now(), ): boolean { // Terminal only, and once. A caller asking to settle into `pending` or // `running` is a bug, not a transition: refusing it here keeps "settled" // meaning "finished forever" for every reader. if (!TERMINAL_REVIEW_ADJUDICATION_STATES.some((terminal) => terminal === state)) return false; return settleAdjudicationRow.run(state, evidence ?? null, at, id).changes > 0; }, recordReviewAdjudicationDisposition(id: string, disposition: string): boolean { return recordAdjudicationDispositionRow.run(disposition, id).changes > 0; }, reviewAdjudication(id: string): ReviewAdjudicationRecord | undefined { const row = selectAdjudicationById.get(id); return row === null || row === undefined ? undefined : toReviewAdjudication(row); }, reviewAdjudicationForHead( project: string, prUrl: string, headSha: string, ): ReviewAdjudicationRecord | undefined { const row = selectAdjudicationForHead.get(project, prUrl, headSha); return row === null || row === undefined ? undefined : toReviewAdjudication(row); }, openReviewAdjudications(project: string): ReviewAdjudicationRecord[] { return selectOpenAdjudications.all(project).map(toReviewAdjudication); }, reviewRevisionsForPr(project: string, prUrl: string): ReviewRevisionRecord[] { return selectRevisionsForPr.all(project, prUrl).map(toReviewRevision); }, reviewAdjudicationsForPr(project: string, prUrl: string): ReviewAdjudicationRecord[] { return selectAdjudicationsForPr.all(project, prUrl).map(toReviewAdjudication); }, claimRunForReview(runId: string): boolean { return claimRunForReviewTx(runId); }, runsNeedingBaseCheck(project: string, limit = 20): RunRecord[] { return selectPendingBaseChecks.all(project, limit).map(toRecord); }, upsertBaseHealth(project: string, row: BaseHealth): void { upsertBaseHealthRow.run( project, row.repo, row.branch, row.headSha, row.verdict, row.runsCount, row.detail ?? null, row.checkedAt, ); }, baseHealth(project: string): BaseHealth[] { return selectBaseHealth.all(project).map(toBaseHealth); }, mergedRepoBranches( project: string, sinceEpochMs: number, ): { repo: string; baseRef?: string }[] { return selectMergedRepoBranches.all(project, sinceEpochMs).map((row) => row.baseRef === null ? { repo: row.repo } : { repo: row.repo, baseRef: row.baseRef }, ); }, baseFreeze(project: string, repo: string): BaseFreeze | undefined { const row = selectBaseFreeze.get(project, repo); return row === null ? undefined : toBaseFreeze(row); }, freezes(project: string): BaseFreeze[] { return selectFreezes.all(project).map(toBaseFreeze); }, setBaseFreeze( project: string, freeze: { repo: string; culpritSha: string; detail?: string; setAt?: number }, ): boolean { const setAt = freeze.setAt ?? Date.now(); const wasActive = (wasBaseFrozen.get(project, freeze.repo)?.n ?? 0) > 0; setBaseFreezeRow.run( project, freeze.repo, freeze.culpritSha, freeze.detail ?? null, setAt, ); return !wasActive; }, clearBaseFreeze( project: string, repo: string, by: string, reason: string, at = Date.now(), ): boolean { return clearBaseFreezeRow.run(at, by, reason, project, repo).changes > 0; }, activeReleaseComposition(project: string): ReleaseComposition | undefined { const row = selectActiveReleaseComposition.get(project); return row === null ? undefined : toReleaseComposition(row); }, declareReleaseComposition( project: string, composition: { campaign: string; allowedPrUrls: string[]; declaredBy?: string; declaredAt?: number; }, ): ReleaseComposition | undefined { // One active composition per project: a second declaration while one // is live is the caller's error to see, so hand the conflict straight // back. The transactional re-check keeps two declarers from racing. const conflict = declareReleaseCompositionTx( project, composition.campaign, JSON.stringify(composition.allowedPrUrls), composition.declaredBy ?? null, composition.declaredAt ?? Date.now(), ); return conflict === null ? undefined : toReleaseComposition(conflict); }, completeReleaseComposition(project: string, by: string, reason: string, at = Date.now()): boolean { return retireReleaseCompositionRow.run(at, by, reason, project).changes > 0; }, cancelReleaseComposition(project: string, by: string, reason: string, at = Date.now()): boolean { return retireReleaseCompositionRow.run(at, by, reason, project).changes > 0; }, grantReleaseCompositionOverride( project: string, override: { campaign: string; prUrl: string; by: string; reason: string; at?: number }, ): ReleaseCompositionOverride { const record: ReleaseCompositionOverride = { id: newReportId(), project, campaign: override.campaign, prUrl: override.prUrl, at: override.at ?? Date.now(), by: override.by, reason: override.reason, }; insertReleaseCompositionOverride.run( record.id, record.project, record.campaign, record.prUrl, record.at, record.by, record.reason, ); return record; }, releaseCompositionOverride( project: string, campaign: string, prUrl: string, ): ReleaseCompositionOverride | undefined { const row = selectReleaseCompositionOverride.get(project, campaign, prUrl); return row === null ? undefined : { ...row }; }, releaseCompositionOverrides(project: string, campaign: string): ReleaseCompositionOverride[] { return selectReleaseCompositionOverrides.all(project, campaign).map((row) => ({ ...row })); }, salvagedRuns(project: string): RunRecord[] { return selectSalvaged.all(project).map(toRecord); }, quarantinedRuns(project: string): RunRecord[] { return selectQuarantined.all(project).map(toRecord); }, attemptsFor(project: string, issue: number): number { return countAttempts.get(project, issue)?.n ?? 0; }, failuresFor(project: string, issue: number): number { return countFailures.get(project, issue, project, issue)?.n ?? 0; }, continuationsFor(project: string, issue: number): number { return countContinuations.get(project, issue, project, issue)?.n ?? 0; }, classCountFor(project: string, issue: number, cls: FailureClass): number { return countByClass.get(project, issue, cls)?.n ?? 0; }, latestRun(project: string, issue: number): RunRecord | undefined { const row = selectLatestRun.get(project, issue); return row ? toRecord(row) : undefined; }, runsForIssue(project: string, issue: number): RunRecord[] { return selectRunsForIssue.all(project, issue).map(toRecord); }, /** * Spends the one model escalation this issue chain is allowed (#807), and * everything that makes it true: the marker, the `swapFrom` → `queueLabel` * transition that hands the continuation back, and the settling run's * recovery stamp — one transaction, so a restart never finds a chain that * reads as escalated with no continuation owed. * * `true` means this call spent it; `false` means the chain had already * escalated and nothing was written, which is the second-cap decomposition * verdict. Throws when `runId` names no row: the marker rolls back with it * rather than recording an escalation nothing can spend. */ escalateModel(escalation: { project: string; issue: number; model: string; failureClass: FailureClass; runId: string; swapFrom: string; queueLabel: string; at?: number; }): boolean { return escalateModelTx( escalation.project, escalation.issue, escalation.model, escalation.failureClass, escalation.runId, escalation.swapFrom, escalation.queueLabel, escalation.at ?? Date.now(), ); }, /** The escalation this chain has already spent, or `undefined` for a chain * that never capped — which is what a fresh issue always reads. */ modelEscalation(project: string, issue: number): ModelEscalation | undefined { const row = selectModelEscalation.get(project, issue); return row === null ? undefined : { model: row.model, failureClass: row.failureClass as FailureClass, runId: row.runId, at: row.at, }; }, setTurnOverride(project: string, issue: number, maxTurns: number, setAt = Date.now()): void { setTurnOverrideTx(project, issue, maxTurns, setAt); }, turnOverride(project: string, issue: number): number | undefined { return selectTurnOverride.get(project, issue)?.maxTurns; }, clearTurnOverride(project: string, issue: number): void { deleteTurnOverride.run(project, issue); }, listTurnOverrides(project: string): TurnOverride[] { return selectTurnOverrides.all(project); }, turnOverrideLedger( project: string, opts: { issue?: number; limit?: number } = {}, ): TurnOverride[] { const issue = opts.issue ?? 0; return selectTurnOverrideLedger.all(project, issue, issue, opts.limit ?? 50); }, runsStartedSince(project: string, sinceEpochMs: number): number { return countStartedSince.get(project, sinceEpochMs)?.n ?? 0; }, spendSince(project: string, sinceEpochMs: number): number { return sumSpendSince.get(project, sinceEpochMs)?.total ?? 0; }, reservedSpendUsd(project: string): number { return sumLiveReservations.get(project, ...LIVE_STATES)?.total ?? 0; }, statsRuns(project: string, sinceEpochMs: number): RunRecord[] { return selectStatsRuns.all(project, sinceEpochMs, project, sinceEpochMs).map(toRecord); }, wasNotified(key: string): boolean { return (countNotified.get(key)?.n ?? 0) > 0; }, markNotified(key: string): void { // ponytail: notifications is append-only and never pruned. It gains one // short row per escalation, so it is decades from mattering; prune by // `at` if it ever does. insertNotified.run(key, Date.now()); }, orchestratorIncident(project: string): OrchestratorIncident | undefined { const row = selectOrchestratorIncident.get(project); return row === null ? undefined : toOrchestratorIncident(row); }, openOrchestratorIncident(draft: OrchestratorIncidentDraft): boolean { return ( insertOrchestratorIncident.run( draft.project, draft.mode, draft.cause ?? null, draft.since, ).changes === 1 ); }, bumpOrchestratorDiverted(project: string, by = 1): void { // Only meaningful while an incident is open: a healthy or external // orchestrator has no row, and the UPDATE touches nothing. bumpOrchestratorIncident.run(by, project); }, /** The workspace ids conductor recorded as its own worker surface for * this project, oldest first (#1035 review). Durable across daemon AND * Herdr restarts, which is the point: Herdr restores workspaces without * their metadata tokens, so discovery needs a second authority. */ workerWorkspaceIds(project: string): string[] { return selectHerdrWorkerWorkspaces.all(project).map((row) => row.workspace_id); }, rememberWorkerWorkspace(project: string, workspaceId: string): void { insertHerdrWorkerWorkspace.run(project, workspaceId, Date.now()); }, forgetWorkerWorkspace(project: string, workspaceId: string): void { deleteHerdrWorkerWorkspace.run(project, workspaceId); }, closeOrchestratorIncident(project: string, _at: number): OrchestratorIncident | undefined { const row = deleteOrchestratorIncident.get(project); return row === null ? undefined : toOrchestratorIncident(row); }, recordDaemonStop(draft: DaemonStopDraft): DaemonStop { const entry: DaemonStop = { ...draft, id: crypto.randomUUID(), at: Date.now() }; insertDaemonStop.run( entry.id, entry.at, entry.controlPath, toSql(entry.callerPid), toSql(entry.callerUid), toSql(entry.role), entry.scope, toSql(entry.project), toSql(entry.daemonPid), toSql(entry.runtimeDir), JSON.stringify(entry.affected), entry.reason, entry.unattributed ? 1 : 0, ); return entry; }, latestDaemonStop(): DaemonStop | undefined { const row = selectLatestDaemonStop.get(); return row === null ? undefined : toDaemonStop(row); }, recordDispatch(project: string, summary: DispatchSummary): void { const previous = selectDispatch.get(project); upsertDispatch.run(project, JSON.stringify(summary)); if ( previous !== null && toDispatchSummary(previous.summary)?.completedAt === summary.completedAt ) { return; } for (const hold of summary.holds) { if (!isFrictionHoldReason(hold.reason)) continue; recordFriction(project, { kind: `admission:${hold.reason}`, occurrences: hold.count, issues: hold.issues, ...(hold.issues.length === 0 ? {} : { sample: `issues ${hold.issues.map((issue) => `#${issue}`).join(", ")}` }), at: summary.completedAt, }); } }, latestDispatch(project: string): DispatchSummary | undefined { const row = selectDispatch.get(project); return row === null ? undefined : toDispatchSummary(row.summary); }, lastDigestDedupeKey(project: string): string | undefined { const row = selectLastDigestKey.get(project); return row?.key ?? undefined; }, addHeldNotice(notice: HeldNoticeDraft): void { insertHeldNoticeDraft(notice); }, undigestedNotices( project: string, limit = DIGEST_BACKLOG_LIMIT, availabilityOnly = false, categories?: readonly InterruptCategory[], urgent?: boolean, ): HeldNotice[] { if (categories?.length === 0) return []; const rows = availabilityOnly && categories !== undefined ? db .query( `SELECT id, category, summary, detail, createdAt, releaseOnAvailable, urgent FROM held_notices WHERE ${undigestedNoticeWhere} AND releaseOnAvailable = 1 AND category IN (${categories.map(() => "?").join(", ")}) ${urgent === undefined ? "" : "AND urgent = ?"} ORDER BY createdAt ASC, rowid ASC LIMIT ?`, ) .all(project, ...categories, ...(urgent === undefined ? [] : [urgent ? 1 : 0]), limit) : (availabilityOnly ? selectAvailabilityHeldNotices : selectUndigestedNotices).all( project, limit, ); return rows.map((row) => ({ id: row.id, category: row.category as HeldNotice["category"], summary: row.summary, detail: row.detail, createdAt: row.createdAt, ...(row.releaseOnAvailable === 1 ? { releaseOnAvailable: true as const } : {}), ...(row.urgent === 1 ? { urgent: true as const } : {}), })); }, recordInstallSurfaces(observation: InstallSurfaceObservation): void { upsertInstallSurfaces.run( observation.at, observation.cliVersion, observation.ompVersion ?? null, observation.herdrSource ?? null, observation.telegramInstalled ?? null, observation.telegramDaemon ?? null, observation.telegramPublished ?? null, ); }, installSurfaces(): InstallSurfaceObservation | undefined { const row = selectInstallSurfaces.get(); if (row === null) return undefined; return { at: row.at, cliVersion: row.cliVersion, ...(row.ompVersion === null ? {} : { ompVersion: row.ompVersion }), ...(row.herdrSource === null ? {} : { herdrSource: row.herdrSource }), ...(row.tgInstalled === null ? {} : { telegramInstalled: row.tgInstalled }), ...(row.tgDaemon === null ? {} : { telegramDaemon: row.tgDaemon }), ...(row.tgPublished === null ? {} : { telegramPublished: row.tgPublished }), }; }, recordMaterialEvent(event: MaterialEventDraft): MaterialEvent { const record: MaterialEvent = { id: newReportId(), ...event, }; insertMaterialEvent.run( record.id, record.project, record.category, record.summary, record.evidence, record.occurredAt, record.recordedAt, ); return record; }, getMaterialEvent(id: string): MaterialEvent | undefined { const row = selectMaterialEvent.get(id); return row === null ? undefined : toMaterialEvent(row); }, undigestedMaterialEvents(project: string, limit = DIGEST_BACKLOG_LIMIT): MaterialEvent[] { return selectUndigestedMaterialEvents.all(project, limit).map(toMaterialEvent); }, digestBacklog(project: string): DigestBacklog { const material = selectMaterialEventBacklog.get(project); const held = selectHeldNoticeBacklog.get(project); return { materialCount: material?.count ?? 0, ...(material?.oldestAt === null || material?.oldestAt === undefined ? {} : { materialOldestAt: material.oldestAt }), heldNoticeCount: held?.count ?? 0, availabilityHeldNoticeCount: held?.availabilityCount ?? 0, ...(held?.oldestAt === null || held?.oldestAt === undefined ? {} : { heldNoticeOldestAt: held.oldestAt }), ...(held?.digestOldestAt === null || held?.digestOldestAt === undefined ? {} : { digestHeldNoticeOldestAt: held.digestOldestAt }), ...(held?.availabilityOldestAt === null || held?.availabilityOldestAt === undefined ? {} : { availabilityHeldNoticeOldestAt: held.availabilityOldestAt }), }; }, recordFriction, recordGhRefusal, ghRefusalsSince(sinceMs: number): { count: number; latestAt?: number } { const row = selectGhRefusalsSince.get(sinceMs); if (row === null || row === undefined || row.n === 0 || row.latest === null) { return { count: 0 }; } return { count: row.n, latestAt: row.latest }; }, recordOrchestratorRestart(project: string, at: number, reason: string): void { // Same guard as `recordGhRefusal`: a bad clock must not corrupt the // store, and here it would corrupt the cap itself — a restart stamped in // 1970 falls outside every window and is a free restart forever. if (!Number.isSafeInteger(at) || at < 0) return; recordOrchestratorRestartTx(project, at, reason); }, orchestratorRestartsSince(project: string, since: number): { at: number; reason: string }[] { return selectOrchestratorRestartsSince.all(project, since); }, recordSessionSpend(row: SessionSpendRow): void { insertSessionSpend.run( row.project, row.role, row.issue ?? null, row.model ?? null, row.resolvedModel ?? null, row.turns, // `?? null` and NOT `?? 0`: an absent spend is an unmetered session, // and writing 0 here would report it as free. The two readings are // distinguishable at the row precisely because this stays NULL. row.spendUsd ?? null, row.at, ); }, sessionSpendSince(project: string, since: number): SessionSpendRow[] { return selectSessionSpendSince.all(project, since).map(toSessionSpend); }, bumpGhCalls(day: string, source: string): void { bumpGhCall.run(day, source); }, ghCallsToday(day: string): { source: string; calls: number }[] { return selectGhCalls.all(day); }, ghCallsBetween(sinceDay: string, untilDay: string): number { return sumGhCallsBetween.get(sinceDay, untilDay)?.total ?? 0; }, pendingFriction( project: string, sinceEpochMs: number, minimumObservations: number, surfacedBeforeEpochMs: number, ): FrictionSignal[] { const sinceDay = new Date(sinceEpochMs).toISOString().slice(0, 10); const surfaces = new Map(selectFrictionSurfaces.all(project).map((row) => [row.kind, row.at])); const combined = new Map(); for (const row of selectFrictionSince.all(project, sinceDay)) { if ( !isFrictionKind(row.kind) || row.latestAt < sinceEpochMs || row.observations < 1 || row.occurrences < 1 ) { continue; } const prior = combined.get(row.kind); combined.set(row.kind, { kind: row.kind, observations: (prior?.observations ?? 0) + row.observations, occurrences: (prior?.occurrences ?? 0) + row.occurrences, issues: boundedUnique( prior?.issues ?? [], parseNumberList(row.issues), FRICTION_ISSUE_SAMPLES, ), samples: boundedUnique( prior?.samples ?? [], parseTextList(row.samples), FRICTION_TEXT_SAMPLES, ), latestAt: Math.max(prior?.latestAt ?? 0, row.latestAt), }); } return [...combined.values()] .filter( (signal) => signal.observations >= minimumObservations && (surfaces.get(signal.kind) ?? 0) <= surfacedBeforeEpochMs, ) .sort( (a, b) => b.observations - a.observations || b.occurrences - a.occurrences || b.latestAt - a.latestAt, ); }, markFrictionSurfaced(project: string, kinds: readonly FrictionKind[], at: number): void { for (const kind of new Set(kinds)) upsertFrictionSurface.run(project, kind, at); }, enqueueReport(draft: ReportDraft): ReportEnqueue { return enqueueReportRecord(draft); }, enqueueDigestReport( draft: ReportDraft, materialEventIds: readonly string[], heldNoticeIds: readonly string[], ): ReportEnqueue { return enqueueDigestReportTx(draft, materialEventIds, heldNoticeIds); }, enqueueAvailabilityReport( draft: ReportDraft, heldNoticeIds: readonly string[], ): ReportEnqueue | undefined { return enqueueAvailabilityReportTx.immediate(draft, heldNoticeIds); }, getReport(id: string): ReportRecord | undefined { const row = selectReport.get(id); return row === null ? undefined : toReport(row); }, withdrawHandoff( id: string, project: string, actor: string, reason: string, at: number, ): HandoffWithdrawalResult { return withdrawHandoffTx.immediate(id, project, actor, reason, at); }, handoffWithdrawals(project: string, limit = 5): HandoffWithdrawal[] { return selectHandoffWithdrawals.all(project, limit).map(toHandoffWithdrawal); }, deferPendingReportToNotice(id: string, notice: HeldNoticeDraft): boolean { return deferPendingReportToNoticeTx(id, notice); }, releasePendingAvailabilityReport(id: string, project: string, possibleRepeat = false): boolean { return releasePendingAvailabilityReportTx(id, project, possibleRepeat); }, reserveAvailabilityDigest( project: string, cycleKey: string, at: number, expiresAt: number, ): boolean { return reserveAvailabilityDigestTx.immediate(project, cycleKey, at, expiresAt); }, dueReports(project: string, now: number, limit: number): ReportRecord[] { return selectDueReports.all(project, now, limit).map(toReport); }, claimReport(id: string, attemptId: string, at: number): ReportRecord | undefined { // The row is written `sending` *before* the caller's request leaves, so a // crash in the window leaves an explicitly ambiguous row rather than a // silently lost one. Losing the race is normal, not an error: it means // another pass already owns this attempt. if (claimReportRow.run(attemptId, at, id, at).changes === 0) return undefined; const row = selectReport.get(id); return row === null ? undefined : toReport(row); }, markReportDelivered( id: string, attemptId: string, messageIds: readonly number[], at: number, ): boolean { // `messageId` stays the first accepted id so rows written before // multi-part sends and the operator-facing rendering never disagree. return ( deliverReportRow.run( toSql(messageIds[0]), toSql(messageIds.length === 0 ? null : JSON.stringify(messageIds)), at, at, id, attemptId, ).changes > 0 ); }, markReportPartsSent( id: string, attemptId: string, sentParts: number, sentPartsHash: string, at: number, ): boolean { return markPartsSentRow.run(sentParts, sentPartsHash, at, id, attemptId).changes > 0; }, markReportUncertain(id: string, attemptId: string, error: string): boolean { return uncertainReportRow.run(error, id, attemptId).changes > 0; }, markReportPending( id: string, attemptId: string, nextAttemptAt: number, error: string, at: number, ): boolean { return requeueReportRow.run(nextAttemptAt, error, at, id, attemptId).changes > 0; }, markReportFailed(id: string, attemptId: string, error: string, at: number): boolean { return failReportRow.run(error, at, id, attemptId).changes > 0; }, recoverSendingReports(project: string, staleAt: number, at: number): ReportRecord[] { const stale = selectStaleSending.all(project, staleAt); if (stale.length === 0) return []; recoverSendingRows.run(at, at, project, staleAt); // Re-read rather than patching the pre-update rows: the caller logs and // renders these, and a row that says `sending` while the ledger says // `pending` is exactly the kind of quiet disagreement #123 is about. return stale .map((row) => selectReport.get(row.id)) .filter((row): row is ReportRow => row !== null) .map(toReport); }, /** * One escalation handoff per event while the first has not landed (#347, * #453). The notifications ledger cannot decide this — its rows are * permanent, so an identical retry would be refused forever once the first * occurrence was marked (#453). The question is instead "is the first * handoff still owed?": an escalation report still in the outbox (the * `openReports` pending/sending/failed set) for the same project, category * and folded text, or a held notice of the same category and text that has * not been digested yet. Once the report is delivered, or the notice * digested, the identical text is a new occurrence, not a double-send. * Material reports describe one ordinary outcome each time and never * dedupe. */ hasOpenEscalationHandoff(project: string, category: InterruptCategory, body: string): boolean { const flat = body.replace(/\s+/g, " ").trim(); return ( selectOpenReports .all(project) .some((row) => row.kind === category && row.body.replace(/\s+/g, " ").trim() === flat) || selectUndigestedNotices .all(project, DIGEST_BACKLOG_LIMIT) .some((row) => row.category === category && row.detail.replace(/\s+/g, " ").trim() === flat) ); }, openReports(project: string): ReportRecord[] { return selectOpenReports.all(project).map(toReport); }, runsNeedingClassification(project: string, limit = 20): RunRecord[] { return selectUnclassified.all(project, limit).map(toRecord); }, historicalInfraCandidates( project: string, limit = 20, after?: { startedAt: number; rowid: number }, ): HistoricalInfraCandidate[] { const cursorAt = after?.startedAt ?? null; const cursorRowid = after?.rowid ?? null; return selectHistoricalInfraCandidates .all(project, cursorAt, cursorAt, cursorAt, cursorRowid, limit) .map((row) => ({ ...toRecord(row), rowid: row.rowid })); }, historicalInfraCursor(project: string) { const row = selectHistoricalInfraCursor.get(project); return row == null ? undefined : { startedAt: row.cursorStartedAt, rowid: row.cursorRowid, classifierVersion: row.classifierVersion }; }, setHistoricalInfraCursor( project: string, startedAt: number, rowid: number, classifierVersion: string, ): void { upsertHistoricalInfraCursor.run(project, startedAt, rowid, classifierVersion, Date.now()); }, reclassifyInfra(id: string): boolean { return reclassifyToInfra.run(id).changes > 0; }, recentSpendSamples(project: string, limit: number): { turns: number; spendUsd: number }[] { return selectSpendSamples.all(project, limit).map((row) => ({ ...row })); }, recentLatencySamples(project: string, limit: number): RunRecord[] { return selectLatencySamples.all(project, limit).map(toRecord); }, failureClassCounts(project: string): { cls: FailureClass; n: number }[] { return selectFailureClassCounts .all(project) .map((row) => ({ cls: row.cls as FailureClass, n: row.n })); }, recordedOnlyClassCounts(project: string): { cls: FailureClass; n: number }[] { return selectRecordedOnlyClassCounts .all(project) .map((row) => ({ cls: row.cls as FailureClass, n: row.n })); }, recoveredSince(project: string, since: number): RunRecord[] { return selectRecoveredSince.all(project, since).map(toRecord); }, createDecision(draft: DecisionDraft): DecisionRecord { return writeDecision(draft); }, createDecisionGroup(drafts: readonly DecisionDraft[], specIssue: number): DecisionRecord[] { return createDecisionGroupTx(drafts, specIssue); }, decisionGroup(groupId: string): DecisionRecord[] { return selectDecisionGroup.all(groupId).map(toDecision); }, openDecisions(project: string): DecisionRecord[] { return selectOpenDecisions.all(project).map(toDecision); }, decision(id: string): DecisionRecord | undefined { const row = selectDecision.get(id); return row === null ? undefined : toDecision(row); }, resolveDecision(id: string, state: "answered" | "withdrawn", resolution: string, at: number): boolean { return resolveDecisionRow.run(state, resolution, at, id).changes > 0; }, markDecisionConditionMet(id: string, at: number, head?: string): boolean { // The window is computed from the observation instant, not from `now()`, // so a pass that observed a condition and wrote it a second later cannot // give the row a second longer to live. return markConditionMetRow.run(at, toSql(head), at + DECISION_TTL_MS, id).changes > 0; }, clearDecisionConditionMet(id: string): boolean { return clearConditionMetRow.run(DECISION_WATCH_NEVER_EXPIRES, id).changes > 0; }, expireDueDecisions(project: string, now: number): DecisionRecord[] { const due = selectDueDecisions.all(project, now); if (due.length === 0) return []; expireDecisionRows.run(now, project, now); // Re-read for the same reason `recoverSendingReports` does: the caller // renders these, and a row that still says `open` while the ledger says // `expired` is the disagreement this table exists to remove. return due .map((row) => selectDecision.get(row.id)) .filter((row): row is DecisionRow => row !== null) .map(toDecision); }, appendVerbLedger(draft: VerbLedgerDraft): VerbLedgerEntry { const entry: VerbLedgerEntry = { ...draft, id: crypto.randomUUID(), at: Date.now() }; insertVerbLedger.run( entry.id, entry.project, toSql(entry.runId), toSql(entry.issue), entry.verb, entry.role, JSON.stringify(entry.args), entry.decision, toSql(entry.refusal), entry.detail, toSql(entry.sha), entry.at, ); return entry; }, /** The surviving `unknown` rows, newest first, with whatever terminal * evidence the sweep recorded on them (#986). Bounded per call. */ unknownRuns(project: string, limit = 20): RunRecord[] { return selectUnknownEvidence.all(project, limit).map(toRecord); }, verbLedger( project: string, opts: { runId?: string; issue?: number; limit?: number } = {}, ): VerbLedgerEntry[] { // Sentinels rather than a composed SQL string: the filters are optional, // and building the WHERE clause by concatenation is how a project name // ends up interpolated into a query in a package that otherwise binds // every value (see `UPDATABLE_COLUMNS` for the one place that cannot). const runId = opts.runId ?? ""; const issue = opts.issue ?? 0; const limit = opts.limit ?? 50; return selectVerbLedger.all(project, runId, runId, issue, issue, limit).map(toVerbLedgerEntry); }, acquireMergeLock( project: string, holder: string, prUrl: string, now: number, staleAfterMs: number, ): MergeLock | undefined { if (!takeMergeLock(project, holder, prUrl, now, now - staleAfterMs)) return undefined; return { project, holder, prUrl, at: now }; }, releaseMergeLock(project: string, holder: string): void { dropMergeLock.run(project, holder); }, mergeLock(project: string): MergeLock | undefined { const row = selectMergeLock.get(project); return row === null ? undefined : { ...row }; }, enqueueLabelOps( project: string, ops: readonly { issue: number; op: "add" | "remove"; label: string }[], ): void { if (ops.length === 0) return; const now = Date.now(); enqueueLabelOpsTx( ops.map((op) => ({ ...op, project, createdAt: now })), ); }, pendingLabelOps(project: string, now: number): LabelOp[] { return selectPendingLabelOps.all(project, now).map(toLabelOp); }, pendingLabelOpsFor(project: string, issue: number): LabelOp[] { return selectPendingLabelOpsFor.all(project, issue).map(toLabelOp); }, settleLabelOp(id: number): void { deleteLabelOp.run(id); }, deferLabelOp(id: number, error: string, nextAttemptAt: number, countAttempt = true): void { if (countAttempt) bumpLabelOpAttempt.run(error, nextAttemptAt, id); else parkLabelOp.run(error, nextAttemptAt, id); }, countPendingLabelOps(project: string): number { return countLabelOps.get(project)?.n ?? 0; }, oldestPendingLabelOpAt(project: string): number | undefined { const row = selectOldestPendingLabelOp.get(project); return row === null ? undefined : row.createdAt; }, recordIntake(draft: IntakeDraft): IntakeItem { // A mined signal's id is a function of its provenance, an operator's own // idea gets a fresh random one. That single choice is the whole // idempotency mechanism: the machine keeps landing on its own row, the // human never does, and two identical thoughts on two days stay two // thoughts. const id = draft.source === undefined ? newReportId() : minedIntakeId(draft.project, draft.source); insertIntake.run(id, draft.project, draft.text, draft.at, draft.source ?? null); const stored = selectIntakeById.get(id); // The read cannot miss — the row is either the one just inserted or the // one that made the insert a no-op — but a fabricated pending item would // be worse than a loud failure, since the caller would report having // filed something that is not on file. if (stored === null) { throw new Error(`intake item ${id} vanished immediately after insert`); } return toIntakeItem(stored); }, pendingIntake(project: string): IntakeItem[] { return selectPendingIntake.all(project).map(toIntakeItem); }, resolveIntake(id: string, state: "groomed" | "dismissed", issueUrl?: string): boolean { const result = state === "groomed" ? groomIntake.run(issueUrl ?? null, Date.now(), id) : dismissIntake.run(id); return result.changes > 0; }, upsertGrooming(draft: GroomingDraft): void { upsertGroomingRow.run( draft.project, draft.issue, draft.verdict, draft.reason, draft.evidence, draft.at, ); }, grooming(project: string, issue: number): GroomingRecord | undefined { const row = selectGrooming.get(project, issue); return row === null ? undefined : toGrooming(row); }, groomingVerdicts(project: string, verdict?: GroomingVerdict): GroomingRecord[] { const rows = verdict === undefined ? selectGroomingAll.all(project) : selectGroomingByVerdict.all(project, verdict); return rows.map(toGrooming); }, reconcileGrooming( project: string, holds: readonly { issue: number; reason: AdmissionHoldReason; detail?: string }[], ): void { reconcileGroomingTx(project, holds); }, retireGroomingNotOpen(project: string, openIssues: readonly number[]): number[] { // The caller owns "is this snapshot complete"; this owns "which rows does // it condemn". Computed here rather than in SQL so the returned list is // exactly what was deleted, which is what the log line reports. const open = new Set(openIssues); const condemned = selectGroomingAll .all(project) .map(toGrooming) .filter((row) => !open.has(row.issue)) .map((row) => row.issue); if (condemned.length === 0) return []; retireGroomingTx(project, condemned); return condemned; }, markGroomingPromoted(project: string, issue: number, at: number, by: PromotedBy): boolean { // One guarded UPDATE, so "did I win" is SQLite's answer rather than a // read-then-write that two daemon passes could both pass (#1041). return markGroomingPromotedRow.run(at, by, project, issue).changes > 0; }, promotionsSince(project: string, since: number): GroomingRecord[] { return selectPromotionsSince.all(project, since).map(toGrooming); }, approveEpic(project: string, issue: number, at: number, by: string): void { // Refused rather than stored: an approval is a standing gate other code // reads, and a row keyed on #0 or #1.5 would be a permanent fact about an // issue that cannot exist. The table's CHECK would catch the non-positive // case as an opaque SQLite error; this names the argument instead. if (!Number.isInteger(issue) || issue <= 0) { throw new Error(`an epic approval needs a positive issue number, got ${String(issue)}`); } insertEpicApproval.run(project, issue, at, by); }, revokeEpicApproval(project: string, issue: number): boolean { if (!Number.isInteger(issue) || issue <= 0) { throw new Error(`an epic approval needs a positive issue number, got ${String(issue)}`); } return deleteEpicApproval.run(project, issue).changes > 0; }, epicApproval(project: string, issue: number): EpicApproval | undefined { const row = selectEpicApproval.get(project, issue); return row === null ? undefined : { ...row }; }, epicApprovals(project: string): EpicApproval[] { return selectEpicApprovals.all(project).map((row) => ({ ...row })); }, close(): void { db.close(false); }, }; }