From 988a4b674900d951284929789725cd4b04815a43 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:51:28 -0600 Subject: [PATCH] feat(recovery): ownable-kind registry and report-only ownership reconciler (#19861) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A control-plane restart leaves in-progress records owned by dead processes, and about 31 per-subsystem proofs decide when to release them > - The durable ownership plan (section C) asks for one registry of ownable record kinds and one reconciler > - Phase 1 must measure first: how many in-progress records exist, who owns them, and which invariants they break, with no behavior change > - This pull request registers 46 kinds from the Phase 0 inventory's section C and adds a report-only pass to the existing native finalization reconciler > - The benefit is per-kind invariant-violation counts from chaos waves and staging. That data unlocks Phase 2 (moving kinds onto one settle path) and the removal of the company-wide admission gate ## Linked Issues or Issue Description Refs #19831 (plan, section C and Phase 1). Refs #19846 (Phase 0 inventory, section C). Stacked on #19853 (the prover) and #19851 (one process identity). **Problem** No single place lists the control plane's in-progress records or checks the plan's invariants. Each subsystem releases its own records by its own rules, so nobody can count how many records break "owned", "settled" or "visible next action". **Expected behavior** One report-only pass lists every registered kind, classifies each owner as gone, alive or unprovable, and records each invariant violation once. It changes no record and stays quiet when nothing is in progress. **Impact** Phase 2 moves kinds onto the shared settle path in the order these counts show. Phase 3 removes the admission gate once the I3 data exists. ## What Changed - New `server/src/services/recovery/ownership-registry.ts`: 46 `OwnableKind` entries (`id`, `obligation`, `storage`, `listInProgress`, `settle: null`, `invariantChecks`). Each record names its owner as a process stamp, a fence token, a run, an operator, or none. The skill-cache lock is registered with a named coverage gap because its lock files live in per-skill roots. - New `server/src/services/recovery/ownership-invariant-report.ts`: one pass classifies every owner with `proveOwnerGone` (or the owning run's status), then runs: - I1 owned: an in-progress record with no owner; - I2 settled: an owner proven gone on two consecutive passes while the record is still in progress; - I3 admission independence: terminal runs whose accounting outlives the 10 min closure, which hold company admission; - I4 exactly-once intents: wake requests ended `skipped`/`failed` on a temporary refusal (heuristic until section E); - I7 visible next action: an agent-assigned `in_progress` issue with no live run, no pending wake and no open recovery action for 5 min. - I5, I6 and I8 are reported as named coverage gaps; I8 comes in the speed-budget PR. - Violations are written once per process to the activity log as `recovery.invariant_violation` (`actorId: ownership-reconciler`, `details.mode: report_only`), insert-only with no live-event or plugin fan-out. Nothing goes to Telemetry (AGENTS.md §5.7). - `native-finalization-reconciler.ts`: `registerOwnershipReconciliationWork` runs the pass in the shared delivery-work scheduler (new queue `ownership-invariant-report`). There is no second loop. - `index.ts` registers it after residual recovery and does not await its first pass, so readiness is not delayed. ### Sleep-aware trigger (rules/sleep-aware-background-work.mdc) - **Trigger:** one pass at registration (startup reconstruction from durable truth). After that, a settled `active-execution-recovery` signal, which run lifecycle, controller renewal and process-evidence writes already emit. Signals are coalesced to at most one pass per 30 s. No producer gained a write. - **Pending predicate and next deadline:** a 30 s follow-up only while the last pass saw a changing in-progress set, or an owner proven gone that needs the second I2 pass. - **Quiet:** an empty set or the same stuck set twice sets no deadline. The test asserts `nextWakeAt() === null` after an idle pass. - **Turn path:** the pass never runs before `turn.start`. Stamps read only cached identity. ## Verification - `npx vitest run src/__tests__/ownership-reconciliation-report.test.ts` (embedded Postgres, in `server/`): 4 tests pass: - every kind is registered once with no settle path; - verdicts per class: alive (this process's claim), gone (lease of a failed run), unprovable (no owner; another PID space); - each rule fires on its fixture, and I2 fires only on the second pass; - every touched table is byte-identical before and after (report-only); - activity rows are written once per violation, and none are written with `record: false`; - the worker is quiet when idle and schedules a follow-up only while records change. - `npx vitest run src/__tests__/delivery-work-coordinator.test.ts src/services/owner-proof.test.ts src/__tests__/server-startup-feedback-export.test.ts`: pass. - `npx tsc --noEmit -p server`: no errors. ## Risks - No record changes, so no settle risk. The risk is load: one pass runs about 46 small indexed or `limit 201` queries, at most once per 30 s while runs are active, and none when idle. - I4 matches refusal text with a heuristic. It can over- or under-count until section E gives wake requests closed terminal reasons. It is labelled as such in the code. - Violations dedupe per process, so a restart writes a still-open violation again (once per boot). Chaos reports count distinct `(rule, entityId)`. - Previous-release story (`rules/upgrade-safe-run-state.mdc`): no persisted run state changes. The pass reads rows of any age, and a missing field classifies as unprovable. - `heartbeat_runs` and `native_run_finalizations` record no host or PID space, so foreign-boot owners there classify as unprovable (`missing_identity`). This is the inventory's finding 2, and the report makes it countable. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, in Claude Code with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../ownership-reconciliation-report.test.ts | 375 ++++++++++ server/src/index.ts | 7 + .../services/delivery-work-notifications.ts | 1 + .../native-finalization-reconciler.ts | 131 +++- .../recovery/ownership-invariant-report.ts | 428 +++++++++++ .../services/recovery/ownership-registry.ts | 681 ++++++++++++++++++ 6 files changed, 1622 insertions(+), 1 deletion(-) create mode 100644 server/src/__tests__/ownership-reconciliation-report.test.ts create mode 100644 server/src/services/recovery/ownership-invariant-report.ts create mode 100644 server/src/services/recovery/ownership-registry.ts diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts new file mode 100644 index 00000000000..6f7ab9a1ed5 --- /dev/null +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -0,0 +1,375 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { + activityLog, + agentWakeupRequests, + agents, + budgetPolicies, + companies, + createDb, + environmentLeases, + heartbeatRuns, + issues, + nativeRunFinalizations, + sandboxCreationOperations, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { currentControllerBootId, currentProcessSpaceHash } from "../services/controller-process-identity.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerOwnershipReconciliationWork } from "../services/native-runtime/native-finalization-reconciler.js"; +import { + createOwnershipReportState, + OWNERSHIP_INVARIANT_VIOLATION_ACTION, + runOwnershipReconciliationPass, + type OwnershipReport, +} from "../services/recovery/ownership-invariant-report.js"; +import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); + +const minutesAgo = (minutes: number) => new Date(Date.now() - minutes * 60_000); +const kindReport = (report: OwnershipReport, kind: string) => report.kinds.find(entry => entry.kind === kind)!; +const rules = (report: OwnershipReport, kind: string) => + report.violations.filter(violation => violation.kind === kind).map(violation => violation.rule).sort(); + +describe("report-only ownership reconciliation", () => { + let database: Awaited> | undefined; + let db: ReturnType; + let companyId: string; + let agentId: string; + let ids: Record; + + let home: string; + let orphanSpool: string; + + beforeAll(async () => { + // Spool kinds read the instance root; never let a test read a real instance. + home = await mkdtemp(join(tmpdir(), "ownership-report-home-")); + orphanSpool = join(home, "orphan-spool"); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR", orphanSpool); + if (!external) database = await startEmbeddedPostgresTestDatabase("ownership-reconciliation-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 120_000); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); + vi.unstubAllEnvs(); + await rm(home, { recursive: true, force: true }); + }); + + beforeEach(async () => { + await rm(orphanSpool, { recursive: true, force: true }); + await db.execute(sql`truncate budget_policies, activity_log, agent_wakeup_requests, sandbox_creation_operations, environment_leases, + native_run_finalizations, heartbeat_runs, issues, agents, companies cascade`); + companyId = randomUUID(); agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Ownership fixture", issuePrefix: "OWN" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", role: "engineer", adapterType: "codex_local" }); + }); + + /** One record per owner class: alive here, provably gone, unprovable, unowned. */ + async function seed() { + ids = { + liveRun: randomUUID(), deadRun: randomUUID(), debtRun: randomUUID(), boundedRun: randomUUID(), + liveIssue: randomUUID(), stalledIssue: randomUUID(), + orphanLease: randomUUID(), unownedCleanup: randomUUID(), foreignIntent: randomUUID(), wake: randomUUID(), + }; + await db.insert(issues).values([ + { id: ids.liveIssue, companyId, title: "Live", status: "in_progress", assigneeAgentId: agentId }, + { id: ids.stalledIssue, companyId, title: "Stalled", status: "in_progress", assigneeAgentId: agentId, + updatedAt: minutesAgo(30) }, + ]); + await db.insert(heartbeatRuns).values([ + { id: ids.liveRun, companyId, agentId, issueId: ids.liveIssue, status: "running", invocationSource: "manual", + runtimeMode: "native", nativeIssueId: ids.liveIssue }, + { id: ids.deadRun, companyId, agentId, status: "failed", invocationSource: "manual", finishedAt: minutesAgo(1) }, + // Holds admission: a succeeded run with no observed cost. + { id: ids.debtRun, companyId, agentId, status: "succeeded", invocationSource: "manual", finishedAt: minutesAgo(20), + costAccountingPending: true }, + // Does not hold admission: its observed cost already counts as exposure. + { id: ids.boundedRun, companyId, agentId, status: "succeeded", invocationSource: "manual", finishedAt: minutesAgo(20), + costAccountingPending: true, usageJson: { budgetExposure: { costCents: 12 } } }, + ]); + await db.insert(budgetPolicies).values({ companyId, scopeType: "company", scopeId: companyId, + windowKind: "calendar_month_utc", amount: 1000 }); + // A fresh native claim held by this process. + await db.insert(nativeRunFinalizations).values({ runId: ids.liveRun, companyId, issueId: ids.liveIssue, + phase: "observed", leaseOwner: `${currentControllerBootId()}:1:token`, leaseExpiresAt: new Date(Date.now() + 60_000), + controllerBootId: currentControllerBootId(), controllerPid: process.pid, controllerGeneration: 1 }); + await db.insert(environmentLeases).values([ + // An active lease whose run already failed: the owner is provably gone. + { id: ids.orphanLease, companyId, heartbeatRunId: ids.deadRun, status: "active" }, + // A cleanup obligation that names no owner. + { id: ids.unownedCleanup, companyId, status: "pending_cleanup", metadata: { pendingCleanupRetryAttempts: 1 } }, + ]); + // A creation intent stamped by a process in another PID space. + await db.insert(sandboxCreationOperations).values({ id: ids.foreignIntent, companyId, + originalEnvironmentId: randomUUID(), runId: ids.deadRun, requestKey: "fixture", provider: "fixture", + driverKey: "fixture", state: "pending", config: {}, nextAttemptAt: new Date(), + metadata: { creationController: { bootId: "other-boot", pid: 1, hostHash: "another-pid-space", + processStartedAt: "2026-11-01T00:00:00.000Z" } } }); + // A wake request ended terminally on a temporary refusal. + await db.insert(agentWakeupRequests).values({ id: ids.wake, companyId, agentId, source: "assignment", + status: "skipped", error: "Company budget hold: admission paused" }); + } + + async function snapshot() { + return { + leases: await db.select({ id: environmentLeases.id, status: environmentLeases.status, metadata: environmentLeases.metadata }) + .from(environmentLeases).orderBy(environmentLeases.id), + claims: await db.select().from(nativeRunFinalizations), + intents: await db.select({ id: sandboxCreationOperations.id, state: sandboxCreationOperations.state }).from(sandboxCreationOperations), + issues: await db.select({ id: issues.id, status: issues.status, executionRunId: issues.executionRunId }).from(issues).orderBy(issues.id), + runs: await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, pending: heartbeatRuns.costAccountingPending }) + .from(heartbeatRuns).orderBy(heartbeatRuns.id), + }; + } + + it("registers every section-C kind once, with no settle path", () => { + const names = OWNABLE_KINDS.map(kind => kind.id); + expect(new Set(names).size).toBe(names.length); + expect(names.length).toBeGreaterThanOrEqual(40); + expect(OWNABLE_KINDS.every(kind => kind.settle === null)).toBe(true); + expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); + }); + + it("classifies each owner and reports invariant violations without changing any record", async () => { + await seed(); + const before = await snapshot(); + const state = createOwnershipReportState(); + + const first = await runOwnershipReconciliationPass(db, { state }); + + expect(kindReport(first, "native_executor_claim")).toMatchObject({ inProgress: 1, + verdicts: { alive: 1, gone: 0, unprovable: 0 }, reasons: { current_process: 1 } }); + expect(kindReport(first, "environment_run_lease_active")).toMatchObject({ inProgress: 1, + verdicts: { gone: 1 }, reasons: { owner_run_terminal: 1 } }); + expect(kindReport(first, "environment_pending_cleanup")).toMatchObject({ inProgress: 1, + verdicts: { unprovable: 1 }, reasons: { no_owner_stamp: 1 } }); + // A different PID space is never proof. + expect(currentProcessSpaceHash()).not.toBe("another-pid-space"); + expect(kindReport(first, "sandbox_creation_intent")).toMatchObject({ inProgress: 1, + verdicts: { unprovable: 1 }, reasons: { different_process_space: 1 } }); + expect(kindReport(first, "run_accounting_debt")).toMatchObject({ inProgress: 2 }); + + expect(rules(first, "environment_pending_cleanup")).toEqual(["I1_owned"]); + // I2 needs the owner proven gone on two consecutive passes. + expect(rules(first, "environment_run_lease_active")).toEqual([]); + expect(first.violations.filter(violation => violation.rule === "I3_admission_independent") + .map(violation => violation.recordId)).toEqual([ids.debtRun]); + expect(rules(first, "wake_request")).toEqual(["I4_exactly_once_intent"]); + expect(first.violations.filter(violation => violation.rule === "I7_visible_next_action") + .map(violation => violation.recordId)).toEqual([ids.stalledIssue]); + expect(Object.keys(first.coverageGaps)).toEqual(["I5_work_based_budgets", "I6_generation_fencing", "I8_fast_turns", + "I7_scope", "record_cap", "wake_coverage"]); + + expect(first.awaitingSecondProof).toBe(1); + const second = await runOwnershipReconciliationPass(db, { state }); + expect(rules(second, "environment_run_lease_active")).toEqual(["I2_settled"]); + expect(second.newViolations.map(violation => `${violation.rule}:${violation.kind}`)).toEqual(["I2_settled:environment_run_lease_active"]); + + // A stuck record already reported for I2 needs no further pass. + expect(second.awaitingSecondProof).toBe(0); + expect((await runOwnershipReconciliationPass(db, { state })).awaitingSecondProof).toBe(0); + + // Report-only: nothing was settled, released or retried. + expect(await snapshot()).toEqual(before); + + const rows = await db.select({ entityId: activityLog.entityId, actorId: activityLog.actorId, details: activityLog.details }) + .from(activityLog).where(eq(activityLog.action, OWNERSHIP_INVARIANT_VIOLATION_ACTION)); + // Each violation is written once per process, on the pass that first sees it. + const written = [...first.newViolations, ...second.newViolations]; + expect(rows).toHaveLength(written.length); + expect(rows.every(row => row.actorId === "ownership-reconciler")).toBe(true); + expect(rows.map(row => `${(row.details as Record).rule}:${row.entityId}`).sort()) + .toEqual(written.map(violation => `${violation.rule}:${violation.recordId}`).sort()); + expect(rows.find(row => row.entityId === ids.orphanLease)?.details).toMatchObject({ + rule: "I2_settled", kind: "environment_run_lease_active", reason: "owner_run_terminal", runId: ids.deadRun, mode: "report_only", + }); + }); + + it("proves a foreign-boot coordinator gone when its PID-space stamp matches and its PID is dead", async () => { + const issueId = randomUUID(); const runId = randomUUID(); const legacyRunId = randomUUID(); const legacyIssueId = randomUUID(); + await db.insert(issues).values([{ id: issueId, companyId, title: "Dead owner" }, { id: legacyIssueId, companyId, title: "Legacy" }]); + await db.insert(heartbeatRuns).values([ + { id: runId, companyId, agentId, issueId, status: "running", invocationSource: "manual", runtimeMode: "native", nativeIssueId: issueId }, + { id: legacyRunId, companyId, agentId, issueId: legacyIssueId, status: "running", invocationSource: "manual", + runtimeMode: "native", nativeIssueId: legacyIssueId }, + ]); + const claim = { companyId, phase: "observed", leaseOwner: "previous-boot:1:token", + leaseExpiresAt: new Date(Date.now() + 60_000), controllerBootId: "previous-boot", controllerPid: 4_194_303, + controllerProcessStartedAt: new Date("2026-11-07T09:00:00.000Z"), controllerGeneration: 1 }; + await db.insert(nativeRunFinalizations).values([ + { ...claim, runId, issueId, controllerHostHash: currentProcessSpaceHash() }, + // Written before the PID space was stamped: never provable. + { ...claim, runId: legacyRunId, issueId: legacyIssueId, controllerHostHash: null }, + ]); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(kindReport(report, "native_executor_claim")).toMatchObject({ inProgress: 2, + verdicts: { gone: 1, unprovable: 1 }, reasons: { pid_dead: 1, missing_identity: 1 } }); + }); + + it("writes nothing when recording is off", async () => { + await seed(); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.length).toBeGreaterThan(0); + expect(await db.select().from(activityLog)).toEqual([]); + }); + + it("does not report admission holds without an active budget policy", async () => { + await seed(); + await db.update(budgetPolicies).set({ isActive: false }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.rule === "I3_admission_independent")).toEqual([]); + }); + + it("reads every temporary refusal in a busy window, past unrelated failures", async () => { + const state = createOwnershipReportState(); + await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + await db.insert(agentWakeupRequests).values(Array.from({ length: 5 }, () => ({ companyId, agentId, + source: "assignment", status: "failed", error: "adapter exited with code 1" }))); + const refused = Array.from({ length: 5 }, () => randomUUID()); + await db.insert(agentWakeupRequests).values(refused.map(id => ({ id, companyId, agentId, source: "assignment", + status: "skipped", error: "issue lock busy" }))); + const report = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + expect(report.violations.filter(violation => violation.rule === "I4_exactly_once_intent") + .map(violation => violation.recordId).sort()).toEqual([...refused].sort()); + const again = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + expect(again.violations.filter(violation => violation.rule === "I4_exactly_once_intent")).toEqual([]); + }); + + it("attributes spool files to their company", async () => { + await mkdir(orphanSpool, { recursive: true }); + const runId = randomUUID(); + await writeFile(join(orphanSpool, "orphan-1.json"), JSON.stringify({ companyId, heartbeatRunId: runId })); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(report.violations.filter(violation => violation.kind === "sandbox_orphan_cleanup_spool")).toEqual([ + expect.objectContaining({ rule: "I1_owned", recordId: "orphan-1", companyId, runId }), + ]); + expect(await db.select({ entityId: activityLog.entityId }).from(activityLog) + .where(eq(activityLog.action, OWNERSHIP_INVARIANT_VIOLATION_ACTION))).toEqual([{ entityId: "orphan-1" }]); + }); + + it("counts admission holds only for policies whose scope includes the run", async () => { + await seed(); + await db.update(budgetPolicies).set({ scopeType: "agent", scopeId: randomUUID() }); + const other = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(other.violations.filter(violation => violation.rule === "I3_admission_independent")).toEqual([]); + await db.update(budgetPolicies).set({ scopeId: agentId }); + const own = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(own.violations.filter(violation => violation.rule === "I3_admission_independent").map(violation => violation.recordId)) + .toEqual([ids.debtRun]); + }); + + it("treats a saved monitor deadline as a next action", async () => { + await db.insert(issues).values({ companyId, title: "Monitored", status: "in_progress", assigneeAgentId: agentId, + updatedAt: minutesAgo(30), monitorNextCheckAt: new Date(Date.now() + 60 * 60_000) }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.rule === "I7_visible_next_action")).toEqual([]); + }); + + it("returns the end of the next-action grace window for an issue still inside it", async () => { + const updatedAt = minutesAgo(2); + await db.insert(issues).values({ companyId, title: "Fresh", status: "in_progress", assigneeAgentId: agentId, updatedAt }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.rule === "I7_visible_next_action")).toEqual([]); + expect(report.nextDueAt?.getTime()).toBe(updatedAt.getTime() + 5 * 60_000); + }); + + it("retries a violation whose activity row failed to save", async () => { + await mkdir(orphanSpool, { recursive: true }); + const laterCompany = randomUUID(); + await writeFile(join(orphanSpool, "orphan-2.json"), JSON.stringify({ companyId: laterCompany })); + const state = createOwnershipReportState(); + // The company does not exist yet, so the activity insert fails its foreign key. + const first = await runOwnershipReconciliationPass(db, { state }); + expect(first.retryNeeded).toBe(true); + expect(await db.select().from(activityLog)).toEqual([]); + + await db.insert(companies).values({ id: laterCompany, name: "Later", issuePrefix: "LAT" }); + await rm(orphanSpool, { recursive: true, force: true }); + const second = await runOwnershipReconciliationPass(db, { state }); + expect(second.retryNeeded).toBe(false); + expect(await db.select({ entityId: activityLog.entityId, companyId: activityLog.companyId }).from(activityLog)) + .toEqual([{ entityId: "orphan-2", companyId: laterCompany }]); + }); + + it("drops a violation whose activity row keeps failing after five attempts", async () => { + await mkdir(orphanSpool, { recursive: true }); + await writeFile(join(orphanSpool, "orphan-3.json"), JSON.stringify({ companyId: randomUUID() })); + const state = createOwnershipReportState(); + const passes: boolean[] = []; + for (let pass = 0; pass < 6; pass++) passes.push((await runOwnershipReconciliationPass(db, { state })).retryNeeded); + expect(passes).toEqual([true, true, true, true, false, false]); + }); + + it("counts late-receipt debt as an admission hold, as the gate does", async () => { + await seed(); + await db.update(heartbeatRuns).set({ costAccountingPending: false, usageJson: { accountingLateReceiptPending: true } }) + .where(eq(heartbeatRuns.id, ids.debtRun)); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.rule === "I3_admission_independent").map(violation => violation.recordId)) + .toEqual([ids.debtRun]); + }); + + it("reads a kind larger than one page in a single pass, so second proofs compare the same records", async () => { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "failed", invocationSource: "manual" }); + const leases = [randomUUID(), randomUUID(), randomUUID()].sort(); + await db.insert(environmentLeases).values(leases.map(id => ({ id, companyId, heartbeatRunId: runId, status: "active" }))); + const state = createOwnershipReportState(); + const first = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + expect(kindReport(first, "environment_run_lease_active")).toMatchObject({ inProgress: 3, truncated: false, verdicts: { gone: 3 } }); + expect(first.awaitingSecondProof).toBe(3); + const second = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + expect(second.violations.filter(violation => violation.rule === "I2_settled").map(violation => violation.recordId).sort()) + .toEqual(leases); + expect(second.awaitingSecondProof).toBe(0); + expect(second.retryNeeded).toBe(false); + }); + + it("keeps a wake that lands while dispatch is drained", async () => { + let allowed = false; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: () => {} }); + const reports: OwnershipReport[] = []; + const worker = registerOwnershipReconciliationWork(db, coordinator, () => allowed, { onReport: report => reports.push(report) }); + try { + await worker.ready; + expect(reports).toHaveLength(0); + // The skipped startup pass is still pending, so a retry stays due. + expect(coordinator.nextWakeAt()).not.toBeNull(); + allowed = true; + worker.wake(); + await expect.poll(() => reports.length, { timeout: 60_000, interval: 100 }).toBe(1); + } finally { + await coordinator.stop(); + } + }, 90_000); + + it("goes quiet with nothing in progress and follows up while records change", async () => { + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const reports: OwnershipReport[] = []; + const worker = registerOwnershipReconciliationWork(db, coordinator, () => true, { onReport: report => reports.push(report) }); + try { + await worker.ready; + expect(reports).toHaveLength(1); + expect(reports[0]!.inProgress).toBe(0); + // An idle instance: no deadline, so no SQL until a producer signals. + expect(coordinator.nextWakeAt()).toBeNull(); + + await seed(); + worker.wake(); + await expect.poll(() => reports.length, { timeout: 60_000, interval: 100 }).toBe(2); + expect(reports[1]!.inProgress).toBeGreaterThan(0); + expect(coordinator.nextWakeAt()).not.toBeNull(); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); + } + }, 90_000); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 74818db2dd3..9f4c0d05e85 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,4 +1,5 @@ import type { IdleDeadlineOwners } from "./services/idle-deadline-types.js"; +import { registerOwnershipReconciliationWork } from "./services/native-runtime/native-finalization-reconciler.js"; import { bindIdleDeadlineOwner } from "./services/idle-deadline-owner.js"; import { readIdleRetrySchedule } from "./services/heartbeat/idle-retry-schedule.js"; import { readIdleQuotaSchedule } from "./services/heartbeat/idle-quota-schedule.js"; @@ -1731,6 +1732,12 @@ async function startServerWithDatabaseTeardown( if (config.heartbeatSchedulerEnabled) await heartbeat.registerResidualRecoveryWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; + // Report-only ownership reconciliation (durable ownership plan, Phase 1). + // Its first pass runs in the background so it never delays readiness. + if (config.heartbeatSchedulerEnabled) void registerOwnershipReconciliationWork(db as any, deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + { enabled: async () => !(await heartbeat.resolveSchedulingSuppression()).suppressed }).ready + .catch(err => logger.warn({ err }, "ownership reconciliation report pass failed")); if (config.heartbeatSchedulerEnabled) await heartbeat.registerActiveRunSilenceWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) { diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 17a600d3289..487532c1684 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -12,6 +12,7 @@ export const DELIVERY_QUEUES = { nativeProcessSessions: "native-process-sessions", nativeQuestionCancellations: "native-question-cancellations", activeExecutionRecovery: "active-execution-recovery", + ownershipInvariantReport: "ownership-invariant-report", settledNativeProcesses: "settled-native-processes", orphanActiveLeases: "orphan-active-leases", legacyShutdownSettlement: "legacy-shutdown-settlement", diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 5d97112d267..c1b60e2334f 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -13,7 +13,14 @@ import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { recoverLegacyUnsafeWorkspaceExports } from "./native-workspace-export-recovery.js"; import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js"; import { logger } from "../../middleware/logger.js"; -import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork, subscribeDeliveryWork } from "../delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { + createOwnershipReportState, + ownershipReportFingerprint, + runOwnershipReconciliationPass, + type OwnershipReport, +} from "../recovery/ownership-invariant-report.js"; import { createHash, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, not, notInArray, or, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -768,3 +775,125 @@ async function reconcileNativeFinalizationsPass( } return results; } + +/** Existing settled-work topics whose producers write registered kinds: + * run lifecycle and controller renewal, environment leases, settled native + * processes, legacy shutdown fences, native process sessions, deferred + * receipts, stale issue locks, queued dispatch and accounting. */ +const OWNERSHIP_REPORT_WAKE_QUEUES = [ + DELIVERY_QUEUES.activeExecutionRecovery, + DELIVERY_QUEUES.orphanActiveLeases, + DELIVERY_QUEUES.settledNativeProcesses, + DELIVERY_QUEUES.legacyShutdownSettlement, + DELIVERY_QUEUES.nativeProcessSessions, + DELIVERY_QUEUES.deferredExecutionReceipts, + DELIVERY_QUEUES.staleIssueLocks, + DELIVERY_QUEUES.queuedRunDispatch, + DELIVERY_QUEUES.accounting, +] as const; + +/** Interval between report passes while in-progress records keep changing. */ +export const OWNERSHIP_REPORT_INTERVAL_MS = 30_000; + +/** + * The ownership reconciliation pass of this reconciler (plan section C, + * Phase 1). It is report-only: it classifies the owner of every registered + * in-progress record and records invariant violations, and never settles. + * + * Trigger and quiet condition (rules/sleep-aware-background-work.mdc): + * - one pass when the worker registers at startup, from durable truth; + * - a pass after any settled `active-execution-recovery` signal, which run + * lifecycle, controller renewal and process-evidence writes already emit, + * coalesced to at most one pass per 30 s; + * - a 30 s follow-up only while the last pass saw in-progress records whose + * set changed, or owners proven gone for the first time (rule I2 needs a + * second pass); + * - the end of rule I7's grace window for an issue that is still inside it. + * A pass that finds nothing in progress, or the same stuck set twice, sets no + * deadline: an idle instance runs no SQL here. The pass runs in the shared + * scheduler only, never on the path before `turn.start`. + */ +export function registerOwnershipReconciliationWork( + db: Db, + coordinator: ReturnType, + canRun: () => boolean, + options: { + onReport?: (report: OwnershipReport) => void; + /** False while heartbeat scheduling is suppressed for this instance. */ + enabled?: () => Promise; + } = {}, +) { + const state = createOwnershipReportState(); + let nextRunAt: number | null = null; + let lastFingerprint: string | null = null; + let lastReport: OwnershipReport | null = null; + // A wake is consumed only by a pass that actually ran, so a wake that lands + // during idle drain survives until dispatch reopens. + let pendingWake = true; + // While scheduling is suppressed the worker parks with no deadline and runs + // no SQL; the next producer signal re-checks suppression. + let parked = false; + let lastPassAt: number | null = null; + const unsubscribers: Array<() => void> = []; + const worker = coordinator.register(DELIVERY_QUEUES.ownershipInvariantReport, { + retryMs: OWNERSHIP_REPORT_INTERVAL_MS, + // A wake is handled at once, so a suppressed instance parks immediately; + // full passes are throttled below without SQL. + coalesceMs: 0, + pausePassiveMaintenance: true, + run: async () => { + if (!canRun()) return; + if (options.enabled && !(await options.enabled())) { + parked = true; + nextRunAt = null; + return; + } + parked = false; + // At most one full pass per interval; a wake inside it waits for the + // interval's end (in memory, no SQL). + if (lastPassAt !== null && Date.now() - lastPassAt < OWNERSHIP_REPORT_INTERVAL_MS) { + nextRunAt = lastPassAt + OWNERSHIP_REPORT_INTERVAL_MS; + return; + } + lastPassAt = Date.now(); + pendingWake = false; + const report = await runOwnershipReconciliationPass(db, { state }); + const fingerprint = ownershipReportFingerprint(report); + const changing = report.inProgress > 0 && fingerprint !== lastFingerprint; + const followUp = changing || report.awaitingSecondProof > 0 || report.retryNeeded + ? Date.now() + OWNERSHIP_REPORT_INTERVAL_MS : null; + // A record inside a grace window (rule I7) is a real future obligation. + const graceDue = report.nextDueAt?.getTime() ?? null; + nextRunAt = followUp === null ? graceDue : graceDue === null ? followUp : Math.min(followUp, graceDue); + if (fingerprint !== lastFingerprint) { + // One line per change of the in-progress set: the verdict counts per + // kind are the Phase 1 data (server log only, never Telemetry). + logger.info({ + inProgress: report.inProgress, + violations: report.violations.length, + kinds: Object.fromEntries(report.kinds.filter(kind => kind.inProgress > 0 || kind.error) + .map(kind => [kind.kind, { ...kind.verdicts, ...(kind.truncated ? { truncated: true } : {}), + ...(kind.error ? { error: kind.error } : {}) }])), + }, "ownership reconciliation pass"); + } + lastFingerprint = fingerprint; + lastReport = report; + options.onReport?.(report); + }, + hasPending: async () => pendingWake && !parked, + nextRunAt: () => nextRunAt, + onStop: () => { for (const unsubscribe of unsubscribers.splice(0)) unsubscribe(); }, + }); + // Producers of the registered kinds that already publish settled work + // signals. Kinds whose writers publish none are read on any of these + // wakes and at startup (named in the report's coverage gaps). + for (const queue of OWNERSHIP_REPORT_WAKE_QUEUES) { + unsubscribers.push(subscribeDeliveryWork(db, queue, () => { pendingWake = true; parked = false; worker.wake(); })); + } + return { + ready: worker.ready, + wake: worker.wake, + lastReport: () => lastReport, + stop: worker.stop, + }; +} diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts new file mode 100644 index 00000000000..9c037b07444 --- /dev/null +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -0,0 +1,428 @@ +import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql } from "drizzle-orm"; +import { + agentWakeupRequests, + budgetPolicies, + heartbeatRuns, + nativeRunFinalizations, + issueRecoveryActions, + issues, + type Db, +} from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import { persistActivity } from "../activity-log.js"; +import { proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js"; +import { + OWNABLE_KINDS, + resolveFenceHolders, + type InProgressRecord, + type OwnableKind, + type RecordOwner, +} from "./ownership-registry.js"; + +/** + * Report-only ownership reconciliation (plan section C, Phase 1). + * + * One pass lists every registered kind's in-progress records, classifies each + * owner as gone, alive or unprovable, and runs the plan's invariant checks. + * Violations are written once per process to the activity log as + * `recovery.invariant_violation`. Nothing is settled, released or retried, and + * nothing is sent to Telemetry (AGENTS.md §5.7). + */ + +export const OWNERSHIP_INVARIANT_VIOLATION_ACTION = "recovery.invariant_violation"; +export const OWNERSHIP_RECONCILER_ACTOR_ID = "ownership-reconciler"; + +export type InvariantRule = + | "I1_owned" + | "I2_settled" + | "I3_admission_independent" + | "I4_exactly_once_intent" + | "I7_visible_next_action"; + +export type InvariantViolation = { + rule: InvariantRule; + kind: string; + recordId: string; + companyId: string | null; + runId: string | null; + issueId: string | null; + reason: string; +}; + +export type KindReport = { + kind: string; + obligation: OwnableKind["obligation"]; + storage: OwnableKind["storage"]; + inProgress: number; + truncated: boolean; + verdicts: { gone: number; alive: number; unprovable: number }; + reasons: Record; + coverageGap?: string; + error?: string; +}; + +export type OwnershipReport = { + at: string; + inProgress: number; + kinds: KindReport[]; + violations: InvariantViolation[]; + /** Violations first seen in this pass (written to the activity log). */ + newViolations: InvariantViolation[]; + /** Invariants this pass cannot measure yet, and why. */ + coverageGaps: Record; + /** Owners proven gone for the first time, whose I2 check needs one more pass. */ + awaitingSecondProof: number; + /** Earliest time a record inside a grace window becomes checkable, if any. */ + nextDueAt: Date | null; + /** Work left for a later pass: failed listings or checks, unsaved + * violations, and an unfinished I4 window. The worker keeps a retry due. */ + retryNeeded: boolean; +}; + +const COVERAGE_GAPS: Record = { + I5_work_based_budgets: "Needs the section B outcome records (accepted turn.start) to tell charged from uncharged attempts.", + I6_generation_fencing: "Runner events carry no generation stamp until section G lands.", + I8_fast_turns: "Measured per run by the speed-budget check in the run log, not by this pass.", + I7_scope: "I7 measures agent-assigned in_progress issues. todo, in_review and blocked issues are not measured yet.", + record_cap: "Each kind, and the I3 and I7 checks, read at most 2,000 records per pass; a larger kind is reported as truncated.", + wake_coverage: "Writers of workspace runtime services, remote workspace operations, snapshots, restore points, " + + "git cache intents, provider artifacts and harness backups publish no work signal. Those kinds are read at " + + "startup and on any other report wake. Phase 2 adds a signal when each kind moves to the shared settle path.", +}; + +/** In-memory state kept across passes of one process. */ +export type OwnershipReportState = { + /** Records whose owner was proven gone in the previous pass (rule I2 needs two passes). */ + goneLastPass: Set; + /** Violation keys already written to the activity log by this process. */ + reported: Set; + /** Keyset cursor of the wake-request scan of rule I4 (updated_at, id). */ + intentCursor: { at: string; id: string } | null; + /** Violations whose activity row failed to save, with failed attempts; + * retried on later passes and dropped after `MAX_WRITE_ATTEMPTS`. */ + unwritten: Map; + /** True when the I4 scan stopped with more rows to read. */ + intentBacklog: boolean; +}; + +export function createOwnershipReportState(): OwnershipReportState { + return { goneLastPass: new Set(), reported: new Set(), intentCursor: null, unwritten: new Map(), intentBacklog: false }; +} + +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; +const ACTIVE_RUN_STATUSES = ["queued", "running", "scheduled_retry"]; +/** Accounting closure deadline used by the admission gate (run-cost-accounting.ts). */ +const ACCOUNTING_CLOSURE_MS = 10 * 60_000; +/** An in-progress issue must show a next action within this time. */ +const NEXT_ACTION_GRACE_MS = 5 * 60_000; +/** Records read per kind (and per I3/I7 check) per pass; beyond this the + * kind is reported truncated (a named coverage gap). */ +const MAX_RECORDS_PER_KIND = 2_000; +/** Failed activity writes of one violation before it is dropped. */ +const MAX_WRITE_ATTEMPTS = 5; +/** First I4 scan after boot looks back this far. */ +const INTENT_LOOKBACK_MS = 60 * 60_000; +/** Refusal texts that name a temporary condition (invariant 4), as a + * PostgreSQL case-insensitive regular expression. */ +const TEMPORARY_REFUSAL_PATTERN = "budget|paus|lock|busy|unavailable|retry|admission|hold|drain|restart|capacity|rate.?limit|timeout"; + +type Classification = { verdict: OwnerProof["verdict"]; reason: string }; + +async function classifyRunOwners(db: Db, records: InProgressRecord[]): Promise> { + const runIds = [...new Set(records.flatMap(record => record.owner.kind === "run" ? [record.owner.runId] : []))]; + const proofs = new Map(); + for (let index = 0; index < runIds.length; index += 500) { + const batch = runIds.slice(index, index + 500); + const rows = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status }).from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, batch)); + const byId = new Map(rows.map(row => [row.id, row.status])); + for (const runId of batch) { + const status = byId.get(runId); + proofs.set(runId, status === undefined ? { verdict: "gone", reason: "owner_run_missing" } + : TERMINAL_RUN_STATUSES.includes(status) ? { verdict: "gone", reason: "owner_run_terminal" } + : { verdict: "alive", reason: "owner_run_active" }); + } + } + return proofs; +} + +async function classify(owner: RecordOwner, runProofs: Map, probe: ProcessProbe | undefined, + now: Date): Promise { + switch (owner.kind) { + case "process": return proveOwnerGone(owner.stamp, probe); + case "fence": return proveOwnerGone(owner.stamp, owner.fence, now); + case "run": return runProofs.get(owner.runId) ?? { verdict: "unprovable", reason: "owner_run_unread" }; + case "operator": return { verdict: "alive", reason: "operator_owned" }; + case "none": return { verdict: "unprovable", reason: "no_owner_stamp" }; + } +} + +const violationKey = (violation: InvariantViolation) => `${violation.rule}:${violation.kind}:${violation.recordId}`; + +/** Admission still waits on accounting (inventory finding 3). This mirrors + * the `waitingRunCount` gate of budget-exposure.ts, including late-receipt + * debt: a terminal run with + * pending accounting holds its company's new work unless it is a succeeded + * run with observed cost, or a native run whose coordinator is still + * recovering. Only companies with an active budget policy have the gate, and + * only runs past the accounting closure deadline are reported. */ +async function checkAdmissionIndependence(db: Db, now: Date, limit: number): Promise { + const usage = heartbeatRuns.usageJson; + const lateReceiptDebt = sql`coalesce(${usage}->>'accountingLateReceiptPending', 'false') = 'true'`; + const boundedSucceeded = sql`(${heartbeatRuns.status} = 'succeeded' and ${usage}->'budgetExposure'->>'costCents' is not null)`; + const closingReceipt = sql`${usage}->>'accountingReceiptReady' = 'false' and ${usage}->>'accountingCaptureFailed' is distinct from 'true'`; + const recovering = sql`(${heartbeatRuns.runtimeMode} = 'native' and not (${lateReceiptDebt}) and exists (select 1 from ${nativeRunFinalizations} + where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.resultId} is null + and (${nativeRunFinalizations.phase} <> 'terminal_failure' or (${closingReceipt}))))`; + const gated = sql`exists (select 1 from ${budgetPolicies} where ${budgetPolicies.companyId} = ${heartbeatRuns.companyId} + and ${budgetPolicies.isActive} = true + and (${budgetPolicies.scopeType} = 'company' + or (${budgetPolicies.scopeType} = 'agent' and ${budgetPolicies.scopeId} = ${heartbeatRuns.agentId}) + or (${budgetPolicies.scopeType} = 'project' and ${usage}->'ledgerScope'->>'projectId' = ${budgetPolicies.scopeId}::text)) + and (${budgetPolicies.windowKind} <> 'calendar_month_utc' or (${lateReceiptDebt}) + or ${heartbeatRuns.finishedAt} >= (date_trunc('month', ${now.toISOString()}::timestamptz at time zone 'UTC') at time zone 'UTC')))`; + const violations: InvariantViolation[] = []; + for (let offset = 0; offset < MAX_RECORDS_PER_KIND; offset += limit) { + const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.issueId }) + .from(heartbeatRuns).where(and(or(eq(heartbeatRuns.costAccountingPending, true), lateReceiptDebt), + inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES), + lt(heartbeatRuns.finishedAt, new Date(now.getTime() - ACCOUNTING_CLOSURE_MS)), + sql`not ${boundedSucceeded}`, sql`not ${recovering}`, gated)) + .orderBy(asc(heartbeatRuns.id)).limit(limit).offset(offset); + for (const row of rows) { + violations.push({ rule: "I3_admission_independent", kind: "run_accounting_debt", recordId: row.id, + companyId: row.companyId, runId: row.id, issueId: row.issueId, reason: "accounting_debt_holds_admission" }); + } + if (rows.length < limit) break; + } + return violations; +} + +/** A wake request that ended terminally on a refusal that names a temporary + * condition. Heuristic until section E gives wake requests closed reasons. The + * filter runs in SQL and the scan pages by (updated_at, id), so a busy window + * is read in full across passes and unrelated failures never hide a match. */ +async function checkExactlyOnceIntents(db: Db, state: OwnershipReportState, now: Date, limit: number): Promise { + // The cursor keeps PostgreSQL's microsecond text form: a JS Date would round + // it down and read the boundary row again. + const cursor = state.intentCursor ?? { at: new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString(), + id: "00000000-0000-0000-0000-000000000000" }; + const violations: InvariantViolation[] = []; + let position = cursor; + for (let page = 0; page < 10; page++) { + const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + runId: agentWakeupRequests.runId, updatedAt: sql`${agentWakeupRequests.updatedAt}::text` }) + .from(agentWakeupRequests).where(and(inArray(agentWakeupRequests.status, ["skipped", "failed"]), + sql`(${agentWakeupRequests.updatedAt}, ${agentWakeupRequests.id}) > (${position.at}::timestamptz, ${position.id}::uuid)`, + sql`(coalesce(${agentWakeupRequests.error}, '') || ' ' || coalesce(${agentWakeupRequests.reason}, '')) ~* ${TEMPORARY_REFUSAL_PATTERN}`)) + .orderBy(asc(agentWakeupRequests.updatedAt), asc(agentWakeupRequests.id)).limit(limit); + for (const row of rows) { + violations.push({ rule: "I4_exactly_once_intent", kind: "wake_request", recordId: row.id, companyId: row.companyId, + runId: row.runId, issueId: null, reason: "terminal_on_temporary_refusal" }); + } + if (rows.length > 0) position = { at: rows[rows.length - 1]!.updatedAt, id: rows[rows.length - 1]!.id }; + if (rows.length < limit) { state.intentBacklog = false; break; } + // Ten full pages: the rest of a busy window waits for the next pass. + state.intentBacklog = page === 9; + } + state.intentCursor = position; + return violations; +} + +/** An agent-assigned in-progress issue with no live run, no pending wake and + * no open recovery action has no next action (invariant 7). Issues still in + * their grace window return the time they become checkable. */ +async function checkVisibleNextAction(db: Db, now: Date, limit: number): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { + const rows: Array<{ id: string; companyId: string; executionRunId: string | null; updatedAt: Date }> = []; + for (let offset = 0; offset < MAX_RECORDS_PER_KIND; offset += limit) { + const page = await db.select({ id: issues.id, companyId: issues.companyId, executionRunId: issues.executionRunId, + updatedAt: issues.updatedAt }) + .from(issues).where(and(eq(issues.status, "in_progress"), isNotNull(issues.assigneeAgentId), + // A saved monitor deadline is a first-class wait path (SPEC-implementation). + or(isNull(issues.monitorNextCheckAt), lt(issues.monitorNextCheckAt, now)), + notExists(db.select({ one: sql`1` }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, issues.companyId), inArray(heartbeatRuns.status, ACTIVE_RUN_STATUSES), + or(eq(heartbeatRuns.issueId, issues.id), eq(heartbeatRuns.nativeIssueId, issues.id))))), + notExists(db.select({ one: sql`1` }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, issues.companyId), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution", "claimed"]), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issues.id}::text`))), + notExists(db.select({ one: sql`1` }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, issues.companyId), eq(issueRecoveryActions.sourceIssueId, issues.id), + isNull(issueRecoveryActions.resolvedAt)))), + )).orderBy(asc(issues.updatedAt), asc(issues.id)).limit(limit).offset(offset); + rows.push(...page); + if (page.length < limit) break; + } + const cutoff = now.getTime() - NEXT_ACTION_GRACE_MS; + const violations: InvariantViolation[] = []; + let nextDueAt: Date | null = null; + for (const row of rows) { + if (row.updatedAt.getTime() < cutoff) { + violations.push({ rule: "I7_visible_next_action", kind: "issue", recordId: row.id, companyId: row.companyId, + runId: row.executionRunId, issueId: row.id, reason: row.executionRunId ? "lock_without_live_run" : "no_owner_no_wake" }); + } else { + // Rows are ordered by updated_at, so the first one in grace is earliest. + nextDueAt ??= new Date(row.updatedAt.getTime() + NEXT_ACTION_GRACE_MS); + } + } + return { violations, nextDueAt }; +} + +export async function runOwnershipReconciliationPass(db: Db, options: { + state: OwnershipReportState; + kinds?: readonly OwnableKind[]; + limitPerKind?: number; + probe?: ProcessProbe; + now?: Date; + /** Write new violations to the activity log. Defaults to true. */ + record?: boolean; +}): Promise { + const now = options.now ?? new Date(); + const limit = options.limitPerKind ?? 200; + const kinds = options.kinds ?? OWNABLE_KINDS; + const state = options.state; + const kindReports: KindReport[] = []; + const violations: InvariantViolation[] = []; + const goneThisPass = new Set(); + let retryNeeded = false; + + for (const kind of kinds) { + const report: KindReport = { kind: kind.id, obligation: kind.obligation, storage: kind.storage, inProgress: 0, + truncated: false, verdicts: { gone: 0, alive: 0, unprovable: 0 }, reasons: {}, + ...(kind.coverageGap ? { coverageGap: kind.coverageGap } : {}) }; + kindReports.push(report); + if (!kind.listInProgress) continue; + // Read the whole kind in one pass, a page at a time, so rule I2 compares + // the same records on consecutive passes. Past the cap the kind is + // reported as truncated; that alone schedules no follow-up. + const records: InProgressRecord[] = []; + try { + for (let offset = 0; ; offset += limit) { + const listed = await kind.listInProgress(db, limit, offset); + records.push(...listed.records); + if (!listed.truncated) break; + if (records.length >= MAX_RECORDS_PER_KIND) { report.truncated = true; break; } + } + await resolveFenceHolders(db, records); + } catch (error) { + report.error = error instanceof Error ? error.message : String(error); + retryNeeded = true; + continue; + } + // Offset pages can repeat a row when a lower id is inserted between pages. + const unique = [...new Map(records.map(record => [record.recordId, record])).values()]; + report.inProgress = unique.length; + const runProofs = await classifyRunOwners(db, unique); + for (const record of unique) { + const proof = await classify(record.owner, runProofs, options.probe, now); + report.verdicts[proof.verdict] += 1; + report.reasons[proof.reason] = (report.reasons[proof.reason] ?? 0) + 1; + const base = { kind: kind.id, recordId: record.recordId, companyId: record.companyId, + runId: record.runId, issueId: record.issueId }; + if (kind.invariantChecks.includes(1) && record.owner.kind === "none") { + violations.push({ ...base, rule: "I1_owned", reason: "no_owner_stamp" }); + } + if (kind.invariantChecks.includes(2) && proof.verdict === "gone") { + const key = `${kind.id}:${record.recordId}`; + goneThisPass.add(key); + // Settled means terminal within one reconciler pass of the proof. + if (state.goneLastPass.has(key)) violations.push({ ...base, rule: "I2_settled", reason: proof.reason }); + } + } + } + + let nextDueAt: Date | null = null; + for (const check of [ + () => checkAdmissionIndependence(db, now, limit), + () => checkExactlyOnceIntents(db, state, now, limit), + async () => { + const result = await checkVisibleNextAction(db, now, limit); + nextDueAt = result.nextDueAt; + return result.violations; + }, + ]) { + try { + violations.push(...await check()); + } catch (error) { + retryNeeded = true; + logger.warn({ err: error }, "ownership invariant check failed; the report continues"); + } + } + + // Only a first proof needs a second pass; a record already reported for I2 + // is not a reason to keep scanning. + if (state.intentBacklog) retryNeeded = true; + const awaitingSecondProof = [...goneThisPass].filter(key => !state.goneLastPass.has(key)).length; + state.goneLastPass = goneThisPass; + // Violations whose activity row failed last time are written again, even + // when a cursor-based check (I4) does not read them a second time. + const byKey = new Map(violations.map(violation => [violationKey(violation), violation])); + for (const [key, entry] of state.unwritten) if (!byKey.has(key)) byKey.set(key, entry.violation); + violations.length = 0; + violations.push(...byKey.values()); + const newViolations = violations.filter(violation => !state.reported.has(violationKey(violation))); + if (options.record !== false) { + for (const violation of newViolations) { + if (!violation.companyId) continue; + try { + await persistActivity(db, { + companyId: violation.companyId, actorType: "system", actorId: OWNERSHIP_RECONCILER_ACTOR_ID, + action: OWNERSHIP_INVARIANT_VIOLATION_ACTION, entityType: "ownership_record", entityId: violation.recordId, + details: { rule: violation.rule, kind: violation.kind, reason: violation.reason, + runId: violation.runId, issueId: violation.issueId, mode: "report_only" }, + }); + } catch (error) { + logger.warn({ err: error, rule: violation.rule, kind: violation.kind }, "failed to record ownership invariant violation"); + const key = violationKey(violation); + const attempts = (state.unwritten.get(key)?.attempts ?? 0) + 1; + if (attempts >= MAX_WRITE_ATTEMPTS) { + // A poison row (for example a company that no longer exists) must + // not keep the report retrying for the life of the process. + state.unwritten.delete(key); + state.reported.add(key); + } else { + state.unwritten.set(key, { violation, attempts }); + retryNeeded = true; + } + continue; + } + state.unwritten.delete(violationKey(violation)); + state.reported.add(violationKey(violation)); + } + for (const violation of newViolations) { + if (violation.companyId) continue; + // No company to attach an activity row to (for example an unreadable + // spool file): the server log is the only record. + logger.warn({ rule: violation.rule, kind: violation.kind, recordId: violation.recordId, reason: violation.reason }, + "ownership invariant violation without a company"); + state.reported.add(violationKey(violation)); + } + } + + const inProgress = kindReports.reduce((total, report) => total + report.inProgress, 0); + const report: OwnershipReport = { at: now.toISOString(), inProgress, kinds: kindReports, violations, + newViolations, coverageGaps: COVERAGE_GAPS, awaitingSecondProof, nextDueAt, retryNeeded }; + if (newViolations.length > 0 || kindReports.some(kind => kind.error)) { + logger.info({ + inProgress, violations: violations.length, newViolations: newViolations.length, + byRule: countBy(violations, violation => violation.rule), + byKind: countBy(violations, violation => violation.kind), + errors: kindReports.filter(kind => kind.error).map(kind => ({ kind: kind.kind, error: kind.error })), + }, "ownership reconciliation report"); + } + return report; +} + +function countBy(items: T[], key: (item: T) => string): Record { + const counts: Record = {}; + for (const item of items) counts[key(item)] = (counts[key(item)] ?? 0) + 1; + return counts; +} + +/** A stable fingerprint of the in-progress set, so the worker can tell a + * changing set from a stuck one. */ +export function ownershipReportFingerprint(report: OwnershipReport): string { + return report.kinds.map(kind => `${kind.kind}:${kind.inProgress}:${kind.verdicts.gone}:${kind.verdicts.alive}`).join("|"); +} diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts new file mode 100644 index 00000000000..40aeecdb9ed --- /dev/null +++ b/server/src/services/recovery/ownership-registry.ts @@ -0,0 +1,681 @@ +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { and, asc, eq, inArray, isNotNull, isNull, like, notInArray, notLike, or, sql, type SQL } from "drizzle-orm"; +import { + agentWakeupRequests, + environmentLeases, + executionWorkspaceRestorePoints, + executionWorkspaceRuntimeLeases, + executionWorkspaceSnapshots, + executionWorkspaces, + heartbeatRuns, + issueRecoveryActions, + issues, + nativeHarnessBackups, + nativeProviderArtifacts, + nativeRunFinalizations, + sandboxCreationOperations, + workspaceOperations, + workspaceRuntimeServices, + type Db, +} from "@paperclipai/db"; +import { EXECUTION_RECONCILIATION_CAUSES } from "@paperclipai/shared"; +import { readControllerProcess } from "../controller-process-identity.js"; +import { idleAccountingSpoolPath, idleOrphanSpoolPath } from "../idle-local-work.js"; +import { + NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT, + NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, +} from "../native-runtime/native-runner-ownership.js"; +import type { LocalOwnerStamp, RemoteFenceState, RemoteOwnerStamp } from "../owner-proof.js"; + +/** + * Registry of ownable record kinds (plan section C). + * + * Each kind lists its in-progress records and says who owns each one. Phase 1 + * is report-only: no kind registers a `settle`, so the reconciler classifies + * owners and runs invariant checks but never changes a record. Phase 2 adds a + * compare-and-swap `settle` per kind, one PR at a time, and deletes the kind's + * bespoke release path. + * + * The kinds are the section-C rows of doc/plans/2026-11-07-ownership-inventory.md. + * Each kind names exactly one obligation; settling one never settles another. + */ + +export type Obligation = "ownership" | "accepted_result" | "workspace_preservation" | "accounting"; + +/** Who may move a record forward. */ +export type RecordOwner = + /** A process, proved by `proveOwnerGone(stamp)`. */ + | { kind: "process"; stamp: LocalOwnerStamp } + /** A fence or lease token, proved against the fence's current holder. */ + | { kind: "fence"; stamp: RemoteOwnerStamp; fence: RemoteFenceState } + /** A run id only. The record's owner is that run's execution. */ + | { kind: "run"; runId: string } + /** A person or agent assignment, such as an operator-owned recovery action. */ + | { kind: "operator"; ownerType: string | null } + /** Nothing on the record names an owner (invariant 1). */ + | { kind: "none" }; + +export type InProgressRecord = { + recordId: string; + companyId: string | null; + runId: string | null; + issueId: string | null; + owner: RecordOwner; + /** When the record last changed, when the storage records it. */ + updatedAt: Date | null; +}; + +export type ListResult = { records: InProgressRecord[]; truncated: boolean }; + +export type OwnableKind = { + id: string; + title: string; + obligation: Obligation; + storage: "database" | "local_disk"; + /** Null when Phase 1 cannot enumerate the kind; the report names the gap. */ + /** Lists one page in a stable order; the report pages through the kind + * within one pass. */ + listInProgress: ((db: Db, limit: number, offset: number) => Promise) | null; + /** Report-only in Phase 1. Phase 2 registers a compare-and-swap settle. */ + settle: null; + /** Plan invariants this kind is checked against (1–8). */ + invariantChecks: readonly number[]; + /** Why the kind cannot be listed, when `listInProgress` is null. */ + coverageGap?: string; +}; + +const TERMINAL_PHASES = ["committed", "terminal_failure", "applied"]; + +function page(rows: T[], limit: number): { rows: T[]; truncated: boolean } { + return rows.length > limit ? { rows: rows.slice(0, limit), truncated: true } : { rows, truncated: false }; +} + +function json(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +function str(value: unknown): string | null { + return typeof value === "string" && value ? value : null; +} + +function int(value: unknown): number | null { + return typeof value === "number" && Number.isInteger(value) ? value : null; +} + +const runOwner = (runId: string | null | undefined): RecordOwner => runId ? { kind: "run", runId } : { kind: "none" }; + +// ---------------------------------------------------------------------------- +// native_run_finalizations + +type NrfRow = typeof nativeRunFinalizations.$inferSelect; + +/** NRF records the controller's boot, PID, start time and, since #19868, its + * PID space (`controller_host_hash`). Rows written before that have no space + * and stay unprovable for a foreign boot (inventory finding 2). */ +function nrfStamp(row: NrfRow, scope: string): LocalOwnerStamp { + return { + kind: "local", scope, generation: row.controllerGeneration ?? null, host: null, + processSpace: row.controllerHostHash ?? null, + bootId: row.controllerBootId, pid: row.controllerPid, + processStartedAt: row.controllerProcessStartedAt?.toISOString() ?? null, processGroupId: null, + }; +} + +function nrfKind(id: string, title: string, obligation: Obligation, where: SQL | undefined, + owner: (row: NrfRow) => RecordOwner = row => ({ kind: "process", stamp: nrfStamp(row, id) }), + invariantChecks: readonly number[] = [1, 2]): OwnableKind { + return { + id, title, obligation, storage: "database", settle: null, invariantChecks, + listInProgress: async (db, limit, offset) => { + const rows = await db.select().from(nativeRunFinalizations).where(where).orderBy(asc(nativeRunFinalizations.runId)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.runId, companyId: row.companyId, runId: row.runId, issueId: row.issueId, + owner: owner(row), updatedAt: row.updatedAt, + })) }; + }, + }; +} + +const notFinalizerOrCleanup = and(notLike(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), + notLike(nativeRunFinalizations.leaseOwner, "native-cleanup:%")); + +/** A token-only NRF lease (finalizer or cleanup) is a fence held by the token. */ +function nrfFenceOwner(row: NrfRow, scope: string): RecordOwner { + return { + kind: "fence", + stamp: { kind: "remote", scope, generation: row.controllerGeneration ?? null, fenceToken: row.leaseOwner }, + fence: { currentFenceToken: row.leaseOwner, currentGeneration: row.controllerGeneration ?? null, + leaseExpiresAt: row.leaseExpiresAt }, + }; +} + +// ---------------------------------------------------------------------------- +// heartbeat_runs + +type RunRow = Pick; + +const runColumns = { + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.issueId, status: heartbeatRuns.status, + controllerBootId: heartbeatRuns.controllerBootId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, processHostHash: heartbeatRuns.processHostHash, + updatedAt: heartbeatRuns.updatedAt, + runnerProfileJson: heartbeatRuns.runnerProfileJson, +}; + +function runKind(id: string, title: string, obligation: Obligation, where: SQL | undefined, + owner: (row: RunRow) => RecordOwner, invariantChecks: readonly number[] = [1, 2]): OwnableKind { + return { + id, title, obligation, storage: "database", settle: null, invariantChecks, + listInProgress: async (db, limit, offset) => { + const rows = await db.select(runColumns).from(heartbeatRuns).where(where).orderBy(asc(heartbeatRuns.id)).limit(limit + 1).offset(offset); + const result = page(rows as RunRow[], limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.id, issueId: row.issueId, + owner: owner(row), updatedAt: row.updatedAt, + })) }; + }, + }; +} + +/** heartbeat_runs records the adapter child's PID, start time and, since + * #19871, its PID space (`process_host_hash`). Older rows have no space and + * stay unprovable. */ +function runProcessStamp(row: RunRow, scope: string): LocalOwnerStamp { + return { + kind: "local", scope, generation: null, host: null, processSpace: row.processHostHash ?? null, bootId: null, + pid: row.processPid, processStartedAt: row.processStartedAt?.toISOString() ?? null, + processGroupId: row.processGroupId, + }; +} + +// ---------------------------------------------------------------------------- +// environment_leases + +type LeaseRow = Pick; + +function leaseKind(id: string, title: string, obligation: Obligation, where: SQL | undefined, + owner: (row: LeaseRow) => RecordOwner, invariantChecks: readonly number[] = [1, 2]): OwnableKind { + return { + id, title, obligation, storage: "database", settle: null, invariantChecks, + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: environmentLeases.id, companyId: environmentLeases.companyId, + heartbeatRunId: environmentLeases.heartbeatRunId, issueId: environmentLeases.issueId, + metadata: environmentLeases.metadata, updatedAt: environmentLeases.updatedAt }) + .from(environmentLeases).where(where).orderBy(asc(environmentLeases.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.heartbeatRunId, issueId: row.issueId, + owner: owner(row), updatedAt: row.updatedAt, + })) }; + }, + }; +} + +/** `ControllerProcess` metadata ({bootId, pid, hostHash, processStartedAt}). */ +function controllerStamp(value: unknown, scope: string): RecordOwner { + const controller = readControllerProcess(value); + if (!controller) return { kind: "none" }; + return { kind: "process", stamp: { + kind: "local", scope, generation: null, host: null, processSpace: controller.hostHash, + bootId: controller.bootId, pid: controller.pid, processStartedAt: controller.processStartedAt, processGroupId: null, + } }; +} + +// ---------------------------------------------------------------------------- +// workspace_operations + +type OperationRow = Pick; + +function operationKind(id: string, title: string, obligation: Obligation, where: SQL | undefined, + owner: (row: OperationRow) => RecordOwner): OwnableKind { + return { + id, title, obligation, storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: workspaceOperations.id, companyId: workspaceOperations.companyId, + heartbeatRunId: workspaceOperations.heartbeatRunId, issueId: workspaceOperations.issueId, + metadata: workspaceOperations.metadata, updatedAt: workspaceOperations.updatedAt }) + .from(workspaceOperations).where(where).orderBy(asc(workspaceOperations.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.heartbeatRunId, issueId: row.issueId, + owner: owner(row), updatedAt: row.updatedAt, + })) }; + }, + }; +} + +// ---------------------------------------------------------------------------- +// issue_recovery_actions + +type RecoveryActionRow = Pick; + +function recoveryActionKind(id: string, title: string, obligation: Obligation, where: SQL | undefined, + owner: (row: RecoveryActionRow) => RecordOwner, invariantChecks: readonly number[] = [1, 2, 7], + runIdOf: (row: RecoveryActionRow) => string | null = () => null): OwnableKind { + return { + id, title, obligation, storage: "database", settle: null, invariantChecks, + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: issueRecoveryActions.id, companyId: issueRecoveryActions.companyId, + sourceIssueId: issueRecoveryActions.sourceIssueId, ownerType: issueRecoveryActions.ownerType, + evidence: issueRecoveryActions.evidence, updatedAt: issueRecoveryActions.updatedAt }) + .from(issueRecoveryActions).where(and(isNull(issueRecoveryActions.resolvedAt), where)).orderBy(asc(issueRecoveryActions.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: runIdOf(row), issueId: row.sourceIssueId, + owner: owner(row), updatedAt: row.updatedAt, + })) }; + }, + }; +} + +// ---------------------------------------------------------------------------- +// Local-disk spools + +/** Spool files carry their company and run (`companyId` plus `runId` or + * `heartbeatRunId`); a file that cannot be read keeps a null attribution. */ +async function readSpoolAttribution(file: string): Promise<{ companyId: string | null; runId: string | null }> { + try { + const handle = await fs.open(file, "r"); + try { + const buffer = Buffer.alloc(64 * 1024); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0); + const record = json(JSON.parse(buffer.subarray(0, bytesRead).toString("utf8"))); + return { companyId: str(record.companyId), runId: str(record.runId) ?? str(record.heartbeatRunId) }; + } finally { + await handle.close(); + } + } catch { + return { companyId: null, runId: null }; + } +} + +function spoolKind(id: string, title: string, obligation: Obligation, directory: () => string): OwnableKind { + return { + id, title, obligation, storage: "local_disk", settle: null, invariantChecks: [1, 2], + listInProgress: async (_db, limit, offset) => { + let names: string[]; + const root = directory(); + try { + names = (await fs.readdir(root)).filter(name => name.endsWith(".json")).sort(); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return { records: [], truncated: false }; + throw error; + } + const result = page(names.slice(offset), limit); + const records: InProgressRecord[] = []; + for (const name of result.rows) { + const attribution = await readSpoolAttribution(path.join(root, name)); + records.push({ recordId: path.basename(name, ".json"), companyId: attribution.companyId, + runId: attribution.runId, issueId: null, owner: { kind: "none" }, updatedAt: null }); + } + return { truncated: result.truncated, records }; + }, + }; +} + +// ---------------------------------------------------------------------------- +// The registry + +export const OWNABLE_KINDS: readonly OwnableKind[] = [ + // Native runtime and Runner + nrfKind("native_executor_claim", "Executor execution claim (fresh run)", "ownership", + and(eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId), + isNotNull(nativeRunFinalizations.leaseOwner), notFinalizerOrCleanup, isNull(nativeRunFinalizations.recoveryState))), + nrfKind("native_restart_recovery_claim", "Restart-recovery claim", "ownership", + inArray(nativeRunFinalizations.recoveryState, ["awaiting_runner_reattach", "resuming_session", "bootstrap_incomplete"])), + nrfKind("native_awaiting_evidence", "\"Awaiting evidence\" wait", "ownership", + eq(nativeRunFinalizations.recoveryState, "awaiting_evidence"), undefined, [1, 2, 7]), + nrfKind("native_accepted_result_lease", "Accepted-result executor lease", "accepted_result", + and(isNotNull(nativeRunFinalizations.resultId), notInArray(nativeRunFinalizations.phase, TERMINAL_PHASES), + isNotNull(nativeRunFinalizations.leaseOwner), notFinalizerOrCleanup)), + nrfKind("native_finalizer_lease", "Finalizer lease", "accepted_result", + and(like(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), notInArray(nativeRunFinalizations.phase, TERMINAL_PHASES)), + row => nrfFenceOwner(row, "native_finalizer_lease")), + nrfKind("native_cleanup_lease", "Cleanup lease", "workspace_preservation", + like(nativeRunFinalizations.leaseOwner, "native-cleanup:%"), + row => nrfFenceOwner(row, "native_cleanup_lease")), + runKind("native_ownership_unverified_hold", "Ownership-unverified hold", "ownership", + and(eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "running"), or( + and(eq(heartbeatRuns.errorCode, NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE), eq(heartbeatRuns.nativePhase, "terminal_failure")), + eq(heartbeatRuns.errorCode, NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT))), + () => ({ kind: "none" }), [1, 2, 7]), + { + id: "native_startup_context", title: "Native startup context", obligation: "ownership", storage: "database", + settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ ...runColumns, leaseOwner: nativeRunFinalizations.leaseOwner, + leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, generation: nativeRunFinalizations.controllerGeneration }) + .from(heartbeatRuns).leftJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId))) + .where(and(eq(heartbeatRuns.status, "running"), sql`${heartbeatRuns.runnerProfileJson} ? 'nativeStartupContext'`)) + .orderBy(asc(heartbeatRuns.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => { + const context = json(json(row.runnerProfileJson).nativeStartupContext); + const token = str(context.leaseOwner); + return { + recordId: row.id, companyId: row.companyId, runId: row.id, issueId: row.issueId, updatedAt: row.updatedAt, + owner: token ? { + kind: "fence", + stamp: { kind: "remote", scope: "native_startup_context", generation: int(context.generation), fenceToken: token }, + fence: { currentFenceToken: row.leaseOwner ?? null, currentGeneration: row.generation ?? null, + leaseExpiresAt: row.leaseExpiresAt ?? null }, + } : { kind: "none" }, + }; + }) }; + }, + }, + runKind("native_pending_workspace_export", "Pending workspace export (sync reference)", "workspace_preservation", + sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared'`, + row => runOwner(row.id), [1, 2, 7]), + { + id: "runtime_skill_cache_lock", title: "Runtime skill cache publication lock", obligation: "ownership", + storage: "local_disk", settle: null, invariantChecks: [2, 7, 8], listInProgress: null, + coverageGap: "Lock files live in per-skill cache roots under each managed skill directory; Phase 1 does not walk them.", + }, + { + id: "native_harness_backup", title: "Native harness backup publication", obligation: "workspace_preservation", + storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.status, "pending")).orderBy(asc(nativeHarnessBackups.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.runId, issueId: row.issueId, + owner: runOwner(row.runId), updatedAt: row.updatedAt, + })) }; + }, + }, + { + id: "native_provider_artifact_upload", title: "Provider artifact upload", obligation: "workspace_preservation", + storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: nativeProviderArtifacts.id, companyId: nativeProviderArtifacts.companyId, + updatedAt: nativeProviderArtifacts.updatedAt }).from(nativeProviderArtifacts) + .where(inArray(nativeProviderArtifacts.status, ["pending", "retiring"])).orderBy(asc(nativeProviderArtifacts.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: null, issueId: null, + owner: { kind: "none" }, updatedAt: row.updatedAt, + })) }; + }, + }, + + // Environments and sandboxes + { + id: "sandbox_creation_intent", title: "Sandbox creation intent", obligation: "ownership", storage: "database", + settle: null, invariantChecks: [1, 2, 3, 7], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: sandboxCreationOperations.id, companyId: sandboxCreationOperations.companyId, + runId: sandboxCreationOperations.runId, issueId: sandboxCreationOperations.issueId, + metadata: sandboxCreationOperations.metadata, updatedAt: sandboxCreationOperations.updatedAt }) + .from(sandboxCreationOperations).where(eq(sandboxCreationOperations.state, "pending")).orderBy(asc(sandboxCreationOperations.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.runId, issueId: row.issueId, + owner: controllerStamp(json(row.metadata).creationController, "sandbox_creation_intent"), updatedAt: row.updatedAt, + })) }; + }, + }, + leaseKind("environment_run_lease_active", "Run lease (active)", "ownership", + and(eq(environmentLeases.status, "active"), isNotNull(environmentLeases.heartbeatRunId)), + row => runOwner(row.heartbeatRunId)), + leaseKind("environment_teardown_claim", "Teardown claim (in-flight cleanup)", "ownership", + and(eq(environmentLeases.status, "pending_cleanup"), sql`(${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean is true`), + row => controllerStamp(json(row.metadata).pendingCleanupController, "environment_teardown_claim")), + leaseKind("environment_pending_cleanup", "Pending-cleanup obligation (not in flight)", "ownership", + and(eq(environmentLeases.status, "pending_cleanup"), + sql`coalesce((${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean, false) is false`), + () => ({ kind: "none" }), [1, 2, 7]), + leaseKind("environment_stop_and_retain_intent", "Stop-and-retain intent", "workspace_preservation", + and(eq(environmentLeases.status, "pending_cleanup"), sql`${environmentLeases.metadata} ? 'sandboxStopAndRetain'`), + row => runOwner(row.heartbeatRunId)), + leaseKind("environment_native_export_resume", "Native export resume (stop-only)", "workspace_preservation", + and(eq(environmentLeases.status, "pending_cleanup"), sql`${environmentLeases.metadata} ? 'nativeWorkspaceExportResume'`), + row => runOwner(row.heartbeatRunId)), + leaseKind("environment_retained_hold", "Retained / reusable hold", "workspace_preservation", + eq(environmentLeases.status, "retained"), () => ({ kind: "none" }), [1, 7]), + spoolKind("sandbox_orphan_cleanup_spool", "Orphan cleanup buffer/spool", "ownership", idleOrphanSpoolPath), + leaseKind("environment_login_probe_lease", "Login/probe lease", "ownership", + and(eq(environmentLeases.status, "active"), isNull(environmentLeases.heartbeatRunId)), () => ({ kind: "none" })), + + // Workspaces + operationKind("workspace_runtime_control_operation", "Managed runtime-control operation", "ownership", + and(eq(workspaceOperations.status, "running"), sql`${workspaceOperations.metadata} ? 'runtimeControlOwner'`), + row => { + const owner = json(json(row.metadata).runtimeControlOwner); + const pid = int(owner.pid); + // The owner stores a per-process UUID, a PID and, since #19868, its PID + // space; no start time, so a live PID stays unprovable. + return pid ? { kind: "process", stamp: { kind: "local", scope: "workspace_runtime_control_operation", + generation: null, host: null, processSpace: str(owner.hostHash), bootId: str(owner.ownerId), pid, + processStartedAt: null, processGroupId: null } } : { kind: "none" }; + }), + operationKind("native_workspace_finalize_attempt", "Native workspace finalize attempt (copyback export)", "workspace_preservation", + and(eq(workspaceOperations.phase, "workspace_finalize"), inArray(workspaceOperations.status, ["running", "pending"])), + row => runOwner(row.heartbeatRunId)), + operationKind("remote_provision_command", "Remote provision command intent", "workspace_preservation", + and(eq(workspaceOperations.phase, "remote_workspace_provision"), or(eq(workspaceOperations.status, "running"), + sql`(${workspaceOperations.metadata}->>'uncertain')::boolean is true`)), + row => runOwner(row.heartbeatRunId)), + operationKind("remote_snapshot_save", "Remote snapshot save", "workspace_preservation", + and(eq(workspaceOperations.phase, "remote_workspace_snapshot"), inArray(workspaceOperations.status, ["running", "pending"])), + row => runOwner(row.heartbeatRunId)), + operationKind("remote_workspace_migration", "Remote workspace migration op", "workspace_preservation", + and(eq(workspaceOperations.phase, "remote_workspace_migration"), eq(workspaceOperations.status, "running")), + () => ({ kind: "none" })), + { + id: "execution_workspace_runtime_lease", title: "Execution workspace runtime lease", obligation: "ownership", + storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select().from(executionWorkspaceRuntimeLeases).orderBy(asc(executionWorkspaceRuntimeLeases.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.ownerRunId, issueId: row.ownerIssueId, + owner: row.ownerRunId ? runOwner(row.ownerRunId) : { kind: "operator", ownerType: row.ownerKey.split(":")[0] ?? null }, + updatedAt: row.updatedAt, + })) }; + }, + }, + { + id: "workspace_runtime_service", title: "Workspace runtime service (local process)", obligation: "ownership", + storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: workspaceRuntimeServices.id, companyId: workspaceRuntimeServices.companyId, + issueId: workspaceRuntimeServices.issueId, updatedAt: workspaceRuntimeServices.updatedAt }) + .from(workspaceRuntimeServices) + .where(inArray(workspaceRuntimeServices.status, ["provisioning", "starting", "running"])).orderBy(asc(workspaceRuntimeServices.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + // The PID lives in a local service registry file, not on the row. + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: null, issueId: row.issueId, + owner: { kind: "none" }, updatedAt: row.updatedAt, + })) }; + }, + }, + { + id: "remote_workspace_unsaved_files", title: "Remote workspace unsaved-files flag", obligation: "workspace_preservation", + storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: executionWorkspaces.id, companyId: executionWorkspaces.companyId, + metadata: executionWorkspaces.metadata, updatedAt: executionWorkspaces.updatedAt }).from(executionWorkspaces) + .where(sql`(${executionWorkspaces.metadata}->'remoteWorkspace'->>'dirty')::boolean is true`).orderBy(asc(executionWorkspaces.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => { + const writerRunId = str(json(json(row.metadata).remoteWorkspace).writerRunId); + return { recordId: row.id, companyId: row.companyId, runId: writerRunId, issueId: null, + owner: runOwner(writerRunId), updatedAt: row.updatedAt }; + }) }; + }, + }, + { + id: "remote_snapshot_ledger", title: "Remote snapshot ledger", obligation: "workspace_preservation", + storage: "database", settle: null, invariantChecks: [2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: executionWorkspaceSnapshots.id, companyId: executionWorkspaceSnapshots.companyId, + sourceRunId: executionWorkspaceSnapshots.sourceRunId, updatedAt: executionWorkspaceSnapshots.updatedAt }) + .from(executionWorkspaceSnapshots).where(inArray(executionWorkspaceSnapshots.status, ["pending", "uploaded"])).orderBy(asc(executionWorkspaceSnapshots.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.sourceRunId, issueId: null, + owner: runOwner(row.sourceRunId), updatedAt: row.updatedAt, + })) }; + }, + }, + { + id: "git_restore_point", title: "Git restore point", obligation: "workspace_preservation", + storage: "database", settle: null, invariantChecks: [1, 2], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: executionWorkspaceRestorePoints.id, companyId: executionWorkspaceRestorePoints.companyId, + issueId: executionWorkspaceRestorePoints.sourceIssueId, updatedAt: executionWorkspaceRestorePoints.updatedAt }) + .from(executionWorkspaceRestorePoints) + .where(inArray(executionWorkspaceRestorePoints.status, ["pending", "expiring"])).orderBy(asc(executionWorkspaceRestorePoints.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: null, issueId: row.issueId, + owner: { kind: "none" }, updatedAt: row.updatedAt, + })) }; + }, + }, + { + id: "git_cache_expiration_intent", title: "Git cache expiration intent", obligation: "workspace_preservation", + storage: "database", settle: null, invariantChecks: [1], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: executionWorkspaces.id, companyId: executionWorkspaces.companyId, + updatedAt: executionWorkspaces.updatedAt }).from(executionWorkspaces) + .where(sql`${executionWorkspaces.metadata}->'gitCacheExpiration' is not null + and coalesce((${executionWorkspaces.metadata}->'gitCacheExpiration'->>'completed')::boolean, false) is false`) + .orderBy(asc(executionWorkspaces.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: null, issueId: null, + owner: { kind: "none" }, updatedAt: row.updatedAt, + })) }; + }, + }, + leaseKind("legacy_shutdown_settlement_fence", "Legacy graceful-shutdown workspace settlement fence", "workspace_preservation", + sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`, + row => { + const bootId = str(json(json(row.metadata).legacyShutdownWorkspaceSettlement).controllerBootId); + return bootId ? { kind: "process", stamp: { kind: "local", scope: "legacy_shutdown_settlement_fence", + generation: null, host: null, processSpace: null, bootId, pid: null, processStartedAt: null } } : { kind: "none" }; + }, [2]), + leaseKind("legacy_restore_recovery_hold", "Legacy restore-recovery preservation hold", "workspace_preservation", + and(eq(environmentLeases.leasePolicy, "retain_on_failure"), sql`${environmentLeases.metadata} ? 'workspaceRestoreRecovery'`, + notInArray(environmentLeases.status, ["released", "expired", "failed"])), + row => runOwner(row.heartbeatRunId), [2, 7]), + recoveryActionKind("system_recovery_attempt", "System recovery attempt for retained work", "workspace_preservation", + sql`${issueRecoveryActions.evidence}->'systemRecovery'->>'state' in ('scheduled', 'running', 'awaiting_evidence')`, + () => ({ kind: "none" })), + recoveryActionKind("copyback_owner_hold", "Copyback-owner hold (RC-50)", "ownership", + sql`${issueRecoveryActions.evidence}->'workspaceOwnerRecovery'->>'state' = 'operator_required'`, + row => { + const owner = json(json(json(row.evidence).workspaceOwnerRecovery).owner); + const pid = int(owner.pid); + // The receipt records a hostname but no PID space: unprovable by design. + return pid ? { kind: "process", stamp: { kind: "local", scope: "copyback_owner_hold", generation: null, + host: str(owner.hostname), processSpace: null, bootId: str(owner.controllerBootId), pid, + processStartedAt: str(owner.processStartedAt), processGroupId: null } } : { kind: "none" }; + }), + recoveryActionKind("observed_workspace_preservation_fence", "Observed-run workspace preservation fence", "workspace_preservation", + sql`${issueRecoveryActions.evidence}->'observedWorkspacePreservation'->>'state' = 'stop_pending'`, + row => { + const fence = json(json(row.evidence).observedWorkspacePreservation); + const token = str(fence.leaseOwner); + // The current holder is read in the report pass (`resolveFenceHolders`). + return token ? { kind: "fence", stamp: { kind: "remote", scope: "observed_workspace_preservation_fence", + generation: int(fence.controllerGeneration), fenceToken: token }, + fence: { currentFenceToken: null, currentGeneration: null, leaseExpiresAt: null } } : { kind: "none" }; + }, [1, 2, 7], row => str(json(json(row.evidence).observedWorkspacePreservation).runId)), + + // Accounting and budgets + runKind("run_accounting_debt", "Run accounting debt", "accounting", + and(eq(heartbeatRuns.costAccountingPending, true), + inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"])), + () => ({ kind: "none" }), [2, 3]), + spoolKind("usage_receipt_spool", "Usage-receipt spool file (durable file)", "accounting", idleAccountingSpoolPath), + + // Wake queue and recovery + { + id: "wake_request_claimed", title: "Wake request (claimed)", obligation: "ownership", storage: "database", + settle: null, invariantChecks: [2, 4], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + runId: agentWakeupRequests.runId, updatedAt: agentWakeupRequests.updatedAt }).from(agentWakeupRequests) + .where(eq(agentWakeupRequests.status, "claimed")).orderBy(asc(agentWakeupRequests.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.runId, issueId: null, + owner: runOwner(row.runId), updatedAt: row.updatedAt, + })) }; + }, + }, + runKind("legacy_controller_lease", "Legacy controller lease", "ownership", + and(eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"), isNotNull(heartbeatRuns.controllerBootId)), + row => ({ kind: "process", stamp: { kind: "local", scope: "legacy_controller_lease", generation: null, host: null, + processSpace: null, bootId: row.controllerBootId, pid: null, processStartedAt: null } })), + runKind("detached_process_marker", "Detached-process marker", "ownership", + and(eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, "process_detached")), + row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "detached_process_marker") } : { kind: "none" }, + [2, 7]), + runKind("hot_restart_adoption_marker", "Hot-restart adoption marker", "ownership", + and(eq(heartbeatRuns.status, "running"), sql`${heartbeatRuns.resultJson} ? 'hotRestartAdoption'`), + row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "hot_restart_adoption_marker") } : { kind: "none" }, + [2]), + { + id: "issue_execution_lock", title: "Issue execution lock", obligation: "ownership", storage: "database", + settle: null, invariantChecks: [2, 3, 7], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: issues.id, companyId: issues.companyId, runId: issues.executionRunId, + updatedAt: issues.updatedAt }).from(issues).where(isNotNull(issues.executionRunId)).orderBy(asc(issues.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.runId, issueId: row.id, + owner: runOwner(row.runId), updatedAt: row.updatedAt, + })) }; + }, + }, + { + id: "issue_checkout_lock", title: "Issue checkout lock", obligation: "ownership", storage: "database", + settle: null, invariantChecks: [2, 7], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: issues.id, companyId: issues.companyId, runId: issues.checkoutRunId, + updatedAt: issues.updatedAt }).from(issues).where(isNotNull(issues.checkoutRunId)).orderBy(asc(issues.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.runId, issueId: row.id, + owner: runOwner(row.runId), updatedAt: row.updatedAt, + })) }; + }, + }, + recoveryActionKind("execution_reconciliation_hold", "Execution reconciliation hold", "ownership", + and(inArray(issueRecoveryActions.status, ["active", "escalated"]), + inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES])), + row => ({ kind: "operator", ownerType: row.ownerType }), [3, 7]), +]; + +/** Resolve the current holder of fences that live on another row (the + * native coordinator), so a fence stamp is judged against current truth. */ +export async function resolveFenceHolders(db: Db, records: InProgressRecord[]): Promise { + const pending = records.filter(record => record.owner.kind === "fence" && record.runId && + record.owner.fence.currentFenceToken === null && record.owner.fence.currentGeneration === null); + if (pending.length === 0) return; + const runIds = [...new Set(pending.map(record => record.runId!))]; + const holders = await db.select({ runId: nativeRunFinalizations.runId, leaseOwner: nativeRunFinalizations.leaseOwner, + generation: nativeRunFinalizations.controllerGeneration, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }) + .from(nativeRunFinalizations).where(inArray(nativeRunFinalizations.runId, runIds)); + const byRun = new Map(holders.map(holder => [holder.runId, holder])); + for (const record of pending) { + if (record.owner.kind !== "fence") continue; + const holder = byRun.get(record.runId!); + record.owner.fence = { currentFenceToken: holder?.leaseOwner ?? null, + currentGeneration: holder?.generation ?? null, leaseExpiresAt: holder?.leaseExpiresAt ?? null }; + } +}