diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index a90cd123991..5b425870898 100644 --- a/doc/idle-sleep-safety.md +++ b/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 diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index d25597c4452..3c455240488 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -260,6 +260,22 @@ export function postgresJsOptions(options: DatabaseClientOptions): Record Db>(); +const ownedDatabaseListenerFactories = new WeakMap Db>(); + +export interface OwnedDatabaseListener { + listen: Db["$client"]["listen"]; + close(): Promise; +} + +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(db: Db, action: (dedicated: Db) => Promise): Promise { 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)); diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 6cab39406a0..c513660e99e 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -1,6 +1,8 @@ export { createDb, withDedicatedDbConnection, + createOwnedDatabaseListener, + type OwnedDatabaseListener, livenessDb, closeLivenessDb, checkpointDb, diff --git a/server/src/__tests__/agent-runtime-assignments.test.ts b/server/src/__tests__/agent-runtime-assignments.test.ts index a804f3d046c..6c9b9bc2550 100644 --- a/server/src/__tests__/agent-runtime-assignments.test.ts +++ b/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 | undefined; const work = change === "bind" ? service.bindProfile(source.id, { targetType: "agent", targetId: agentId, priority: 100 }) diff --git a/server/src/__tests__/agent-runtime-hooks.test.ts b/server/src/__tests__/agent-runtime-hooks.test.ts index b36afba82d5..9ed76d192ab 100644 --- a/server/src/__tests__/agent-runtime-hooks.test.ts +++ b/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(); @@ -287,7 +312,7 @@ describeDatabase("agent runtime synchronization hooks", () => { }); await new Promise(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(resolve => proxy.close(() => resolve())); } diff --git a/server/src/__tests__/plugin-lifecycle-delivery.test.ts b/server/src/__tests__/plugin-lifecycle-delivery.test.ts index 96e1cd7a073..5c1cbaf99a4 100644 --- a/server/src/__tests__/plugin-lifecycle-delivery.test.ts +++ b/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 }>; + const deliveries = new Set>(); + 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["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) { @@ -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(resolve => { entered = resolve; }); + const gate = new Promise(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(resolve => { entered = resolve; }); + const gate = new Promise(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(resolve => { entered = resolve; }); + const gate = new Promise(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(resolve => { entered = resolve; }); + const gate = new Promise(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(resolve => { entered = resolve; }); + const gate = new Promise(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(); }); }); diff --git a/server/src/app.ts b/server/src/app.ts index f978a50459b..41212bea06c 100644 --- a/server/src/app.ts +++ b/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; }; diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index 99f6572dc55..0d7d28c588f 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -135,7 +135,13 @@ function withTaskDrainTransition(run: () => Promise): Promise { return turn; } -export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManager, prepareIdleDatabaseBackup?: () => Promise, deadlineOwners?: IdleDeadlineOwners) { +export function instanceSettingsRoutes( + db: Db, + pluginWorkers?: PluginWorkerManager, + prepareIdleDatabaseBackup?: () => Promise, + 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 }); diff --git a/server/src/services/plugin-lifecycle-delivery.ts b/server/src/services/plugin-lifecycle-delivery.ts index adb8a10c75b..f765865a1ad 100644 --- a/server/src/services/plugin-lifecycle-delivery.ts +++ b/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; @@ -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 | undefined; const retryAt = new Map(); + const unregisterIdle = registerIdleLocalWorkProbe(() => listenerFailure || inspectionFailure ? "unknown" + : listenerUpdates || delivering || listenerClient || pending ? "present" : "none"); const refreshedRuntimeCompanies = new Set(); 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>[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> | undefined; let listenerUpdate = Promise.resolve(); + function updateListener(operation: () => Promise): Promise { + 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 { + 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 { + 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 { + 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; }, }; } diff --git a/server/src/services/task-admission.ts b/server/src/services/task-admission.ts index 46b6d28cf2b..0529e0c823f 100644 --- a/server/src/services/task-admission.ts +++ b/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