fix(server): settle empty environment cleanup notifications (#19316)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Environment workers retain cleanup duties and retry them safely.
> - A metadata change can notify the worker without adding cleanup debt.
> - The pending-cleanup worker kept a work hold during its thirty-second
spacing window even when its owner reported no debt.
> - Historical cleanup already checks that condition without advancing a
provider sweep.
> - This PR applies the same owner check to pending cleanup and
preserves later notifications.
> - An empty notification can settle while real cleanup and uncertain
work still block sleep.

## Linked Issues or Issue Description

Refs #18646. That change covers empty historical-cleanup notifications.
This follow-up covers the pending-cleanup queue. Related runtime work:
#18988.

A search of open and closed PRs and issues found no duplicate
pending-queue fix. This repairs an existing worker under the roadmap's
Cloud deployment work.

**What happened?**

A normal metadata update on a released lease left one tracked work
intent and a scheduled pass. The actual cleanup deadline was null before
and after the update. The worker kept the hold until its spacing window
ended.

**Expected behavior**

The worker can clear an empty notification through its existing owner
predicate after a normal sweep and deadline read complete. It must
retain new debt, unresolved transactions, failed sweeps, failed reads
and later notifications. Provider sweep spacing and cleanup retry
deadlines must stay unchanged.

**Steps to reproduce**

1. Create and release a fixture cleanup lease through the normal
services.
2. Start the registered worker and confirm that its cleanup deadline is
null.
3. Change the released lease's metadata before the thirty-second spacing
window ends.
4. Run the new test. The original worker keeps an active intent even
though eligibility remains empty.

**Paperclip version or commit**

Candidate: `69aaee95cd73322a37e0270354d4a06bf5512ba7`. It normally
merges actual master `cd3832d2357eab481a861331ba1bc944a0b951f4`. The
focused retry repair follows the original two-file fix. Every other tree
blob still matches that master. The diff has two files.

**Deployment mode**

Source checkout with external PostgreSQL 17 and controlled coordinator
time. No deployed host or database suspension result is claimed.

## What Changed

- Apply the existing cheap deadline check to both registered cleanup
queues during provider spacing.
- Clear only the captured notification generation when the owner returns
no deadline and the latest normal sweep completed successfully.
- Invalidate that coverage before every full sweep. Startup and later
sweep failures keep their existing retry even if a metadata notification
has no database deadline.
- Add real PostgreSQL controls for empty metadata changes, concurrent
debt, unresolved transactions, failed reads through drain/resume, and
unread-spool recovery after startup and later sweep failures.
- Retain the original provider sweep, retry, deferred-buffer, creation,
startup, historical and shutdown controls.

## Verification

**Real PostgreSQL red/green qualification passed.** The original
empty-notification counterfactual failed at active work one, expected
zero, after actual null eligibility reads (16.706s). Review then found a
missing failed-sweep combination. Two additional cases using the real
malformed spool and ordinary metadata notifications reproduced lost
retries on the earlier published head, both at the intended pending
predicate (42.144s). One case fails at startup; the other first
completes a successful sweep, then fails a later sweep. Failed fixtures
and original logs remain retained.

At measured source `1ccd227c51e03ab4306d7cb85cf81efb0e3d3678`, all **29
owner cases passed in 113.061s**, without filters or skips. The tests
repair the actual spool, recover the ordinary cleanup lease, perform
provider cleanup once, preserve the later-sweep backoff, and return to
quiet. Normal migrations used a fresh owned external database; each pool
was capped at four with one worker. The first test-fixture attempt
accidentally replaced the real spool driver; that setup failure is
preserved separately and was corrected before the counterfactual. No
final cleanup result or receipt was seeded.

Run the full owner suite against a fresh, owned external test database
with normal migrations applied:

```sh
pnpm --filter @paperclipai/server exec vitest run src/__tests__/environment-cleanup-work.test.ts --maxWorkers=1 --no-file-parallelism
```

Set `PAPERCLIP_ENVIRONMENT_CLEANUP_TEST_DATABASE_URL` to that fixture.
Local qualification uses one worker and a pool of four.

Current native TypeScript checks passed: normal production declaration
layers (44.127s) and the affected test entry using the repository's
actual CI options and references (3.922s). Own dependency builds passed.
A frozen install first found stale upstream ACPX patch metadata;
normally carrying #19294 corrected the hash references for the same
package version. No lock policy was relaxed.

Full local workspace typecheck, tests and build were not run in this
bounded external-PG lease. Fresh checks on the current head
`69aaee95cd73322a37e0270354d4a06bf5512ba7` completed with **53
successes, one neutral Pony review and two skips**, with no pending or
failed gates. Its [PR
workflow](https://github.com/paperclipai/paperclip/actions/runs/42104787089),
aggregate verification and commitperclip review succeeded. The complete
latest edited Greptile and Pony summaries both gave **5/5**. Both review
threads are resolved, and all root and nested thread pages were
inspected.

The current head only removes four changed code-comment lines to follow
CONTRIBUTING.md. Its emitted JavaScript is byte-identical to the
measured parent `1ccd227c51e03ab4306d7cb85cf81efb0e3d3678`; tests,
configuration and dependencies are unchanged. The local 29-case evidence
therefore carries to this head without a duplicate local run. The remote
results above are fresh for the current head.
On the earlier head, Runner CI failed an omitted-context
checkpoint-settlement case. The fresh Runner job on this head passed
without a rerun. The original failure logs are preserved and its cause
remains unproven; current success is recovery evidence, not a causal
explanation.

The recorded test and migration command PIDs were reaped; an independent
audit verifies zero owned database clients before normal cleanup. Failed
fixtures remain retained. The process audit covers recorded command
PIDs, not an exhaustive history of child processes.

The long quiet assertion uses controlled coordinator time and actual SQL
counters. It does not prove a real twenty-minute quiet window, provider
behavior or deployed suspension.

## Risks

The empty read must not erase a notification that arrives while it runs.
The captured generation and transaction controls preserve that boundary.
An unknown or failed read keeps work pending. Empty deadline rows cannot
cover a failed sweep or an unread local spool; only a successful normal
sweep and deadline read establish that coverage. No cleanup row,
resource claim, provider action, authorization rule, sweep spacing or
retry deadline is removed.

## Model Used

OpenAI `gpt-6.1-sol` produced the delegated regression tests and source
review. OpenAI Codex, based on GPT-6, implemented the small worker
change and managed integration and verification. Both used repository
tools and code execution. Exact served backend revisions and context
limits were not exposed.

## 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 linked an existing public issue or described the
issue using the issue template
- [x] I have not referenced internal or instance-local Paperclip issues
or links
- [x] My branch name describes the change and contains no internal issue
ID
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have checked that no documentation change is needed for this
internal worker fix
- [x] I have considered and documented risks
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open findings
- [x] I will address all reviewer comments before requesting merge

server/src/__tests__/environment-cleanup-work.test.ts

@@ -5,6 +5,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { createSandboxOrphanCleanupSpool } from "../services/sandbox-orphan-cleanup-spool.js";
 import { nextEnvironmentCleanupAt } from "../services/environment-cleanup-state.js";
+import * as environmentCleanupState from "../services/environment-cleanup-state.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 import { randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -12,6 +13,7 @@ import { eq, sql } from "drizzle-orm";
 import { companies, createDb, sandboxCreationOperations, environmentLeases, environments, type Db } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { registerEnvironmentCleanupWork } from "../services/environment-cleanup-work.js";
 import { registerDeferredEnvironmentCleanup } from "../services/environment-cleanup-notifications.js";
 import { environmentService } from "../services/environments.js";
@@ -28,7 +30,9 @@ describe.skipIf(!support.supported)("environment cleanup scheduling", () => {
   let companyId: string, environmentId: string, draining: boolean, failing: boolean;
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   let worker: ReturnType<typeof registerEnvironmentCleanupWork> | undefined;
-  let releases: ReturnType<typeof vi.fn>, reads: ReturnType<typeof vi.spyOn>;
+  let releases: ReturnType<typeof vi.fn<() => Promise<void>>>;
+  let reads: ReturnType<typeof vi.spyOn>;
+  let pendingSweeps: ReturnType<typeof vi.fn<(backoffMs: number) => Promise<unknown>>>;
   let inspections: ReturnType<typeof vi.fn<(row: { providerLeaseId: string | null }) => Promise<{ providerLeaseId: string | null; state: string }>>>;
   const errors: unknown[] = [], disposers: Array<() => void> = [];
   beforeAll(async () => {
@@ -56,15 +60,27 @@ describe.skipIf(!support.supported)("environment cleanup scheduling", () => {
     expect(errors).toEqual([]);
   });
   afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
-  async function start(historical = false, sweepOverride?: (backoffMs: number) => Promise<unknown>) {
+  async function start(historical = false, sweepOverride?: (backoffMs: number) => Promise<unknown>, observePendingCheck?: (pending: boolean) => void) {
     const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", retryPendingSandboxTeardown: releases } as any] });
     const heartbeat = heartbeatService(db, { environmentRuntime: runtime });
     coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
       onError: error => errors.push(error) });
+    if (observePendingCheck) {
+      const register = coordinator.register.bind(coordinator);
+      vi.spyOn(coordinator, "register").mockImplementation((queue, task) => register(queue, queue !== DELIVERY_QUEUES.environmentCleanup ? task : {
+        ...task,
+        hasPending() {
+          const result = task.hasPending();
+          void result.then(value => { observePendingCheck(value); }, () => {});
+          return result;
+        },
+      }));
+    }
     const history = historical && historicalCleanupVerificationService(db, { inspectLease: inspections, inspectSetup: inspections });
     let historySettled = !history;
+    pendingSweeps = vi.fn(sweepOverride ?? (backoffMs => heartbeat.sweepPendingCleanupLeases({ backoffMs })));
     worker = registerEnvironmentCleanupWork(db, coordinator, {
-      sweepPending: sweepOverride ?? (backoffMs => heartbeat.sweepPendingCleanupLeases({ backoffMs })),
+      sweepPending: pendingSweeps,
       historical: history ? { hasPending: () => history.hasPending(),
         sweep: async () => { const result = await history.sweep(); historySettled = true; return result; } } : undefined,
       canRun: () => !draining,
@@ -146,6 +162,99 @@ describe.skipIf(!support.supported)("environment cleanup scheduling", () => {
       .toEqual({ version: 1, backgroundWork: "present" });
     await quiet();
   });
+  it("discharges an empty pending cleanup metadata notification during provider spacing", async () => {
+    const lease = await pending();
+    await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" });
+    const startedAt = Date.now();
+    await start();
+    expect(await nextEnvironmentCleanupAt(db)).toBeNull();
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(coordinator!.nextWakeAt()).toBeNull();
+    expect(pendingSweeps).toHaveBeenCalledTimes(1);
+    expect(Date.now() - startedAt).toBeLessThan(30_000);
+    await environmentService(db).updateLeaseMetadata(lease.id, { driver: "sandbox", presentation: "changed" });
+    expect(await nextEnvironmentCleanupAt(db)).toBeNull();
+    await vi.waitFor(() => {
+      expect(idleWorkSnapshot().active).toBe(0);
+      expect(coordinator!.nextWakeAt()).toBeNull();
+    });
+    expect(Date.now() - startedAt).toBeLessThan(30_000);
+    expect(pendingSweeps).toHaveBeenCalledTimes(1);
+    expect(releases).not.toHaveBeenCalled();
+    const executions = vi.spyOn(db, "execute");
+    const count = reads.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000);
+    expect(reads).toHaveBeenCalledTimes(count);
+    expect(executions).not.toHaveBeenCalled();
+    expect(pendingSweeps).toHaveBeenCalledTimes(1);
+    expect(releases).not.toHaveBeenCalled();
+  });
+  it.each([false, true])("preserves pending cleanup debt arriving during an empty fast-path read (unresolved transaction=%s)", async unresolved => {
+    const passive = await pending();
+    await environmentService(db).releaseLease(passive.id, "released", { cleanupStatus: "success" });
+    await start();
+    expect(await nextEnvironmentCleanupAt(db)).toBeNull();
+    let readStarted = false, releaseRead!: () => void, writeReady!: () => void, releaseWrite!: () => void;
+    const readGate = new Promise<void>(resolve => { releaseRead = resolve; });
+    const written = new Promise<void>(resolve => { writeReady = resolve; });
+    const writeGate = new Promise<void>(resolve => { releaseWrite = resolve; });
+    const original = environmentCleanupState.nextEnvironmentCleanupAt;
+    const check = vi.spyOn(environmentCleanupState, "nextEnvironmentCleanupAt").mockImplementationOnce(async (...args) => {
+      const at = await original(...args); expect(at).toBeNull(); readStarted = true; await readGate; return at;
+    });
+    let writing: Promise<unknown> | undefined, debtId: string | undefined;
+    try {
+      await environmentService(db).updateLeaseMetadata(passive.id, { presentation: "changed" });
+      await vi.waitFor(() => expect(readStarted).toBe(true));
+      writing = db.transaction(async tx => {
+        debtId = (await pending(tx as unknown as Db)).id;
+        writeReady(); if (unresolved) await writeGate;
+      });
+      await written;
+      if (!unresolved) await writing;
+      releaseRead();
+      await vi.advanceTimersByTimeAsync(1);
+      expect(idleWorkSnapshot().active).toBeGreaterThan(0);
+      expect(pendingSweeps).toHaveBeenCalledTimes(1);
+      expect(releases).not.toHaveBeenCalled();
+      releaseWrite(); await writing;
+      await vi.waitFor(async () => expect(await original(db)).not.toBeNull());
+      expect(idleWorkSnapshot().active).toBeGreaterThan(0);
+      const ownerId = randomUUID();
+      const hold = { ownerId, draining: true, startedAt: new Date(Date.now() - 1000),
+        expiresAt: new Date(Date.now() + 60_000), activeRuns: 0, pendingWakes: 0 };
+      expect((await readIdleSleepSafety(db, () => hold, Date.now, ownerId, async () => "none")).backgroundWork).toBe("present");
+      await settle(300_000);
+      await vi.waitFor(() => { expect(releases).toHaveBeenCalledTimes(1); expect(idleWorkSnapshot().active).toBe(0); });
+      expect((await environmentService(db).getLeaseById(debtId!))?.status).toBe("expired");
+      expect(await original(db)).toBeNull(); await quiet();
+    } finally { releaseRead(); releaseWrite(); await writing; check.mockRestore(); }
+  });
+  it("retains a failed pending cleanup fast-path read through drain and clears it after resume", async () => {
+    const lease = await pending();
+    await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" });
+    await start();
+    expect(await nextEnvironmentCleanupAt(db)).toBeNull();
+    const original = environmentCleanupState.nextEnvironmentCleanupAt;
+    vi.spyOn(environmentCleanupState, "nextEnvironmentCleanupAt").mockImplementationOnce(async (...args) => {
+      expect(await original(...args)).toBeNull();
+      throw new Error("pending cleanup deadline read unavailable");
+    });
+    await environmentService(db).updateLeaseMetadata(lease.id, { presentation: "changed" });
+    await vi.waitFor(() => expect(errors).toHaveLength(1));
+    expect(idleWorkSnapshot().active).toBeGreaterThan(0); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    expect(pendingSweeps).toHaveBeenCalledTimes(1); expect(releases).not.toHaveBeenCalled();
+    errors.length = 0; draining = true;
+    const count = reads.mock.calls.length;
+    await settle(60_000, false);
+    expect(reads).toHaveBeenCalledTimes(count); expect(idleWorkSnapshot().active).toBeGreaterThan(0);
+    expect(pendingSweeps).toHaveBeenCalledTimes(1);
+    draining = false; await settle();
+    await vi.waitFor(() => { expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); });
+    expect(releases).not.toHaveBeenCalled(); await quiet();
+    worker!.stop(); await coordinator!.stop();
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
   it("discharges an empty historical metadata notification during provider spacing", async () => {
     const lease = await pending();
     await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" });
@@ -341,6 +450,63 @@ describe.skipIf(!support.supported)("environment cleanup scheduling", () => {
       await quiet();
     } finally { await rm(dir, { recursive: true, force: true }); }
   });
+  it.each(["startup", "later"] as const)("retains a failed %s spool sweep through empty metadata during spacing", async phase => {
+    const lease = await pending();
+    await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" });
+    const dir = await mkdtemp(join(tmpdir(), "cleanup-failed-sweep-"));
+    const unread = join(dir, "unread.json");
+    const spool = createSandboxOrphanCleanupSpool(dir);
+    const checks: boolean[] = [];
+    const providerLeaseId = randomUUID();
+    try {
+      if (phase === "startup") await writeFile(unread, "{ truncated");
+      const runtime = environmentRuntimeService(db, { orphanCleanupSpool: spool });
+      vi.spyOn(runtime, "retryPendingSandboxTeardown").mockImplementation(async () => releases());
+      const heartbeat = heartbeatService(db, { environmentRuntime: runtime });
+      await start(false, backoffMs => heartbeat.sweepPendingCleanupLeases({ backoffMs }), value => { checks.push(value); });
+      if (phase === "later") {
+        expect(errors).toEqual([]); expect(pendingSweeps).toHaveBeenCalledTimes(1);
+        expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
+        await vi.advanceTimersByTimeAsync(30_000);
+        await writeFile(unread, "{ truncated");
+        await environmentService(db).updateLeaseMetadata(lease.id, { presentation: "request later sweep" });
+      }
+      await vi.waitFor(() => expect(errors).toHaveLength(1));
+      const sweeps = phase === "startup" ? 1 : 2;
+      expect(pendingSweeps).toHaveBeenCalledTimes(sweeps);
+      expect(await nextEnvironmentCleanupAt(db)).toBeNull();
+      expect(coordinator!.nextWakeAt()).not.toBeNull();
+      expect(releases).not.toHaveBeenCalled();
+      errors.length = 0;
+      const failedAt = Date.now();
+      const before = checks.length;
+      await environmentService(db).updateLeaseMetadata(lease.id, { presentation: "empty hint after failed sweep" });
+      expect(await nextEnvironmentCleanupAt(db)).toBeNull();
+      await vi.waitFor(() => expect(checks.length).toBeGreaterThan(before));
+      expect(checks.at(-1)).toBe(true);
+      expect(idleWorkSnapshot().active).toBeGreaterThan(0);
+      expect(coordinator!.nextWakeAt()).not.toBeNull();
+      expect(Date.now() - failedAt).toBeLessThan(30_000);
+      expect(pendingSweeps).toHaveBeenCalledTimes(sweeps);
+      expect(await spool.append({ companyId, environmentId, executionWorkspaceId: null, issueId: null,
+        heartbeatRunId: null, provider: "fixture", providerLeaseId, metadata: { driver: "sandbox" } })).toBe(true);
+      await rm(unread);
+      await settle(30_000);
+      await vi.waitFor(async () => {
+        const rows = await db.select().from(environmentLeases).where(eq(environmentLeases.providerLeaseId, providerLeaseId));
+        expect(rows).toHaveLength(1);
+      });
+      if (phase === "later") {
+        expect(releases).not.toHaveBeenCalled();
+        await settle(300_000);
+      }
+      await vi.waitFor(() => expect(releases).toHaveBeenCalledTimes(1));
+      const [recovered] = await db.select().from(environmentLeases).where(eq(environmentLeases.providerLeaseId, providerLeaseId));
+      expect(recovered.status).toBe("expired");
+      expect(await spool.loadWithStatus!()).toEqual({ records: [], readFailed: false });
+      await quiet();
+    } finally { await rm(dir, { recursive: true, force: true }); }
+  });
   it("does not delay startup for historical provider checks but joins them on shutdown", async () => {
     let release!: () => void;
     const gate = new Promise<void>(resolve => { release = resolve; });

server/src/services/environment-cleanup-work.ts

@@ -17,6 +17,7 @@ export function registerEnvironmentCleanupWork(
   const spacing = input.minimumSpacingMs ?? 30_000;
   function register(queue: DeliveryQueue, run: () => Promise<unknown>, deadline: () => Promise<number | null>, minimumSpacing = spacing) {
     let requested = 1, reconciled = 0, next: number | null = null, earliest = 0;
+    let sweepComplete = false;
     const worker = coordinator.register(queue, {
       retryMs: spacing, pausePassiveMaintenance: true,
       onSettled: () => { requested++; },
@@ -24,21 +25,18 @@ export function registerEnvironmentCleanupWork(
         if (!input.canRun()) { requested++; next = null; return; }
         if (Date.now() < earliest) {
           next = earliest;
-          if (queue === DELIVERY_QUEUES.historicalCleanup) {
-            // A settled metadata invalidation is not itself cleanup debt. A
-            // cheap owner read can discharge it without advancing the provider
-            // sweep cadence; later notifications keep their own generation.
-            const generation = requested;
-            if (await deadline() === null) { next = null; reconciled = generation; }
-          }
+          const generation = requested;
+          if (sweepComplete && await deadline() === null) { next = null; reconciled = generation; }
           return;
         }
         const generation = ++requested;
         earliest = Date.now() + minimumSpacing;
+        sweepComplete = false;
         await run();
         const due = await deadline();
         next = due === null ? null : Math.max(due, earliest);
         reconciled = generation;
+        sweepComplete = true;
       },
       hasPending: async () => requested !== reconciled,
       nextRunAt: () => next,