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 }; + } +}