diff --git a/server/src/__tests__/agent-runtime-hooks.test.ts b/server/src/__tests__/agent-runtime-hooks.test.ts index 3fba2a95759..10a93c5ae64 100644 --- a/server/src/__tests__/agent-runtime-hooks.test.ts +++ b/server/src/__tests__/agent-runtime-hooks.test.ts @@ -12,6 +12,7 @@ import { pluginAgentRuntimeState } from "../services/plugin-agent-runtime-state. import { pluginLifecycleInbox } from "../services/plugin-lifecycle-inbox.js"; import { registerPluginLifecycleDelivery } from "../services/plugin-lifecycle-delivery.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js"; import { agentFileStore } from "../services/agent-file-store.js"; import { agentInstructionRevisionService } from "../services/agent-instruction-revisions.js"; @@ -199,6 +200,13 @@ 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 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 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); @@ -206,15 +214,27 @@ describeDatabase("agent runtime synchronization hooks", () => { await delivery.ready; await vi.waitFor(() => expect(call).toHaveBeenCalled(), { timeout: 5000 }); const delivered = () => call.mock.calls.filter(call => call[2].companyId === companyId); - const initialCount = (await events()).length; - await vi.waitFor(() => expect(delivered()).toHaveLength(initialCount), { timeout: 5000 }); - const before = delivered().length; + await vi.waitFor(async () => expect((await events()).length).toBeGreaterThan(initialCount), { timeout: 5000 }); + await vi.waitFor(async () => expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toHaveLength(0), { timeout: 5000 }); + const waitForIdle = () => vi.waitFor(() => { + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + await waitForIdle(); + await other.execute(sql`SELECT pg_notify('paperclip_plugin_lifecycle', 'startup-drained')`); + await vi.waitFor(() => expect(notifications).toContain("startup-drained"), { timeout: 5000 }); + await waitForIdle(); + let committedEventId: string | undefined; await other.transaction(async tx => { await tx.update(agents).set({ name: "Independent writer" }).where(eq(agents.id, agentId)); - expect(delivered()).toHaveLength(before); + const [event] = await tx.select({ id: resourceLifecycleEvents.id }).from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), eq(resourceLifecycleEvents.resourceType, "agent_runtime"), eq(resourceLifecycleEvents.resourceId, agentId))) + .orderBy(sql`${resourceLifecycleEvents.id} DESC`).limit(1); + committedEventId = String(event.id); + expect(delivered().map(call => call[2].id)).not.toContain(committedEventId); }); - await vi.waitFor(() => expect(delivered().length).toBeGreaterThan(before), { timeout: 5000 }); - expect(delivered().at(-1)![2]).toMatchObject({ companyId, resourceId: agentId, resourceType: "agent_runtime" }); + 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(); } }); it("recovers work committed during real listener connection loss without another write", async () => {