feat(recovery): model an unclaimed pending-cleanup item as owned by the cleanup queue (#19980)

<!-- Write all pull request text in Simplified Technical English
(ASD-STE100): short sentences, one instruction per sentence, simple
approved vocabulary, and the active voice. -->

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A sandbox that still needs teardown waits as a `pending_cleanup`
environment lease until the cleanup queue retries it
> - The ownership reconciler checks invariant I1: every in-progress
record names an owner
> - An unclaimed pending-cleanup item names no process by design, so I1
reported every one (61 in the Phase 2 data)
> - That noise hides real ownership gaps
> - This pull request models the item honestly: the cleanup queue owns
it, with a visible next action at its saved retry deadline
> - The benefit is that I1 measures real gaps, and a queue that stops
acting shows up as an I7 gap instead of silence

## Linked Issues or Issue Description

Refs #19933

**What happened?**
`environment_pending_cleanup` listed every unclaimed pending-cleanup
item with owner `none`, so I1 reported each as `no_owner_stamp`.

**Expected behavior**
The item names the queue that acts on it and its next due time. It is
reported only when the queue is not acting on it (long past the
deadline). No owner is invented, and nothing is released.

**Steps to reproduce**
1. Let a sandbox teardown fail so its lease waits in `pending_cleanup`.
2. Run the ownership reconciliation pass.
3. See an I1 `no_owner_stamp` violation for a healthy queue item.

## What Changed

- `recovery/ownership-registry.ts`: new owner kind `queue` (`queue`,
`dueAt`, `graceMs`). `environment_pending_cleanup` names the environment
cleanup queue. Its due time is the same retry deadline the cleanup
worker sleeps until, including the at-once first attempt of a settled
orphan.
- `recovery/ownership-invariant-report.ts`: a queue owner is alive
(`queue_scheduled`) while its deadline is current. When it is more than
two retry intervals past, it is unprovable (`queue_overdue`) and
reported as `I7_visible_next_action`. A queue is never proven gone and
nothing is settled.
- The listing reads the in-flight flag as text, so a malformed value
cannot break it.

## Verification

- `cd server && npx vitest run
src/__tests__/environment-pending-cleanup-queue-owner.test.ts
src/__tests__/ownership-reconciliation-report.test.ts` (embedded
Postgres).
- Red on master, green here (written after the code): a current item
names the queue (no I1); a long-overdue item is an I7 `queue_overdue`
gap.
- Green on both: an in-flight item is not listed here (it is a teardown
claim).
- Full typecheck is left to CI (local memory limit).

## Risks

- Low risk. Report only: no record is changed.
- The grace (two retry intervals) is a reporting threshold, not a
release rule. A worker that sleeps longer than that for a legitimate
reason would show as an I7 gap.
- `#19966` changes the same listing predicate to the same text
comparison; a rebase of either resolves trivially.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with
tool use (shell, file edit, embedded Postgres tests).

## 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 <noreply@anthropic.com>

server/src/__tests__/environment-pending-cleanup-queue-owner.test.ts

@@ -0,0 +1,99 @@
+import { randomUUID } from "node:crypto";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import { companies, createDb, environmentLeases } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+const suite = support.supported ? describe : describe.skip;
+const KIND = "environment_pending_cleanup";
+
+suite("pending-cleanup obligation owned by the cleanup queue", () => {
+  let db!: ReturnType<typeof createDb>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  let companyId = "";
+
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("paperclip-pending-cleanup-queue-owner-");
+    db = createDb(temporary.connectionString);
+    companyId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "PCQ",
+      requireBoardApprovalForNewAgents: false });
+  }, 30_000);
+
+  afterEach(async () => {
+    await db.delete(environmentLeases);
+  });
+
+  afterAll(async () => {
+    await temporary?.cleanup();
+  });
+
+  async function item(updatedAt: Date, metadata: Record<string, unknown> = {}) {
+    const id = randomUUID();
+    await db.insert(environmentLeases).values({ id, companyId, status: "pending_cleanup", cleanupStatus: "failed",
+      provider: "fake", providerLeaseId: `sandbox://fake/${id}`, updatedAt, metadata });
+    return id;
+  }
+
+  async function pass() {
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    return { kind: report.kinds.find(entry => entry.kind === KIND)!,
+      violations: report.violations.filter(violation => violation.kind === KIND) };
+  }
+
+  it("names the cleanup queue as owner while the item's retry deadline is current, so I1 reports no gap", async () => {
+    await item(new Date());
+    await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 5,
+      pendingCleanupRetryAfterMs: Date.now() + 20 * 60_000 });
+    const { kind, violations } = await pass();
+    expect(kind).toMatchObject({ inProgress: 2, verdicts: { alive: 2 }, reasons: { queue_scheduled: 2 } });
+    expect(violations).toEqual([]);
+  });
+
+  it("reports an item the queue has not acted on long past its deadline as a visible-next-action gap", async () => {
+    const overdue = await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 1 });
+    const { kind, violations } = await pass();
+    expect(kind).toMatchObject({ verdicts: { unprovable: 1 }, reasons: { queue_overdue: 1 } });
+    expect(violations).toEqual([expect.objectContaining({ rule: "I7_visible_next_action", recordId: overdue,
+      reason: "queue_overdue" })]);
+  });
+
+  it("never reports a settled orphan's first attempt as overdue just because the worker's deadline is 'now'", async () => {
+    // The orphaned-run-lease settle keeps updatedAt and saves when it queued the item.
+    await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 0,
+      pendingCleanupRecoveredAtMs: Date.now() });
+    await db.update(environmentLeases).set({ failureReason: "orphaned_active_lease_recovered" });
+    const { kind, violations } = await pass();
+    expect(kind).toMatchObject({ verdicts: { alive: 1 }, reasons: { queue_scheduled: 1 } });
+    expect(violations).toEqual([]);
+  });
+
+  it("wakes the reconciler when a scheduled item would become overdue", async () => {
+    const updatedAt = new Date(Date.now() - 60_000);
+    await item(updatedAt);
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    // Due at updatedAt + one retry interval; overdue after two more.
+    const expected = updatedAt.getTime() + 5 * 60_000 + 10 * 60_000;
+    expect(Math.abs((report.nextDueAt?.getTime() ?? 0) - expected)).toBeLessThan(5_000);
+  });
+
+  it("keeps an escalated item with the sweep's 30-minute cooldown scheduled, due at its saved deadline", async () => {
+    // The sweep escalates well after boot: let the clock pass the reader's 1 s slack.
+    await new Promise(resolve => setTimeout(resolve, 1_500));
+    const retryAfterMs = Date.now() + 30 * 60_000;
+    await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 5,
+      pendingCleanupRetryAfterMs: retryAfterMs });
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(report.kinds.find(entry => entry.kind === KIND)).toMatchObject({ verdicts: { alive: 1 },
+      reasons: { queue_scheduled: 1 } });
+    expect(report.violations.filter(violation => violation.kind === KIND)).toEqual([]);
+    expect(Math.abs((report.nextDueAt?.getTime() ?? 0) - (retryAfterMs + 10 * 60_000))).toBeLessThan(5_000);
+  });
+
+  it("does not list an item a claimant holds in flight (that is the teardown claim)", async () => {
+    await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupInFlight: true, pendingCleanupAttemptId: "claim" });
+    const { kind } = await pass();
+    expect(kind).toMatchObject({ inProgress: 0 });
+  });
+});

server/src/__tests__/environment-run-lease-ownership.test.ts

@@ -117,6 +117,7 @@ suite("environment_run_lease_active on the shared ownership path", () => {
     expect(report.kinds.find(kind => kind.kind === "environment_run_lease_active")).toMatchObject({
       settled: 1, reasons: { pid_dead: 1 } });
     expect(await row(leaseId)).toMatchObject({ status: "pending_cleanup", failureReason: "orphaned_active_lease_recovered" });
+    expect(typeof (await row(leaseId)).metadata?.pendingCleanupRecoveredAtMs).toBe("number");
   });
 
   it("stops and keeps a dead holder's sandbox instead of destroying it", async () => {

server/src/__tests__/ownership-reconciliation-report.test.ts

@@ -158,15 +158,16 @@ describe("report-only ownership reconciliation", () => {
       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 } });
+    // The cleanup queue owns an unclaimed item while its retry deadline is current.
     expect(kindReport(first, "environment_pending_cleanup")).toMatchObject({ inProgress: 1,
-      verdicts: { unprovable: 1 }, reasons: { no_owner_stamp: 1 } });
+      verdicts: { alive: 1 }, reasons: { queue_scheduled: 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"]);
+    expect(rules(first, "environment_pending_cleanup")).toEqual([]);
     // I2 needs the owner proven gone on two consecutive passes.
     expect(rules(first, "environment_run_lease_active")).toEqual([]);
     // Phase 3: pending accounting is exposure, not a gate. Only a refusal that

server/src/services/environment-run-lease-settlement.ts

@@ -75,7 +75,10 @@ export async function settleEnvironmentRunLease(db: Db, record: InProgressRecord
     // Keep the row's updatedAt, so the pending_cleanup sweep accepts it in the
     // same tick instead of waiting out its own retry backoff.
     const flipped = await writeEnvironmentCleanup(tx, write => write.update(environmentLeases)
-      .set({ status: "pending_cleanup", failureReason: ORPHANED_RUN_LEASE_FAILURE_REASON })
+      .set({ status: "pending_cleanup", failureReason: ORPHANED_RUN_LEASE_FAILURE_REASON,
+        // When the first cleanup attempt became due (the row keeps updatedAt).
+        metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({
+          pendingCleanupRecoveredAtMs: context.now.getTime() })}::jsonb` })
       .where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.status, "active"),
         eq(environmentLeases.heartbeatRunId, runId),
         sql`${environmentLeases.metadata} is not distinct from ${lease.metadata === null ? null : JSON.stringify(lease.metadata)}::jsonb`))

server/src/services/recovery/ownership-invariant-report.ts

@@ -165,6 +165,10 @@ async function classify(owner: RecordOwner, runProofs: Map<string, Classificatio
     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" };
+    // A queue is not a process: it is never proven gone. A deadline long past
+    // means the queue is not acting, which is a gap to report, not a release.
+    case "queue": return now.getTime() - owner.dueAt.getTime() > owner.graceMs
+      ? { verdict: "unprovable", reason: "queue_overdue" } : { verdict: "alive", reason: "queue_scheduled" };
     case "none": return { verdict: "unprovable", reason: "no_owner_stamp" };
   }
 }
@@ -462,6 +466,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
   const kindReports: KindReport[] = [];
   const violations: InvariantViolation[] = [];
   const goneThisPass = new Set<string>();
+  let queueDueAt: Date | null = null;
   const settleRecords = options.settle ?? options.record !== false;
   let retryNeeded = false;
 
@@ -501,6 +506,16 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
       if (kind.invariantChecks.includes(1) && record.owner.kind === "none") {
         violations.push({ ...base, rule: "I1_owned", reason: "no_owner_stamp" });
       }
+      if (kind.invariantChecks.includes(7) && record.owner.kind === "queue") {
+        if (proof.reason === "queue_overdue") {
+          violations.push({ ...base, rule: "I7_visible_next_action", reason: "queue_overdue" });
+        } else {
+          // Re-check when this item would become overdue, so a queue that stops
+          // acting without another write is still reported.
+          const overdueAt = new Date(record.owner.dueAt.getTime() + record.owner.graceMs + 1);
+          if (!queueDueAt || overdueAt < queueDueAt) queueDueAt = overdueAt;
+        }
+      }
       if (settleRecords && kind.settle) {
         const outcome = await settleOrSurface(db, kind, record, proof, now);
         if (outcome === "settled") { report.settled += 1; continue; }
@@ -517,7 +532,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
     }
   }
 
-  let nextDueAt: Date | null = null;
+  let nextDueAt: Date | null = queueDueAt;
   for (const check of [
     () => checkAdmissionIndependence(db, state, now, limit),
     () => checkExactlyOnceIntents(db, state, now, limit),

server/src/services/recovery/ownership-registry.ts

@@ -42,6 +42,11 @@ import {
 } from "../native-runtime/native-token-lease-settlement.js";
 import { ephemeralServerEnabled } from "../ephemeral-server-mode.js";
 import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../environment-lease-controller.js";
+import {
+  ENVIRONMENT_CLEANUP_RETRY_MS,
+  pendingCleanupDeadlineSql,
+  recoveredOrphanFirstAttemptSql,
+} from "../environment-cleanup-state.js";
 import { settleEnvironmentRunLease } from "../environment-run-lease-settlement.js";
 import { settleEnvironmentTeardownClaim } from "../environment-teardown-claim-settlement.js";
 
@@ -71,6 +76,10 @@ export type RecordOwner =
   | { kind: "run"; runId: string }
   /** A person or agent assignment, such as an operator-owned recovery action. */
   | { kind: "operator"; ownerType: string | null }
+  /** A durable work queue that acts on the record by its saved deadline. The
+   * queue owns the record while that deadline is current; a deadline long past
+   * is a gap (the queue is not acting), never a release. */
+  | { kind: "queue"; queue: string; dueAt: Date; graceMs: number }
   /** Nothing on the record names an owner (invariant 1). */
   | { kind: "none" };
 
@@ -362,6 +371,50 @@ function teardownClaimOwner(row: LeaseRow): RecordOwner {
   };
 }
 
+/** A cleanup worker pass that misses a due item by more than this is a gap. */
+const PENDING_CLEANUP_OVERDUE_GRACE_MS = 2 * ENVIRONMENT_CLEANUP_RETRY_MS;
+
+/**
+ * Pending-cleanup obligation (not in flight). An unclaimed queue item names no
+ * process by design: the environment cleanup queue owns it, and its next
+ * action is the queue's next attempt, due at the saved retry deadline (the
+ * same deadline the cleanup worker sleeps until). While that deadline is
+ * current the queue owns the item; once it is long past, the item is reported
+ * as a gap (I7), never released. Invariant 1 then measures only real gaps.
+ */
+function pendingCleanupKind(): OwnableKind {
+  const where = and(eq(environmentLeases.status, "pending_cleanup"),
+    sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`);
+  // The first attempt of a settled orphan is due at once: the worker's "run
+  // now" deadline is 0, so its due time is when the settle queued it (saved
+  // by the settle; a row from before that is due since its last write).
+  // Built per listing: the deadline reader bounds saved deadlines relative to
+  // the current time, so it must not be frozen at module load.
+  const dueMs = () => sql<string>`case when ${recoveredOrphanFirstAttemptSql()}
+      then coalesce(case when jsonb_typeof(${environmentLeases.metadata}->'pendingCleanupRecoveredAtMs') = 'number'
+        then (${environmentLeases.metadata}->>'pendingCleanupRecoveredAtMs')::numeric end,
+        extract(epoch from ${environmentLeases.updatedAt}) * 1000)
+    else greatest(extract(epoch from ${environmentLeases.updatedAt}) * 1000 + ${ENVIRONMENT_CLEANUP_RETRY_MS},
+      ${pendingCleanupDeadlineSql()}) end`;
+  return {
+    id: "environment_pending_cleanup", title: "Pending-cleanup obligation (not in flight)", obligation: "ownership",
+    storage: "database", settle: null, invariantChecks: [1, 2, 7],
+    listInProgress: async (db, limit, offset) => {
+      const rows = await db.select({ id: environmentLeases.id, companyId: environmentLeases.companyId,
+        heartbeatRunId: environmentLeases.heartbeatRunId, issueId: environmentLeases.issueId,
+        updatedAt: environmentLeases.updatedAt, dueMs: dueMs() })
+        .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: { kind: "queue", queue: "environment_cleanup", dueAt: new Date(Number(row.dueMs)),
+          graceMs: PENDING_CLEANUP_OVERDUE_GRACE_MS },
+        updatedAt: row.updatedAt,
+      })) };
+    },
+  };
+}
+
 /** `ControllerProcess` metadata ({bootId, pid, hostHash, processStartedAt}). */
 function controllerStamp(value: unknown, scope: string): RecordOwner {
   const controller = readControllerProcess(value);
@@ -604,10 +657,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [
       teardownClaimOwner),
     settle: settleEnvironmentTeardownClaim,
   },
-  leaseKind("environment_pending_cleanup", "Pending-cleanup obligation (not in flight)", "ownership",
-    and(eq(environmentLeases.status, "pending_cleanup"),
-      sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`),
-    () => ({ kind: "none" }), [1, 2, 7]),
+  pendingCleanupKind(),
   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)),