fix(server): retire owned plugin listeners before idle sleep (#19358)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Plugins receive durable agent lifecycle events through a registered
host.
> - The host uses a PostgreSQL LISTEN connection to wake delivery.
> - An idle report could return `none` while that host-owned connection
stayed open.
> - This change gives that listener an exclusive owner and closes it
during the current idle hold.
> - Release or expiry restores listening before journal replay.
> - Failed close and pending work continue to block sleep.

## Linked Issues or Issue Description

Refs #18988, which is now merged. This focused ten-file change now
targets master. Current integrated candidate:
`0f83005122ef1bdb1f1e08f8f58424c9a840ebb1`, with actual master
`644ee66f5d4ae8365fe8b9eca57df2be7405b5ab` preserved through normal
merges. No history was rewritten.

**What happened?**

A normal local App with an installed lifecycle plugin could return an
idle `none` report while its lifecycle LISTEN backend and socket stayed
open.

**Expected behavior**

The host must close its own listener before it reports idle safety. It
must retain other subscriptions, durable events, company scope, retry
deadlines and useful reactivation.

**Steps to reproduce**

Start the normal server with a lifecycle plugin. Create an agent and
wait for its durable acknowledgment. Stop the server, create another
agent while it is stopped, and restart with the same files. Request an
owned idle hold after ordinary replay. Check the exact listener backend
and matched Node socket before shutdown. Let the valid hold expire
naturally, then make an ordinary authorized agent update and check its
durable plugin acknowledgment.

Live executing or persistent preview services remain active. The
separate hosted preview dependency
[paperclip-cloud#478](https://github.com/paperclipai/paperclip-cloud/pull/478)
retains its own App and operator prerequisites. This PR does not claim
deployed preview or suspension acceptance.

## What Changed

- Add a private, exclusive database listener owner that inherits the
normal client policy.
- Prepare plugin lifecycle retirement after native retirement under the
same current idle hold.
- Use one delivery-eligibility helper for dispatch and idle journal
inspection. It preserves the same manifest, capability, company-disable
query, and skip rules.
- Close only the owned listener. Keep failed close ownership and
inspection failures visible.
- Resume listening on normal release or expiry, before journal replay.
- Preserve disabled-feature retry history and the original deadline on
re-enable.
- Await lifecycle disposal during normal shutdown and consumer teardown.
- Add owner regressions and document the lifecycle contract.

## Verification

**Current integrated source:**
`0f83005122ef1bdb1f1e08f8f58424c9a840ebb1`. Fresh normal Runner
TypeScript passed in 0.785 seconds. The complete normal App layered
semantic check passed in 28.284 seconds. All tracked source, the
lockfile and the 18 separate private witness inputs stayed unchanged.
Known compiler owners closed and the isolated temporary home was
removed.

**Current product qualification:** at measured integration
`896cc9b62993e6ed10418d7a7417e1ddbf8dc89c`, all three whole files passed
with no failures or skips: lifecycle delivery 40, runtime hooks 20, and
runtime assignments 24. The actual physical LISTEN and natural-expiry
controls passed. All owned clients and captured processes closed.
Independent final session checks were zero before normal deletion of
each successful test database.

**Normal full-App physical retirement:** the original case passed in
47.574 seconds. It exercised real startup, readiness, SDK callbacks,
durable ACK, foreign-company exclusion, physical stop, offline input and
cold replay. The initial unknown hold and its ordinary release remain in
the evidence. After the original 30.5-second cadence, a new valid hold
returned quiescent. The exact current cold-App LISTEN backend and its
matched Node socket closed before server stop.

**Normal full-App natural expiry:** the separate original case passed in
77.391 seconds under its original 90-second body cap. The final valid
30-second hold expired naturally. A new listener backend and matched
current App socket appeared before fresh input. An ordinary authorized
agent-name PATCH produced journal event 20, the actual SDK callback and
durable ACK. Foreign exclusion and readiness assertions passed. The
final measured hold had no manual release, replacement hold or shortened
TTL. The original preparatory unknown hold, ordinary release and
30.5-second cadence remain explicit.

Both App cases used the normal current server helpers and actual SDK
bundling/worker entry. A private compatibility change supplied the same
owned HOME at the exact existing child-fork boundary. It did not change
the product helpers, business assertions or timing limits. These
separate acceptance fixtures remain outside the focused product diff.

**Source carry to the current candidate:** all nine other owned files,
listener mechanics, actual producer paths, SDK paths and measured hold
behavior are unchanged. The sole owned intersection from newer master is
an independent environment-delete cleanup callback in `app.ts`. The
plugin witnesses do not call that route. Normal merges preserve every
unrelated master blob and file mode. The fresh current semantic checks
above qualify the integrated source. New master Native crash-recovery
and saved-copyback Retry capabilities are not claimed by these plugin
cases.

**Actual master merge target:** newer master contains retained plugin
Runner release/resume and other Native recovery fixes. The projected
combined target passed fresh normal Runner TypeScript in 0.799 seconds
and the full App semantic check in 29.715 seconds. All ten plugin-owned
blobs and modes remain exact to the published head; every unrelated
master change is preserved. The measured plugin cases create no Native
provider turn or warm Native session. The unchanged
native-first/plugin-second ordering still checks the native result
before listener retirement. Later sparse-checkpoint and Native
response-publication changes do not enter those paths. This is a scoped
source carry, not new hosted Runner acceptance.

**Failures retained:** an earlier reactivation fixture's real timer
returned one millisecond before its original wall-clock expiry. Its
private fixture now waits any positive remainder and retains the same
expiry assertion and limits. The earlier partial-composition `unknown`
result remains distinct from normal-startup proof. Current qualification
also exposed missing normal workspace dependencies and a private guard
that assumed loose Git refs. Normal owned dependency setup and an exact
loose-or-packed reader fixed those setup failures. No business
assertion, timeout, permission boundary or product sleep gate was
relaxed.

Repeat focused product coverage with a fresh external fixture database
and the normal server Vitest command, one worker:

```sh
pnpm --filter @paperclipai/server exec vitest run src/__tests__/plugin-lifecycle-delivery.test.ts --maxWorkers=1 --no-file-parallelism
pnpm --filter @paperclipai/server exec vitest run src/__tests__/agent-runtime-hooks.test.ts --maxWorkers=1 --no-file-parallelism
pnpm --filter @paperclipai/server exec vitest run src/__tests__/agent-runtime-assignments.test.ts --maxWorkers=1 --no-file-parallelism
```

Exact current head `0f83005122ef1bdb1f1e08f8f58424c9a840ebb1` has 60
successful checks, three skips and one neutral result, with no pending
or failing contexts. Greptile completed its full review at 5/5 with no
actionable issue. All root and nested review pages were read; the
earlier eligibility finding is resolved. Ponytail's latest turn stopped
before a verdict. A normal retry was requested; no current Ponytail
verdict or score is claimed. Previous `3ca86ec` CI and scores are
historical. A full local repository typecheck/test/build or browser run
is not claimed. Controlled local App/SDK/PostgreSQL proof does not
establish live provider, deployed Railway/Neon suspension, signed Cloud
alarms or global exactly-once delivery.

## Risks

- A listener close failure or uncertain inspection must continue to
block sleep.
- Release, expiry, configuration changes, and journal arrival can race
with retirement. Epoch and current-hold checks retain conservative
results.
- Delivery remains at least once across uncertain ACK boundaries. This
change does not add global exactly-once delivery.
- The database helper closes only its exclusive listener client. Other
owners and subscriptions keep their resources.

## Model Used

OpenAI `gpt-6.1-sol`, reasoning and code/tool use. The managing Codex
agent implemented shared database/composition seams and reviewed the
change; its exact model ID was not exposed. The runtime does not expose
a verified context-window size.

## 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
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-authored-by: Codex <noreply@openai.com>

doc/idle-sleep-safety.md

@@ -594,6 +594,19 @@ No plugin approval or configuration bypass is supported. Scheduled and
 externally triggered work needs a durable host wake mechanism before it can
 permit sleep.
 
+The host's agent-runtime lifecycle notification listener has separate ownership
+from its SDK worker. A normal owned safety read retires its exclusive listener
+client only after admitted delivery and journal work settle. Failed or uncertain
+retirement remains a blocker; diagnostics never perform this retirement. Closing
+a subscription alone does not prove that its PostgreSQL backend closed. The
+client belongs only to lifecycle delivery, so retirement preserves the main
+pool and other features' notification channels. Releasing the hold or its existing shared expiry deadline
+recreates the listener and reconstructs delivery from the durable journal before
+normal dispatch, including events whose notification arrived while it was closed.
+This preserves at-least-once delivery and existing company filtering; it does not
+promise exactly-once callback invocation or add a wake guarantee for arbitrary
+external SQL writers while compute is stopped.
+
 This is a single-process admission protocol. A restart clears the owner. The
 host must serialize sleep with deploy/wake operations and bind its final check
 to the instance it stops. An unexpected restart after the final check can still

packages/db/src/client.ts

@@ -260,6 +260,22 @@ export function postgresJsOptions(options: DatabaseClientOptions): Record<string
 // work needs for progress writes. Keep connection configuration private to the
 // originating Db lifetime; callers receive neither URLs nor credentials.
 const dedicatedDbFactories = new WeakMap<object, () => Db>();
+const ownedDatabaseListenerFactories = new WeakMap<object, () => Db>();
+
+export interface OwnedDatabaseListener {
+  listen: Db["$client"]["listen"];
+  close(): Promise<void>;
+}
+
+export function createOwnedDatabaseListener(db: Db): OwnedDatabaseListener {
+  const factory = ownedDatabaseListenerFactories.get(db);
+  if (!factory) throw new Error("owned_listener_requires_create_db");
+  const owned = factory();
+  return {
+    listen: owned.$client.listen.bind(owned.$client),
+    close: () => owned.$client.end({ timeout: 1 }),
+  };
+}
 
 export async function withDedicatedDbConnection<T>(db: Db, action: (dedicated: Db) => Promise<T>): Promise<T> {
   const factory = dedicatedDbFactories.get(db);
@@ -361,6 +377,9 @@ function createDbClient(url: string, resolved: DatabaseClientOptions, owner?: ob
   dedicatedDbFactories.set(db, () => createDbClient(url, {
     ...resolved, maxConnections: 1, applicationName: "paperclip-workspace-finalization-lock",
   }, db));
+  ownedDatabaseListenerFactories.set(db, () => createDbClient(url, {
+    ...resolved, maxConnections: 1,
+  }, db));
   livenessDbFactories.set(db, () => createDbClient(url, {
     ...resolved, maxConnections: LIVENESS_POOL_MAX_CONNECTIONS, applicationName: "paperclip-controller-lease",
   }, db));

packages/db/src/index.ts

@@ -1,6 +1,8 @@
 export {
   createDb,
   withDedicatedDbConnection,
+  createOwnedDatabaseListener,
+  type OwnedDatabaseListener,
   livenessDb,
   closeLivenessDb,
   checkpointDb,

server/src/__tests__/agent-runtime-assignments.test.ts

@@ -4,7 +4,7 @@ import { mkdtemp, realpath, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { and, eq, sql } from "drizzle-orm";
-import { agents, heartbeatRuns, issues, companies, companyMemberships, createDb, instanceSettings, pluginConfig, plugins, projects, toolApplications, toolConnections, toolConnectionInstalls, toolProfiles, toolMcpGateways, toolProfileBindings, resourceLifecycleEvents, type Db } from "@paperclipai/db";
+import { agents, heartbeatRuns, issues, companies, companyMemberships, applyPendingMigrations, closeFileStorageDb, createDb, instanceSettings, pluginConfig, plugins, projects, toolApplications, toolConnections, toolConnectionInstalls, toolProfiles, toolMcpGateways, toolProfileBindings, resourceLifecycleEvents, type Db } from "@paperclipai/db";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { pluginAgentRuntimeState } from "../services/plugin-agent-runtime-state.js";
@@ -18,7 +18,18 @@ import { aiConnectionService } from "../services/ai-connections.js";
 
 vi.mock("../services/subscription-refresh.js", () => ({ refreshSubscriptionConnection: vi.fn().mockResolvedValue(undefined) }));
 
-const support = await getEmbeddedPostgresTestSupport();
+const externalUrl = process.env.PAPERCLIP_AGENT_RUNTIME_ASSIGNMENTS_TEST_DATABASE_URL;
+const expectedDatabase = process.env.PAPERCLIP_AGENT_RUNTIME_ASSIGNMENTS_EXPECTED_DATABASE;
+if (externalUrl || expectedDatabase) {
+  if (!externalUrl || !expectedDatabase || !/^sleep2_[a-z0-9_]+_test$/.test(expectedDatabase)) {
+    throw new Error("Explicit external fixture URL and expected database are required");
+  }
+  const target = new URL(externalUrl);
+  if (!["postgres:", "postgresql:"].includes(target.protocol) || target.hostname !== "127.0.0.1" || target.port !== "55448" || target.pathname !== `/${expectedDatabase}` || target.search) {
+    throw new Error("External fixture must match the exact owned database");
+  }
+}
+const support = externalUrl ? { supported: true as const } : await getEmbeddedPostgresTestSupport();
 const describeDatabase = support.supported ? describe : describe.skip;
 
 describeDatabase("connection assignment synchronization", () => {
@@ -30,8 +41,18 @@ describeDatabase("connection assignment synchronization", () => {
   let pluginId: string;
   let applicationId: string;
   let connectionId: string;
-  beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-assignments-"); db = createDb(database.connectionString); }, 90_000);
-  afterAll(async () => { await database?.cleanup(); });
+  beforeAll(async () => {
+    if (externalUrl) await applyPendingMigrations(externalUrl);
+    else database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-assignments-");
+    db = createDb(externalUrl ?? database.connectionString, { maxConnections: 4 });
+  }, 90_000);
+  afterAll(async () => {
+    try { if (db) await closeFileStorageDb(db, 1); }
+    finally {
+      try { await db?.$client.end({ timeout: 1 }); }
+      finally { await database?.cleanup(); }
+    }
+  });
   afterEach(() => vi.unstubAllEnvs());
   beforeEach(async () => {
     companyId = randomUUID(); agentId = randomUUID(); peerId = randomUUID(); pluginId = randomUUID();
@@ -173,7 +194,7 @@ describeDatabase("connection assignment synchronization", () => {
         await vi.waitFor(() => expect(snapshots).toEqual(Array(2).fill({ publicMcp: enabled, chatConnectors: enabled })), { timeout: 5000 });
         await vi.waitFor(async () => expect(await inbox.list(companyId, 100, undefined, ["agent_runtime"])).toEqual([]), { timeout: 5000 });
         expect(await events()).toHaveLength(before + 2);
-      } finally { delivery.dispose(); await coordinator.stop(); }
+      } finally { await delivery.dispose(); await coordinator.stop(); }
     }
   });
 
@@ -214,7 +235,7 @@ describeDatabase("connection assignment synchronization", () => {
       { companyId, profileKey: "single-target", name: "Single target" },
     ]).returning();
     if (change !== "bind") await db.insert(toolProfileBindings).values({ companyId, profileId: source.id, targetType: "agent", targetId: agentId });
-    const single = createDb(database.connectionString, { maxConnections: 1 });
+    const single = createDb(externalUrl ?? database.connectionString, { maxConnections: 1 });
     const service = toolAccessService(single);
     let timer: ReturnType<typeof setTimeout> | undefined;
     const work = change === "bind" ? service.bindProfile(source.id, { targetType: "agent", targetId: agentId, priority: 100 })

server/src/__tests__/agent-runtime-hooks.test.ts

@@ -1,3 +1,4 @@
+import * as databaseModule from "@paperclipai/db";
 import { randomUUID } from "node:crypto";
 import { createServer, connect, type Socket } from "node:net";
 import { EventEmitter } from "node:events";
@@ -5,7 +6,7 @@ import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { and, eq, sql } from "drizzle-orm";
-import { activityLog, agents, companies, createDb, pluginCompanySettings, pluginConfig, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db";
+import { activityLog, agents, companies, applyPendingMigrations, closeFileStorageDb, createDb, pluginCompanySettings, pluginConfig, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import * as instructionModule from "../services/agent-instructions.js";
@@ -18,7 +19,18 @@ import { resolveManagedInstructionsRoot } from "../services/agent-instructions.j
 import { agentFileStore } from "../services/agent-file-store.js";
 import { agentInstructionRevisionService } from "../services/agent-instruction-revisions.js";
 
-const support = await getEmbeddedPostgresTestSupport();
+const externalUrl = process.env.PAPERCLIP_AGENT_RUNTIME_HOOKS_TEST_DATABASE_URL;
+const expectedDatabase = process.env.PAPERCLIP_AGENT_RUNTIME_HOOKS_EXPECTED_DATABASE;
+if (externalUrl || expectedDatabase) {
+  if (!externalUrl || !expectedDatabase || !/^sleep2_[a-z0-9_]+_test$/.test(expectedDatabase)) {
+    throw new Error("Explicit external fixture URL and expected database are required");
+  }
+  const target = new URL(externalUrl);
+  if (!["postgres:", "postgresql:"].includes(target.protocol) || target.hostname !== "127.0.0.1" || target.port !== "55448" || target.pathname !== `/${expectedDatabase}` || target.search) {
+    throw new Error("External fixture must match the exact owned database");
+  }
+}
+const support = externalUrl ? { supported: true as const } : await getEmbeddedPostgresTestSupport();
 const describeDatabase = support.supported ? describe : describe.skip;
 
 describeDatabase("agent runtime synchronization hooks", () => {
@@ -29,11 +41,21 @@ describeDatabase("agent runtime synchronization hooks", () => {
   let agentId: string;
   let home: string;
   beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("paperclip-agent-runtime-");
-    db = createDb(database.connectionString);
+    if (externalUrl) await applyPendingMigrations(externalUrl);
+    else database = await startEmbeddedPostgresTestDatabase("paperclip-agent-runtime-");
+    db = createDb(externalUrl ?? database.connectionString, { maxConnections: 4 });
     home = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-agent-runtime-home-"));
   }, 90_000);
-  afterAll(async () => { await database?.cleanup(); await fs.rm(home, { recursive: true, force: true }); });
+  afterAll(async () => {
+    try { if (db) await closeFileStorageDb(db, 1); }
+    finally {
+      try { await db?.$client.end({ timeout: 1 }); }
+      finally {
+        try { await database?.cleanup(); }
+        finally { if (home) await fs.rm(home, { recursive: true, force: true }); }
+      }
+    }
+  });
   afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); });
   beforeEach(async () => {
     vi.stubEnv("PAPERCLIP_HOME", home);
@@ -232,14 +254,17 @@ describeDatabase("agent runtime synchronization hooks", () => {
   });
 
   it("wakes delivery after a commit on an independent database client, without polling", async () => {
-    const other = createDb(database.connectionString);
+    const other = createDb(externalUrl ?? database.connectionString, { maxConnections: 4 });
     const initialCount = (await events()).length;
     const notifications: string[] = [];
-    const listen = db.$client.listen.bind(db.$client);
-    vi.spyOn(db.$client, "listen").mockImplementation((channel, onnotify, onlisten) => listen(channel, payload => {
-      notifications.push(payload);
-      onnotify(payload);
-    }, onlisten));
+    const createListener = databaseModule.createOwnedDatabaseListener;
+    vi.spyOn(databaseModule, "createOwnedDatabaseListener").mockImplementation(owner => {
+      const listener = createListener(owner);
+      return { listen: (channel, onnotify, onlisten) => listener.listen(channel, payload => {
+        notifications.push(payload);
+        onnotify(payload);
+      }, onlisten), close: () => listener.close() };
+    });
     const call = vi.fn(async (_plugin, _method, event) => ({ eventId: event.id, status: "acknowledged" }));
     const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
     const delivery = registerPluginLifecycleDelivery(coordinator, db, { call, isRunning: () => true, onConfigApplied: () => () => {} } as never, new EventEmitter() as never);
@@ -268,10 +293,10 @@ describeDatabase("agent runtime synchronization hooks", () => {
       });
       await vi.waitFor(() => expect(delivered().map(call => call[2].id)).toContain(committedEventId), { timeout: 5000 });
       expect(delivered().find(call => call[2].id === committedEventId)![2]).toMatchObject({ companyId, resourceId: agentId, resourceType: "agent_runtime" });
-    } finally { delivery.dispose(); await coordinator.stop(); await other.$client.end(); }
+    } finally { await delivery.dispose(); await coordinator.stop(); await other.$client.end(); }
   });
   it("recovers work committed during real listener connection loss without another write", async () => {
-    const target = new URL(database.connectionString);
+    const target = new URL(externalUrl ?? database.connectionString);
     const upstreamHost = target.hostname; const upstreamPort = Number(target.port);
     let blocked = false;
     const sockets = new Set<Socket>();
@@ -287,7 +312,7 @@ describeDatabase("agent runtime synchronization hooks", () => {
     });
     await new Promise<void>(resolve => proxy.listen(0, "127.0.0.1", resolve));
     target.hostname = "127.0.0.1"; target.port = String((proxy.address() as { port: number }).port);
-    const listenerDb = createDb(target.toString(), { connectTimeoutSeconds: 1 });
+    const listenerDb = createDb(target.toString(), { connectTimeoutSeconds: 1, maxConnections: 4 });
     const call = vi.fn(async (_plugin, _method, event) => ({ eventId: event.id, status: "acknowledged" }));
     const coordinator = createDeliveryWorkCoordinator({ owner: listenerDb, canRun: () => true, onError: () => {} });
     const delivery = registerPluginLifecycleDelivery(coordinator, listenerDb, { call, isRunning: () => true, onConfigApplied: () => () => {} } as never, new EventEmitter() as never);
@@ -304,7 +329,7 @@ describeDatabase("agent runtime synchronization hooks", () => {
       blocked = false;
       await vi.waitFor(() => expect(call.mock.calls.some(call => call[2].id === String(queued.id))).toBe(true), { timeout: 10000 });
     } finally {
-      blocked = false; delivery.dispose(); await coordinator.stop(); await listenerDb.$client.end({ timeout: 1 });
+      blocked = false; await delivery.dispose(); await coordinator.stop(); await listenerDb.$client.end({ timeout: 1 });
       for (const socket of sockets) socket.destroy();
       await new Promise<void>(resolve => proxy.close(() => resolve()));
     }

server/src/__tests__/plugin-lifecycle-delivery.test.ts

@@ -3,8 +3,11 @@ import { EventEmitter } from "node:events";
 import { mkdir, mkdtemp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
-import { eq } from "drizzle-orm";
-import { agents, companies, createDb, companySecrets, projects, pluginCompanySettings, pluginConfig, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db";
+import { eq, sql } from "drizzle-orm";
+import * as databaseModule from "@paperclipai/db";
+import * as idleWorkModule from "../services/idle-local-work.js";
+import { idleWorkSnapshot, isIdleTaskDrainActive, registerTaskDrainExpiryWork, startTaskDrain, stopTaskDrain } from "../services/task-admission.js";
+import { agents, companies, applyPendingMigrations, createDb, companySecrets, projects, pluginCompanySettings, pluginConfig, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { logger } from "../middleware/logger.js";
@@ -17,7 +20,14 @@ import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordin
 import type { PluginLifecycleManager } from "../services/plugin-lifecycle.js";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
 
-const support = await getEmbeddedPostgresTestSupport();
+const externalUrl = process.env.PAPERCLIP_PLUGIN_LIFECYCLE_TEST_DATABASE_URL;
+const expectedDatabase = process.env.PAPERCLIP_PLUGIN_LIFECYCLE_EXPECTED_DATABASE;
+if (externalUrl || expectedDatabase) {
+  if (!externalUrl || !expectedDatabase || !/^sleep2_[a-z0-9_]+_test$/.test(expectedDatabase)) throw new Error("Explicit lifecycle test database required");
+  const target = new URL(externalUrl);
+  if (!["postgres:", "postgresql:"].includes(target.protocol) || target.hostname !== "127.0.0.1" || target.port !== "55448" || target.pathname !== `/${expectedDatabase}` || target.search) throw new Error("Unexpected lifecycle test database");
+}
+const support = externalUrl ? { supported: true as const } : await getEmbeddedPostgresTestSupport();
 const describePostgres = support.supported ? describe : describe.skip;
 if (!support.supported) console.warn(`Skipping lifecycle delivery database tests: ${support.reason}`);
 
@@ -26,11 +36,26 @@ describePostgres("durable plugin lifecycle delivery", () => {
   let db: Db;
   let companyId: string;
   let pluginId: string;
-  beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("paperclip-plugin-delivery-"); db = createDb(database.connectionString); }, 90_000);
-  afterAll(async () => { await database?.cleanup(); });
-  afterEach(() => { vi.restoreAllMocks(); });
+  let listenerOwners: Array<{ listen: Db["$client"]["listen"]; close: ReturnType<typeof vi.fn> }>;
+  const deliveries = new Set<ReturnType<typeof registerPluginLifecycleDelivery>>();
+  beforeAll(async () => {
+    if (externalUrl) await applyPendingMigrations(externalUrl);
+    else database = await startEmbeddedPostgresTestDatabase("paperclip-plugin-delivery-");
+    db = createDb(externalUrl ?? database.connectionString, { maxConnections: 4 });
+  }, 90_000);
+  afterAll(async () => { try { await db?.$client.end({ timeout: 1 }); } finally { await database?.cleanup(); } });
+  afterEach(async () => {
+    try { await Promise.all([...deliveries].map(worker => worker.dispose())); }
+    finally { deliveries.clear(); stopTaskDrain(); vi.restoreAllMocks(); }
+  });
   beforeEach(async () => {
+    listenerOwners = [];
     vi.spyOn(db.$client, "listen").mockResolvedValue({ unlisten: vi.fn(), state: {} } as never);
+    vi.spyOn(databaseModule, "createOwnedDatabaseListener").mockImplementation(() => {
+      const owner = { listen: db.$client.listen, close: vi.fn().mockResolvedValue(undefined) };
+      listenerOwners.push(owner);
+      return owner;
+    });
     await db.delete(plugins);
     companyId = randomUUID(); pluginId = randomUUID();
     await db.insert(companies).values({ id: companyId, name: "Delivery fixture", issuePrefix: `L${companyId.slice(0, 6)}` });
@@ -48,13 +73,14 @@ describePostgres("durable plugin lifecycle delivery", () => {
   function delivery() {
     let task!: Parameters<ReturnType<typeof createDeliveryWorkCoordinator>["register"]>[1];
     const wake = vi.fn();
-    const coordinator = { register: vi.fn((_queue, input) => { task = input; return { wake, ready: Promise.resolve() }; }) };
+    const coordinator = { register: vi.fn((_queue, input) => { task = input; return { wake, ready: Promise.resolve(), stop: vi.fn().mockResolvedValue(undefined) }; }) };
     const call = vi.fn(async (_pluginId, _method, input) => ({ eventId: input.id, status: "acknowledged" }));
     const running = vi.fn(() => true);
     const lifecycle = new EventEmitter();
     const configApplied = new Set<() => void>();
     const onConfigApplied = (listener: () => void) => { configApplied.add(listener); return () => { configApplied.delete(listener); }; };
     const worker = registerPluginLifecycleDelivery(coordinator as never, db, { call, isRunning: running, onConfigApplied } as unknown as PluginWorkerManager, lifecycle as unknown as PluginLifecycleManager);
+    deliveries.add(worker);
     return { call, running, wake, lifecycle, worker, applied: () => { for (const listener of configApplied) listener(); }, run: () => task.run(new AbortController().signal), pending: () => task.hasPending(), nextRunAt: () => task.nextRunAt?.(), settled: () => task.onSettled?.() };
   }
   async function withManagedPlugin(enabledCompanies: string[], run: () => Promise<void>) {
@@ -91,7 +117,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       expect(delivered(managedId)).toEqual([resourceId]);
       expect(delivered(ungrantedId)).toEqual([]);
       expect(delivered(pluginId)).toEqual([resourceId]);
-      d.worker.dispose();
+      await d.worker.dispose();
     });
   });
   it("listens for a managed plugin with no config row until its last enabled company disables it", async () => {
@@ -107,8 +133,8 @@ describePostgres("durable plugin lifecycle delivery", () => {
       expect(db.$client.listen).toHaveBeenCalledWith("paperclip_plugin_lifecycle", expect.any(Function), expect.any(Function));
       await db.insert(pluginCompanySettings).values({ pluginId: managedId, companyId, enabled: false });
       d.applied();
-      await vi.waitFor(() => expect(subscription.unlisten).toHaveBeenCalledTimes(1));
-      d.worker.dispose();
+      await vi.waitFor(() => expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1));
+      await d.worker.dispose();
     });
   });
   it("does not listen for a managed plugin that the document enables for no company", async () => {
@@ -124,7 +150,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       const d = delivery();
       await d.worker.ready;
       expect(db.$client.listen).not.toHaveBeenCalled();
-      d.worker.dispose();
+      await d.worker.dispose();
     } finally { vi.unstubAllEnvs(); await rm(root, { recursive: true, force: true }); }
   });
   it("delivers project events to a storage provider only for projects attached to it", async () => {
@@ -145,7 +171,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       expect(d.call).toHaveBeenCalledTimes(2);
       expect(d.call.mock.calls[1][2]).toMatchObject({ resourceId: core.id, action: "update" });
     } finally {
-      d.worker.dispose();
+      await d.worker.dispose();
     }
   });
 
@@ -157,7 +183,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       await d.run();
       expect(d.call.mock.calls[0][2]).toMatchObject({ resourceId: removed, action: "delete" });
     } finally {
-      d.worker.dispose();
+      await d.worker.dispose();
     }
   });
 
@@ -184,7 +210,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       const [stored] = await db.select().from(companySecrets).where(eq(companySecrets.ownerProjectId, project.id));
       expect(stored.latestVersion).toBe(1);
       expect(JSON.stringify(await pluginLifecycleInbox(db, pluginId).list(companyId))).not.toContain("fixture-private-grants");
-    } finally { d.worker.dispose(); vi.unstubAllEnvs(); }
+    } finally { await d.worker.dispose(); vi.unstubAllEnvs(); }
   });
   it("rejects credential patches from plugins without project instance secret scope", async () => {
     await event(); const d = delivery();
@@ -193,7 +219,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       return { eventId: input.id, status: "acknowledged", instanceSecrets: { mount_grants: "private" } };
     });
     try { await d.run(); expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toHaveLength(1); }
-    finally { d.worker.dispose(); }
+    finally { await d.worker.dispose(); }
   });
   it.each([false, true])("does not wait for queued plugin callbacks during startup (runtime subscriber: %s)", async runtime => {
     if (runtime) await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["project", "agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
@@ -209,7 +235,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       await vi.waitFor(() => expect(call).toHaveBeenCalled(), { timeout: 5000 });
       await vi.waitFor(() => expect(ready).toBe(true));
       expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toHaveLength(1);
-    } finally { release(); worker.dispose(); await coordinator.stop(); }
+    } finally { release(); await worker.dispose(); await coordinator.stop(); }
   });
   describe("agent runtime events", () => {
     const runtimeManifest = (capabilities: string[]) => ({ resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe", ...capabilities] }) as never;
@@ -235,7 +261,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
         expect(delivered(d)).toEqual(new Set([owned]));
         expect(await pluginLifecycleInbox(db, pluginId).list(companyId, 100, undefined, ["agent_runtime"])).toEqual([]);
       } finally {
-        d.worker.dispose();
+        await d.worker.dispose();
       }
     });
 
@@ -251,7 +277,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
         await drain(d);
         expect(delivered(d)).toEqual(new Set([id]));
       } finally {
-        d.worker.dispose();
+        await d.worker.dispose();
       }
     });
 
@@ -264,7 +290,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
         await drain(d);
         expect(d.call.mock.calls.at(-1)?.[2]).toMatchObject({ resourceId: id, action: "delete" });
       } finally {
-        d.worker.dispose();
+        await d.worker.dispose();
       }
     });
 
@@ -276,7 +302,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
         await drain(d);
         expect(delivered(d)).toEqual(new Set([id]));
       } finally {
-        d.worker.dispose();
+        await d.worker.dispose();
       }
     });
   });
@@ -284,13 +310,13 @@ describePostgres("durable plugin lifecycle delivery", () => {
   it("drains creation before transitions, recovers acknowledgments across dispatcher replacement, and ignores agent events", async () => {
     const id = await event(); await event(id, "update"); await event(id, "delete");
     await db.transaction(tx => recordResourceCreationEvent(tx as unknown as Db, companyId, "agent", randomUUID()));
-    const first = delivery(); await first.run(); first.worker.dispose();
+    const first = delivery(); await first.run(); await first.worker.dispose();
     expect(first.call.mock.calls.map(c => c[2].action)).toEqual(["create"]);
     const restarted = delivery(); await restarted.run(); await restarted.run(); await restarted.run();
     expect(restarted.call.mock.calls.map(c => c[2].action)).toEqual(["update", "delete"]);
     expect(await restarted.pending()).toBe(false);
     expect((await pluginLifecycleInbox(db, pluginId).list(companyId)).map(e => e.resourceType)).toEqual(["agent"]);
-    restarted.worker.dispose();
+    await restarted.worker.dispose();
   });
   it("replays failures and invalid acknowledgments without blocking independent resources", async () => {
     let now = Date.now(); vi.spyOn(Date, "now").mockImplementation(() => now);
@@ -304,7 +330,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     now += 30_000; await d.run(); await d.run();
     expect(d.call.mock.calls.map(c => c[2].action)).toEqual(["create", "create", "create", "create", "delete"]);
     expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]);
-    d.worker.dispose();
+    await d.worker.dispose();
   });
   it("healthy acknowledgments and new events do not bypass a failed event's retry deadline", async () => {
     let now = Date.now(); vi.spyOn(Date, "now").mockImplementation(() => now);
@@ -325,7 +351,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     await d.run();
     expect(d.call.mock.calls.filter(c => c[2].resourceId === failed)).toHaveLength(2);
     expect(d.nextRunAt()).toBe(now + 30_000);
-    d.worker.dispose();
+    await d.worker.dispose();
   });
   it("defers without a retry deadline, retries unfinished work, and wakes on plugin startup", async () => {
     await event(); const d = delivery();
@@ -335,7 +361,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     d.call.mockImplementationOnce(async (_p, _m, e) => ({ eventId: e.id, status: "retry" }));
     await d.run(); expect(await d.pending()).toBe(true);
     d.running.mockReturnValue(false); await d.run(); expect(await d.pending()).toBe(true);
-    expect(d.call).toHaveBeenCalledTimes(2); d.worker.dispose();
+    expect(d.call).toHaveBeenCalledTimes(2); await d.worker.dispose();
     d.lifecycle.emit("plugin.worker_started"); expect(d.wake).toHaveBeenCalledTimes(1);
   });
   it("requires ready, configured, company-enabled plugins and rechecks disabling within a batch", async () => {
@@ -353,7 +379,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     await d.run(); expect(d.call).toHaveBeenCalledTimes(1);
     await db.update(plugins).set({ status: "ready" }).where(eq(plugins.id, pluginId));
     await db.delete(pluginConfig).where(eq(pluginConfig.pluginId, pluginId));
-    await d.run(); expect(d.call).toHaveBeenCalledTimes(1); d.worker.dispose();
+    await d.run(); expect(d.call).toHaveBeenCalledTimes(1); await d.worker.dispose();
   });
   it("notifies only after the outer transaction settles and wakes deferred work for config changes", async () => {
     const wake = vi.fn(); const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle, wake);
@@ -388,7 +414,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     d.lifecycle.emit("plugin.enabled"); expect(await d.pending()).toBe(true);
     await d.run(); expect(await d.pending()).toBe(false);
     d.worker.wake(); expect(await d.pending()).toBe(true);
-    d.worker.dispose();
+    await d.worker.dispose();
   });
   it("wakes deferred events after live configuration finishes applying", async () => {
     await event(); const d = delivery();
@@ -400,7 +426,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     configured = true; d.applied();
     expect(await d.pending()).toBe(true); expect(d.wake).toHaveBeenCalledTimes(1);
     await d.run(); expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]);
-    d.worker.dispose(); d.wake.mockClear(); d.applied(); expect(d.wake).not.toHaveBeenCalled();
+    await d.worker.dispose(); d.wake.mockClear(); d.applied(); expect(d.wake).not.toHaveBeenCalled();
   });
   it("retains a failed scan across a retry skipped during idle drain", async () => {
     await event(); const d = delivery();
@@ -412,15 +438,17 @@ describePostgres("durable plugin lifecycle delivery", () => {
     select.mockRestore(); await d.run(); await d.run();
     expect(await d.pending()).toBe(false);
     expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]);
-    d.worker.dispose();
+    await d.worker.dispose();
   });
   it("saves config and settings without borrowing another pool connection", async () => {
-    const single = createDb(database.connectionString, { maxConnections: 1 });
-    const registry = pluginRegistryService(single);
-    await registry.upsertConfig(pluginId, companyId, { companyId, configJson: { first: true } });
-    await registry.patchConfig(pluginId, companyId, { companyId, configJson: { second: true } });
-    await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} });
-    expect((await registry.getConfig(pluginId, companyId))?.configJson).toEqual({ first: true, second: true });
+    const single = createDb(externalUrl ?? database.connectionString, { maxConnections: 1 });
+    try {
+      const registry = pluginRegistryService(single);
+      await registry.upsertConfig(pluginId, companyId, { companyId, configJson: { first: true } });
+      await registry.patchConfig(pluginId, companyId, { companyId, configJson: { second: true } });
+      await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} });
+      expect((await registry.getConfig(pluginId, companyId))?.configJson).toEqual({ first: true, second: true });
+    } finally { await single.$client.end({ timeout: 1 }); }
   }, 5000);
   it("retries listener setup through durable delivery after an enable-time failure", async () => {
     const d = delivery();
@@ -434,7 +462,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
     await d.run();
     expect(listen).toHaveBeenCalledTimes(2);
     expect(await d.pending()).toBe(false);
-    d.worker.dispose();
+    await d.worker.dispose();
   });
   it("unregisters the listener on the last disable or unload and can subscribe again", async () => {
     await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
@@ -450,10 +478,10 @@ describePostgres("durable plugin lifecycle delivery", () => {
     await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, pluginId));
     d.lifecycle.emit("plugin.status_changed");
     await d.run();
-    expect(first.unlisten).not.toHaveBeenCalled();
+    expect(listenerOwners[0]!.close).not.toHaveBeenCalled();
     await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, otherPluginId));
     d.lifecycle.emit("plugin.disabled");
-    await vi.waitFor(() => expect(first.unlisten).toHaveBeenCalledTimes(1));
+    await vi.waitFor(() => expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1));
     await db.update(plugins).set({ status: "ready" }).where(eq(plugins.id, pluginId));
     d.lifecycle.emit("plugin.enabled");
     await d.run();
@@ -461,20 +489,283 @@ describePostgres("durable plugin lifecycle delivery", () => {
     const registry = pluginRegistryService(db);
     await registry.upsertCompanySettings(pluginId, companyId, { enabled: false, settingsJson: {} });
     d.applied();
-    await vi.waitFor(() => expect(second.unlisten).toHaveBeenCalledTimes(1));
+    await vi.waitFor(() => expect(listenerOwners[1]!.close).toHaveBeenCalledTimes(1));
     await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} });
     d.applied(); await d.run();
     expect(listen).toHaveBeenCalledTimes(3);
     await db.delete(plugins).where(eq(plugins.id, pluginId));
     d.lifecycle.emit("plugin.unloaded");
-    await vi.waitFor(() => expect(third.unlisten).toHaveBeenCalledTimes(1));
-    expect(first.unlisten).toHaveBeenCalledTimes(1);
-    expect(second.unlisten).toHaveBeenCalledTimes(1);
-    d.worker.dispose();
+    await vi.waitFor(() => expect(listenerOwners[2]!.close).toHaveBeenCalledTimes(1));
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1);
+    expect(listenerOwners[1]!.close).toHaveBeenCalledTimes(1);
+    await d.worker.dispose();
     expect(d.lifecycle.listenerCount("plugin.status_changed")).toBe(0);
     expect(d.lifecycle.listenerCount("plugin.disabled")).toBe(0);
     expect(d.lifecycle.listenerCount("plugin.unloaded")).toBe(0);
   });
+  it("retires disabled-feature retry history without discarding the original backoff on re-enable", async () => {
+    const register = idleWorkModule.registerIdleLocalWorkProbe;
+    let inspect: (() => string) | undefined;
+    vi.spyOn(idleWorkModule, "registerIdleLocalWorkProbe").mockImplementation(probe => {
+      inspect = () => probe({}); return register(probe);
+    });
+    let now = Date.now(); vi.spyOn(Date, "now").mockImplementation(() => now);
+    await event();
+    const d = delivery(); await d.worker.ready;
+    d.call.mockImplementationOnce(async () => { throw new Error("provider retry fixture"); });
+    await d.run();
+    const originalDeadline = now + 30_000;
+    expect(await d.pending()).toBe(true);
+    expect(d.nextRunAt()).toBe(originalDeadline);
+    expect(d.call).toHaveBeenCalledTimes(1);
+    const registry = pluginRegistryService(db);
+    now += 1_000;
+    await registry.upsertCompanySettings(pluginId, companyId, { enabled: false, settingsJson: {} });
+    d.applied(); await d.run();
+    expect(await d.pending()).toBe(false);
+    expect(d.nextRunAt()).toBeNull();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("none");
+    expect(inspect?.()).toBe("none");
+    stopTaskDrain();
+    now += 1_000;
+    await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} });
+    d.applied(); await d.run();
+    expect(await d.pending()).toBe(true);
+    expect(d.nextRunAt()).toBe(originalDeadline);
+    expect(d.call).toHaveBeenCalledTimes(1);
+    expect(inspect?.()).toBe("present");
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("present");
+    stopTaskDrain();
+    now = originalDeadline - 1; await d.run();
+    expect(d.call).toHaveBeenCalledTimes(1);
+    expect(d.nextRunAt()).toBe(originalDeadline);
+    now = originalDeadline; await d.run(); await d.run();
+    expect(d.call).toHaveBeenCalledTimes(2);
+    expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]);
+    expect(await d.pending()).toBe(false);
+    expect(d.nextRunAt()).toBeNull();
+  });
+  it("requires a current owned idle drain even when the caller supplies a true predicate", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready; await d.run(); await d.run();
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("unknown");
+    expect(listenerOwners[0]!.close).not.toHaveBeenCalled();
+  });
+  it("retains an uncertain listener close and refuses retirement after authority changes", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready; await d.run(); await d.run();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    const close = listenerOwners[0]!.close;
+    close.mockRejectedValueOnce(new Error("listener close uncertain"));
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("unknown");
+    expect(close).toHaveBeenCalledTimes(1);
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => false })).toBe("unknown");
+    expect(close).toHaveBeenCalledTimes(1);
+    close.mockRejectedValueOnce(new Error("listener close still uncertain"));
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("unknown");
+    expect(close).toHaveBeenCalledTimes(2);
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("none");
+    expect(close).toHaveBeenCalledTimes(3);
+  });
+  it("recovers a post-close journal inspection failure without creating another listener", async () => {
+    const register = idleWorkModule.registerIdleLocalWorkProbe;
+    let inspect: (() => string) | undefined;
+    vi.spyOn(idleWorkModule, "registerIdleLocalWorkProbe").mockImplementation(probe => {
+      inspect = () => probe({});
+      return register(probe);
+    });
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready; await d.run(); await d.run();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    listenerOwners[0]!.close.mockImplementationOnce(async () => {
+      vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("post-close journal unavailable"); });
+    });
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("unknown");
+    expect(inspect?.()).toBe("unknown");
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1);
+    vi.mocked(db.select).mockRestore();
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("none");
+    expect(inspect?.()).toBe("none");
+    expect(listenerOwners).toHaveLength(1);
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1);
+  });
+  it("recovers an absent-feature inspection failure without manufacturing a listener", async () => {
+    const register = idleWorkModule.registerIdleLocalWorkProbe;
+    let inspect: (() => string) | undefined;
+    vi.spyOn(idleWorkModule, "registerIdleLocalWorkProbe").mockImplementation(probe => {
+      inspect = () => probe({});
+      return register(probe);
+    });
+    const d = delivery(); await d.worker.ready; await d.run();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    const select = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("journal unavailable"); });
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("unknown");
+    expect(inspect?.()).toBe("unknown");
+    select.mockRestore();
+    expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("none");
+    expect(inspect?.()).toBe("none");
+    expect(listenerOwners).toHaveLength(0);
+  });
+  it("retries failed normal disposal without losing its owned listener", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready;
+    listenerOwners[0]!.close.mockRejectedValueOnce(new Error("owned disposal failed"));
+    await expect(d.worker.dispose()).rejects.toThrow("owned disposal failed");
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1);
+    await d.worker.dispose();
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(2);
+  });
+  it("invalidates a retirement when configuration arrives during its owned close", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready; await d.run(); await d.run();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    let release!: () => void, entered!: () => void;
+    const began = new Promise<void>(resolve => { entered = resolve; });
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    listenerOwners[0]!.close.mockImplementationOnce(async () => { entered(); await gate; });
+    const preparation = d.worker.prepareForIdleSleep({ canRetire: () => true });
+    try { await began; d.applied(); }
+    finally { release(); }
+    expect(await preparation).toBe("unknown");
+    expect(await d.pending()).toBe(true);
+    expect(listenerOwners).toHaveLength(1);
+  });
+  it("refuses an owned retirement while a real lifecycle callback remains active", async () => {
+    await event(); const d = delivery(); await d.worker.ready;
+    let release!: () => void, entered!: () => void;
+    const began = new Promise<void>(resolve => { entered = resolve; });
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    d.call.mockImplementationOnce(async (_p, _m, value) => { entered(); await gate; return { eventId: value.id, status: "acknowledged" }; });
+    const run = d.run();
+    try {
+      await began;
+      startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+      expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("present");
+      expect(listenerOwners.every(owner => owner.close.mock.calls.length === 0)).toBe(true);
+    } finally { release(); await run; }
+  });
+  it("keeps journal arrivals during an owned close as actual pending work", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready; await d.run(); await d.run();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    let release!: () => void, entered!: () => void;
+    const began = new Promise<void>(resolve => { entered = resolve; });
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    listenerOwners[0]!.close.mockImplementationOnce(async () => { entered(); await gate; });
+    const preparation = d.worker.prepareForIdleSleep({ canRetire: () => true });
+    try { await began; await db.insert(agents).values({ companyId, name: "Arrived during retirement", adapterType: "process" }); }
+    finally { release(); }
+    expect(await preparation).toBe("present");
+    expect(await pluginLifecycleInbox(db, pluginId).list(companyId, 1, undefined, ["agent_runtime"])).not.toHaveLength(0);
+  });
+  it("finishes exact already-started cleanup but rejects an invalidated retirement permit", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const d = delivery(); await d.worker.ready; await d.run(); await d.run();
+    startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+    let release!: () => void, entered!: () => void, permitted = true;
+    const began = new Promise<void>(resolve => { entered = resolve; });
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    listenerOwners[0]!.close.mockImplementationOnce(async () => { entered(); await gate; });
+    const preparation = d.worker.prepareForIdleSleep({ canRetire: () => permitted });
+    try { await began; permitted = false; }
+    finally { release(); }
+    expect(await preparation).toBe("unknown");
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1);
+  });
+  it("joins an in-progress listener creation on idempotent normal disposal", async () => {
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    let release!: () => void, entered!: () => void;
+    const began = new Promise<void>(resolve => { entered = resolve; });
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    vi.mocked(db.$client.listen).mockImplementationOnce(async () => { entered(); await gate; return { state: {}, unlisten: vi.fn() } as never; });
+    const d = delivery();
+    await began;
+    const disposal = d.worker.dispose();
+    release();
+    await disposal;
+    await d.worker.dispose();
+    expect(listenerOwners[0]!.close).toHaveBeenCalledTimes(1);
+    expect(d.lifecycle.listenerCount("plugin.enabled")).toBe(0);
+  });
+  it("reopens a physically retired listener at normal TTL expiry and delivers newly found journal debt without a manual wake", async () => {
+    vi.mocked(databaseModule.createOwnedDatabaseListener).mockRestore();
+    const createListener = databaseModule.createOwnedDatabaseListener;
+    let created = 0, arrivalId: string | undefined;
+    const foreignCompany = randomUUID();
+    await db.insert(companies).values({ id: foreignCompany, name: "Foreign expiry runtime", issuePrefix: `X${foreignCompany.slice(0, 6)}` });
+    vi.spyOn(databaseModule, "createOwnedDatabaseListener").mockImplementation(owner => {
+      const listener = createListener(owner), ordinal = ++created;
+      return { listen: listener.listen, close: async () => {
+        await listener.close();
+        if (ordinal === 1 && !arrivalId) {
+          await db.transaction(async tx => {
+            const [arrival] = await tx.insert(agents).values({ companyId, name: "Arrived after physical close", adapterType: "process" }).returning();
+            arrivalId = arrival!.id;
+            await tx.insert(agents).values({ companyId: foreignCompany, name: "Foreign after physical close", adapterType: "process" });
+          });
+        }
+      } };
+    });
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const calls = vi.fn(async (_plugin, _method, value) => ({ eventId: value.id, status: "acknowledged" }));
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !isIdleTaskDrainActive(), canReconcile: () => true, onError: error => errors.push(error) });
+    const expiry = registerTaskDrainExpiryWork(coordinator, () => {});
+    const worker = registerPluginLifecycleDelivery(coordinator, db, { call: calls, isRunning: () => true, onConfigApplied: () => () => {} } as never, new EventEmitter() as never);
+    try {
+      await expiry.ready; await worker.ready;
+      await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+      const identity = () => db.execute(sql`SELECT pid, backend_start::text AS started, client_port FROM pg_stat_activity WHERE datname = current_database() AND query = 'listen "paperclip_plugin_lifecycle"'`);
+      const [before] = await identity(); expect(before).toBeDefined();
+      startTaskDrain({ purpose: "idle", ttlMs: 1_000 });
+      await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+      expect(await worker.prepareForIdleSleep({ canRetire: () => true })).toBe("present");
+      expect(await db.execute(sql`SELECT 1 FROM pg_stat_activity WHERE datname = current_database() AND pid = ${before!.pid}::integer AND backend_start = ${before!.started}::timestamptz AND client_port = ${before!.client_port}::integer`)).toHaveLength(0);
+      expect(created).toBe(1);
+      await vi.waitFor(() => expect(calls.mock.calls.some(([, , value]) => value.resourceId === arrivalId && value.companyId === companyId)).toBe(true), { timeout: 5_000 });
+      expect(created).toBe(2);
+      expect(calls.mock.calls.some(([, , value]) => value.companyId === foreignCompany)).toBe(false);
+      const [resumed] = await identity(); expect(resumed).toBeDefined(); expect(resumed!.pid).not.toBe(before!.pid);
+      await vi.waitFor(async () => expect(await db.execute(sql`SELECT 1 FROM plugin_lifecycle_acknowledgments ack JOIN resource_lifecycle_events event ON event.id = ack.event_id WHERE ack.plugin_id = ${pluginId}::uuid AND event.resource_id = ${arrivalId}::uuid AND event.company_id = ${companyId}::uuid`)).not.toHaveLength(0), { timeout: 5_000 });
+      expect(errors).toEqual([]);
+    } finally {
+      try { await worker.dispose(); } finally { try { await expiry.stop(); } finally { await coordinator.stop(); } }
+    }
+  });
+  it("physically closes only its owned LISTEN and recreates it for useful same-company work", async () => {
+    vi.mocked(databaseModule.createOwnedDatabaseListener).mockRestore();
+    vi.mocked(db.$client.listen).mockRestore();
+    await db.update(plugins).set({ manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe"] } as never }).where(eq(plugins.id, pluginId));
+    const foreign = await db.$client.listen("fixture_other_lifecycle", () => {});
+    const d = delivery();
+    try {
+      await d.worker.ready; await d.run(); await d.run();
+      const identity = async (channel: string) => db.execute(sql`SELECT pid, backend_start::text AS started, client_port FROM pg_stat_activity WHERE datname = current_database() AND query = ${'listen "' + channel + '"'}`);
+      const [owned] = await identity("paperclip_plugin_lifecycle");
+      const [other] = await identity("fixture_other_lifecycle");
+      expect(owned).toBeDefined(); expect(other).toBeDefined(); expect(owned!.pid).not.toBe(other!.pid);
+      startTaskDrain({ purpose: "idle", ttlMs: 30_000 });
+      expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("none");
+      expect(await db.execute(sql`SELECT 1 FROM pg_stat_activity WHERE datname = current_database() AND pid = ${owned!.pid}::integer AND backend_start = ${owned!.started}::timestamptz AND client_port = ${owned!.client_port}::integer`)).toHaveLength(0);
+      expect((await identity("fixture_other_lifecycle"))[0]?.pid).toBe(other!.pid);
+      expect(await d.worker.prepareForIdleSleep({ canRetire: () => true })).toBe("none");
+      stopTaskDrain();
+      await d.run(); await d.run();
+      const [resumed] = await identity("paperclip_plugin_lifecycle");
+      expect(resumed).toBeDefined(); expect(resumed!.pid).not.toBe(owned!.pid);
+      const [agent] = await db.insert(agents).values({ companyId, name: "Useful resumed runtime", adapterType: "process" }).returning();
+      const foreignCompany = randomUUID();
+      await db.insert(companies).values({ id: foreignCompany, name: "Foreign runtime", issuePrefix: `F${foreignCompany.slice(0, 6)}` });
+      await db.insert(agents).values({ companyId: foreignCompany, name: "Excluded runtime", adapterType: "process" });
+      await d.run(); await d.run();
+      expect(d.call.mock.calls.some(([, , value]) => value.resourceId === agent!.id && value.companyId === companyId)).toBe(true);
+      expect(d.call.mock.calls.some(([, , value]) => value.companyId === foreignCompany)).toBe(false);
+      const acknowledged = await db.execute(sql`SELECT 1 FROM plugin_lifecycle_acknowledgments ack JOIN resource_lifecycle_events event ON event.id = ack.event_id WHERE ack.plugin_id = ${pluginId}::uuid AND event.company_id = ${companyId}::uuid AND event.resource_id = ${agent!.id}::uuid`);
+      expect(acknowledged.length).toBeGreaterThan(0);
+    } finally { await d.worker.dispose(); await foreign.unlisten(); }
+  });
   it("records failed delivery scope and stage without provider secrets", async () => {
     await event(); const d = delivery();
     const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
@@ -484,7 +775,7 @@ describePostgres("durable plugin lifecycle delivery", () => {
       expect(warn).toHaveBeenCalledWith(expect.objectContaining({ pluginId, companyId, eventId: expect.any(String), stage: "callback", rpcCode: -32001 }), expect.any(String));
       expect(JSON.stringify(warn.mock.calls)).not.toContain("provider-secret-fixture");
       expect(await d.pending()).toBe(true);
-    } finally { warn.mockRestore(); d.worker.dispose(); }
+    } finally { warn.mockRestore(); await d.worker.dispose(); }
   });
   it("rollbacks never expose queue rows even when settlement wakes the dispatcher", async () => {
     const d = delivery();
@@ -494,6 +785,6 @@ describePostgres("durable plugin lifecycle delivery", () => {
     })).rejects.toThrow("rollback");
     await d.run(); expect(d.call).not.toHaveBeenCalled(); expect(await d.pending()).toBe(false);
     expect(await db.select().from(resourceLifecycleEvents).where(eq(resourceLifecycleEvents.companyId, companyId))).toEqual([]);
-    d.worker.dispose();
+    await d.worker.dispose();
   });
 });

server/src/app.ts

@@ -842,7 +842,8 @@ export async function createApp(
   api.use(primaryAgentRoutes(db));
   api.use(workspacePreferenceRoutes(db));
   api.use(inboxDismissalRoutes(db));
-  api.use(instanceSettingsRoutes(db, workerManager, opts.prepareIdleDatabaseBackup, opts.idleDeadlineOwners));
+  api.use(instanceSettingsRoutes(db, workerManager, opts.prepareIdleDatabaseBackup, opts.idleDeadlineOwners,
+    input => resourceLifecycleDelivery.prepareForIdleSleep(input)));
   api.use(ephemeralEnrollmentRoutes(db));
   if (opts.databaseBackupService) {
     api.use(instanceDatabaseBackupRoutes(opts.databaseBackupService));
@@ -1368,26 +1369,27 @@ export async function createApp(
       await publicMcpEvents?.stop();
       await dotMcpEvents?.stop();
       jobCoordinator.stop();
-      resourceLifecycleDelivery.dispose();
-      importSpoolCleanup.unsubscribe();
-      await deliveryWork.stop();
-      devWatcher?.close();
-      viteHtmlRenderer?.dispose();
-      void viteDevServer?.close().catch(() => undefined);
-      viteHmrServer?.close();
-      hostServiceCleanup.disposeAll();
-      hostServiceCleanup.teardown();
-      await stopMuseMaintenance();
-      await emailChannels.shutdown();
-      await chatChannels.shutdown();
-      // End the avatar worker pool, if a request ever started one, so no
-      // render outlives the HTTP teardown.
-      await agentAvatars.close();
-      // Cancel every live setup-token login session and AWAIT the cancellation,
-      // so each direct child stops and the server releases each lease before the
-      // caller stops the database and the provider. A lease release that
-      // fails stays a durable record for the startup reaper.
-      await setupTokenLoginService?.shutdown();
+      await finishServiceShutdown(() => resourceLifecycleDelivery.dispose(), async () => {
+        importSpoolCleanup.unsubscribe();
+        await deliveryWork.stop();
+        devWatcher?.close();
+        viteHtmlRenderer?.dispose();
+        void viteDevServer?.close().catch(() => undefined);
+        viteHmrServer?.close();
+        hostServiceCleanup.disposeAll();
+        hostServiceCleanup.teardown();
+        await stopMuseMaintenance();
+        await emailChannels.shutdown();
+        await chatChannels.shutdown();
+        // End the avatar worker pool, if a request ever started one, so no
+        // render outlives the HTTP teardown.
+        await agentAvatars.close();
+        // Cancel every live setup-token login session and AWAIT the cancellation,
+        // so each direct child stops and the server releases each lease before the
+        // caller stops the database and the provider. A lease release that
+        // fails stays a durable record for the startup reaper.
+        await setupTokenLoginService?.shutdown();
+      });
     });
     return appServicesShutdown;
   };

server/src/routes/instance-settings.ts

@@ -135,7 +135,13 @@ function withTaskDrainTransition<T>(run: () => Promise<T>): Promise<T> {
   return turn;
 }
 
-export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManager, prepareIdleDatabaseBackup?: () => Promise<boolean>, deadlineOwners?: IdleDeadlineOwners) {
+export function instanceSettingsRoutes(
+  db: Db,
+  pluginWorkers?: PluginWorkerManager,
+  prepareIdleDatabaseBackup?: () => Promise<boolean>,
+  deadlineOwners?: IdleDeadlineOwners,
+  prepareIdlePluginLifecycle?: (input: { canRetire: () => boolean }) => Promise<"none" | "present" | "unknown">,
+) {
   const router = Router();
   const svc = instanceSettingsService(db);
   const environments = environmentService(db);
@@ -336,7 +342,8 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
         { deadlineOwners, incomingWake: capabilities, deadlineCapability: domainDeadlines ? capabilities : undefined,
           prepareLocalWork: async ({ canPrepare }) => {
             const result = await heartbeat.prepareIdleWarmNativeSessionsForSleep({ canRetire: canPrepare });
-            return result.busy || result.failed ? "present" : "none";
+            if (result.busy || result.failed) return "present";
+            return prepareIdlePluginLifecycle ? prepareIdlePluginLifecycle({ canRetire: canPrepare }) : "none";
           },
           ...(req.query.idleSleepDiagnostics === "1" ? { collectDiagnostics: true } : {}) });
       res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety });

server/src/services/plugin-lifecycle-delivery.ts

@@ -1,6 +1,6 @@
 import { projectInstanceSecretService } from "./project-instance-secrets.js";
 import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm";
-import { agents, companies, pluginCompanySettings, pluginConfig, plugins, projects, type Db } from "@paperclipai/db";
+import { createOwnedDatabaseListener, type OwnedDatabaseListener, agents, companies, pluginCompanySettings, pluginConfig, plugins, projects, type Db } from "@paperclipai/db";
 import { logger } from "../middleware/logger.js";
 import { pluginLifecycleInbox } from "./plugin-lifecycle-inbox.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
@@ -10,6 +10,8 @@ import { isWorkspaceProvider } from "./project-workspace-provider.js";
 import { managedPluginEnabledCompanies } from "./managed-plugin-worker-env.js";
 import type { PluginWorkerManager } from "./plugin-worker-manager.js";
 import type { PluginLifecycleManager } from "./plugin-lifecycle.js";
+import { registerIdleLocalWorkProbe, type IdleLocalWork } from "./idle-local-work.js";
+import { isIdleTaskDrainActive, subscribeTaskDrainChanges, subscribeTaskDrainExpiry } from "./task-admission.js";
 
 type Coordinator = ReturnType<typeof createDeliveryWorkCoordinator>;
 
@@ -24,7 +26,17 @@ export function registerPluginLifecycleDelivery(
   // The coordinator may reconcile signals while idle drain prevents dispatch.
   let pending = true;
   let disposed = false;
+  let held = false;
+  let epoch = 0;
+  let delivering = 0;
+  let listenerUpdates = 0;
+  let listenerFailure = false;
+  let inspectionFailure = false;
+  let listenerClient: OwnedDatabaseListener | undefined;
+  let disposePromise: Promise<void> | undefined;
   const retryAt = new Map<string, number>();
+  const unregisterIdle = registerIdleLocalWorkProbe(() => listenerFailure || inspectionFailure ? "unknown"
+    : listenerUpdates || delivering || listenerClient || pending ? "present" : "none");
   const refreshedRuntimeCompanies = new Set<string>();
   const retryMs = 30_000;
   async function managedTargets() {
@@ -55,6 +67,13 @@ export function registerPluginLifecycleDelivery(
     return [...configured, ...managed.filter(target => !seen.has(`${target.plugin.id}:${target.companyId}`))].sort((a, b) =>
       (a.plugin.installOrder ?? Infinity) - (b.plugin.installOrder ?? Infinity) || a.companyId.localeCompare(b.companyId));
   }
+  async function deliverableLifecycleTypes({ plugin, companyId }: Awaited<ReturnType<typeof deliveryTargets>>[number]) {
+    const types = plugin.manifestJson.resourceLifecycle;
+    if (!types?.length || !plugin.manifestJson.capabilities.includes("events.subscribe")) return;
+    const [disabled] = await db.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings)
+      .where(and(eq(pluginCompanySettings.pluginId, plugin.id), eq(pluginCompanySettings.companyId, companyId), eq(pluginCompanySettings.enabled, false)));
+    return disabled ? undefined : types;
+  }
   const delivery = coordinator.register(DELIVERY_QUEUES.pluginLifecycle, {
     retryMs,
     hasPending: async () => pending,
@@ -69,16 +88,15 @@ export function registerPluginLifecycleDelivery(
       return next;
     },
     async run(signal) {
+      delivering++;
       pending = false;
       try {
         await ensureListening();
         for (const [key, at] of retryAt) if (at <= Date.now()) retryAt.delete(key);
-        for (const { plugin, companyId } of await deliveryTargets()) {
-          const types = plugin.manifestJson.resourceLifecycle;
-          if (!types?.length || !plugin.manifestJson.capabilities.includes("events.subscribe")) continue;
-          const [disabled] = await db.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings)
-            .where(and(eq(pluginCompanySettings.pluginId, plugin.id), eq(pluginCompanySettings.companyId, companyId), eq(pluginCompanySettings.enabled, false)));
-          if (disabled) continue;
+        for (const target of await deliveryTargets()) {
+          const types = await deliverableLifecycleTypes(target);
+          if (!types) continue;
+          const { plugin, companyId } = target;
           if (types.includes("agent_runtime") && !refreshedRuntimeCompanies.has(companyId)) {
             await db.execute(sql`SELECT paperclip_invalidate_company_agent_runtimes(${companyId}::uuid)`);
             refreshedRuntimeCompanies.add(companyId);
@@ -160,36 +178,85 @@ export function registerPluginLifecycleDelivery(
       } catch (error) {
         pending = true; // A failed scan must survive a retry skipped during idle drain.
         throw error;
-      }
+      } finally { delivering--; }
     },
   });
-  function wake() { if (disposed) return; pending = true; delivery.wake(); }
+  function wake() { if (disposed) return; epoch++; pending = true; delivery.wake(); }
   let listening: Awaited<ReturnType<Db["$client"]["listen"]>> | undefined;
   let listenerUpdate = Promise.resolve();
+  function updateListener<T>(operation: () => Promise<T>): Promise<T> {
+    listenerUpdates++;
+    const update = listenerUpdate.then(operation);
+    listenerUpdate = update.then(() => undefined, () => undefined);
+    return update.finally(() => { listenerUpdates--; });
+  }
+  async function closeListener() {
+    const client = listenerClient;
+    if (!client) return;
+    try { await client.close(); }
+    catch (error) {
+      listenerFailure = true;
+      try { logger.warn({ operation: "plugin_lifecycle_listener_close", failed: true }, "Owned plugin lifecycle listener close failed"); } catch {}
+      throw error;
+    }
+    if (listenerClient === client) { listening = undefined; listenerClient = undefined; }
+    listenerFailure = false;
+  }
   function ensureListening() {
-    const update = listenerUpdate.then(async () => {
-      if (disposed || !db.$client?.listen) return;
-      const configured = await db.select({ manifest: plugins.manifestJson }).from(plugins)
-        .innerJoin(pluginConfig, eq(pluginConfig.pluginId, plugins.id))
-        .leftJoin(pluginCompanySettings, and(eq(pluginCompanySettings.pluginId, plugins.id), eq(pluginCompanySettings.companyId, pluginConfig.companyId)))
-        .where(and(eq(plugins.status, "ready"), or(isNull(pluginCompanySettings.enabled), eq(pluginCompanySettings.enabled, true))));
-      const installed = [...configured, ...(await enabledManagedPlugins()).map(plugin => ({ manifest: plugin.manifestJson }))];
-      if (disposed) return;
+    return updateListener(async () => {
+      if (disposed || held || isIdleTaskDrainActive()) return;
+      let installed: Array<{ manifest: (typeof plugins.$inferSelect)["manifestJson"] }>;
+      try {
+        const configured = await db.select({ manifest: plugins.manifestJson }).from(plugins)
+          .innerJoin(pluginConfig, eq(pluginConfig.pluginId, plugins.id))
+          .leftJoin(pluginCompanySettings, and(eq(pluginCompanySettings.pluginId, plugins.id), eq(pluginCompanySettings.companyId, pluginConfig.companyId)))
+          .where(and(eq(plugins.status, "ready"), or(isNull(pluginCompanySettings.enabled), eq(pluginCompanySettings.enabled, true))));
+        installed = [...configured, ...(await enabledManagedPlugins()).map(plugin => ({ manifest: plugin.manifestJson }))];
+        inspectionFailure = false;
+      } catch (error) { inspectionFailure = true; throw error; }
+      if (disposed || held || isIdleTaskDrainActive()) return;
       const wanted = installed.some(plugin => plugin.manifest.resourceLifecycle?.includes("agent_runtime") && plugin.manifest.capabilities?.includes("events.subscribe"));
-      if (wanted) listening ??= await db.$client.listen("paperclip_plugin_lifecycle", wake, wake);
-      else if (listening) {
-        const subscription = listening;
-        listening = undefined;
-        await subscription.unlisten();
-      }
+      if (!wanted) { await closeListener(); listenerFailure = false; return; }
+      if (listenerFailure) await closeListener();
+      if (disposed || held || isIdleTaskDrainActive()) return;
+      try {
+        listenerClient ??= createOwnedDatabaseListener(db);
+        listening ??= await listenerClient.listen("paperclip_plugin_lifecycle", wake, wake);
+        listenerFailure = false;
+      } catch (error) { listenerFailure = true; throw error; }
     });
-    listenerUpdate = update.catch(() => {});
-    return update;
+  }
+  async function hasJournalWork(current: () => boolean): Promise<IdleLocalWork> {
+    if (!current()) return "unknown";
+    const targets = await deliveryTargets();
+    if (!current()) return "unknown";
+    for (const target of targets) {
+      if (!current()) return "unknown";
+      const types = await deliverableLifecycleTypes(target);
+      if (!current()) return "unknown";
+      if (!types) continue;
+      const { plugin, companyId } = target;
+      const events = await pluginLifecycleInbox(db, plugin.id).list(companyId, 1, undefined, types);
+      if (!current()) return "unknown";
+      if (events.length) { wake(); inspectionFailure = false; return "present"; }
+    }
+    inspectionFailure = false;
+    return "none";
   }
   function pluginChanged() {
     wake();
     void ensureListening().catch(wake);
   }
+  const resumeListening = () => {
+    if (disposed) return;
+    epoch++;
+    if (isIdleTaskDrainActive()) return;
+    held = false;
+    wake();
+    void ensureListening().catch(wake);
+  };
+  const unsubscribeDrain = subscribeTaskDrainChanges(resumeListening);
+  const unsubscribeExpiry = subscribeTaskDrainExpiry(resumeListening);
   const listenerReady = ensureListening();
   const unsubscribeConfig = manager.onConfigApplied(pluginChanged);
   lifecycle.on("plugin.loaded", pluginChanged);
@@ -202,16 +269,54 @@ export function registerPluginLifecycleDelivery(
     ...delivery,
     ready: listenerReady,
     wake,
-    dispose() {
+    async prepareForIdleSleep({ canRetire }: { canRetire: () => boolean }): Promise<IdleLocalWork> {
+      if (disposed || !canRetire() || !isIdleTaskDrainActive() || (listenerFailure && !listenerClient)) return "unknown";
+      if (listenerUpdates || delivering || pending) return "present";
+      const ownerEpoch = epoch;
+      held = true;
+      const current = () => !disposed && held && isIdleTaskDrainActive() && canRetire() && epoch === ownerEpoch;
+      return updateListener(async () => {
+        if (!current()) return "unknown";
+        if (listenerUpdates !== 1 || delivering || pending) return "present";
+        try {
+          const before = await hasJournalWork(current);
+          if (before !== "none") return before;
+          if (!current() || (listenerFailure && !listenerClient)) return "unknown";
+          await closeListener();
+          if (!current()) return "unknown";
+          if (listenerFailure) return "unknown";
+          if (pending || delivering) return "present";
+          return await hasJournalWork(current);
+        } catch {
+          inspectionFailure = true;
+          return "unknown";
+        }
+      });
+    },
+    dispose(): Promise<void> {
+      if (disposePromise) return disposePromise;
       disposed = true;
-      void listenerUpdate.then(() => listening?.unlisten()).catch(() => {});
+      epoch++;
       unsubscribeConfig();
+      unsubscribeDrain();
+      unsubscribeExpiry();
       lifecycle.off("plugin.loaded", pluginChanged);
       lifecycle.off("plugin.enabled", pluginChanged);
       lifecycle.off("plugin.worker_started", pluginChanged);
       lifecycle.off("plugin.status_changed", pluginChanged);
       lifecycle.off("plugin.disabled", pluginChanged);
       lifecycle.off("plugin.unloaded", pluginChanged);
+      disposePromise = (async () => {
+        const failures: unknown[] = [];
+        try { await delivery.stop(); } catch (error) { failures.push(error); }
+        try { await updateListener(closeListener); } catch (error) { failures.push(error); }
+        if (failures.length) {
+          listenerFailure = true;
+          throw failures.length === 1 ? failures[0] : new AggregateError(failures, "Plugin lifecycle delivery disposal failed");
+        }
+        unregisterIdle();
+      })().catch(error => { disposePromise = undefined; throw error; });
+      return disposePromise;
     },
   };
 }

server/src/services/task-admission.ts

@@ -14,6 +14,10 @@ export function subscribeTaskDrainChanges(listener: () => void) {
   changeListeners.add(listener);
   return () => { changeListeners.delete(listener); };
 }
+export function subscribeTaskDrainExpiry(listener: () => void) {
+  expiryListeners.add(listener);
+  return () => { expiryListeners.delete(listener); };
+}
 
 // The shared coordinator owns this process-local deadline and its shutdown.
 // Reading admission can also establish expiry, so both paths notify the same