test(server): verify real health scheduling producers

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/access-service.test.ts

@@ -1,3 +1,4 @@
+import { startToolHealthWorker } from "./helpers/tool-health-worker.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, inArray, sql } from "drizzle-orm";
@@ -345,6 +346,7 @@ describeEmbeddedPostgres("access service", () => {
       credentialPolicy: "per_user",
       status: "active",
       enabled: true,
+      healthCheckedAt: new Date(),
     }).returning().then((rows) => rows[0]!);
     const definition = await db.insert(userSecretDefinitions).values({
       companyId: company.id,
@@ -375,9 +377,15 @@ describeEmbeddedPostgres("access service", () => {
       agentId: agent.id,
       createdByUserId: member.principalId,
     });
-    await accessService(db).archiveMember(company.id, member.id, {
-      reassignment: { assigneeUserId: owner.principalId },
-    });
+    const health = await startToolHealthWorker(db, company.id);
+    try {
+      await health.expectDeadline(connection.healthCheckedAt!.getTime() + 900_000);
+      await accessService(db).archiveMember(company.id, member.id, {
+        reassignment: { assigneeUserId: owner.principalId },
+      });
+      await health.expectDeadline(null);
+      await health.expectQuiet();
+    } finally { await health.stop(); }
 
     expect(await db.select().from(connectionGrantDelegations)).toHaveLength(0);
     expect(await db.select().from(connectionGrantMembers)).toHaveLength(0);

server/src/__tests__/email-channels.integration.test.ts

@@ -1,3 +1,4 @@
+import { startToolHealthWorker } from "./helpers/tool-health-worker.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { aiConnectionService } from "../services/ai-connections.js";
 import { fastResponseService } from "../services/fast-responses.js";
@@ -214,7 +215,7 @@ describe("AgentMail durable email pipeline", () => {
   });
 
   async function fixture(mode: "websocket" | "webhook" = "webhook", storage?: StorageService,
-    beforeSetup?: (input: { fetcher: typeof fetch; wakeup: ReturnType<typeof vi.fn> }) => Promise<void>,
+    beforeSetup?: (input: { companyId: string; fetcher: typeof fetch; wakeup: ReturnType<typeof vi.fn> }) => Promise<void>,
   ) {
     const companyId = randomUUID(),
       agentId = randomUUID(),
@@ -328,7 +329,7 @@ describe("AgentMail durable email pipeline", () => {
       storage,
     });
     services.push(service);
-    await beforeSetup?.({ fetcher, wakeup });
+    await beforeSetup?.({ companyId, fetcher, wakeup });
     await service.setup(
       companyId,
       {
@@ -391,6 +392,28 @@ describe("AgentMail durable email pipeline", () => {
       setWebhookError: (status: number) => { webhookError = status; },
     };
   }
+  it("schedules health after AgentMail activation and disarms it on pause", async () => {
+    let health: Awaited<ReturnType<typeof startToolHealthWorker>> | undefined;
+    try {
+      const f = await fixture("webhook", undefined, async ({ companyId, fetcher }) => {
+        vi.spyOn(globalThis, "fetch").mockImplementation(fetcher);
+        health = await startToolHealthWorker(db, companyId, new Date(), toolAccessService(db).checkHealth);
+        await health.expectDeadline(null);
+        await health.expectQuiet();
+      });
+      const endpoint = await f.service.getEndpoint(f.endpointId);
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(health!.checkHealth).toHaveBeenCalledTimes(1);
+        expect(health!.checkHealth.mock.settledResults[0]?.type).toBe("fulfilled");
+      }, { timeout: 5_000 });
+      const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, endpoint.connectionId));
+      await health!.expectDeadline(connection.healthCheckedAt!.getTime() + 900_000);
+      await f.service.control(f.endpointId, "pause", { userId: "email-board" });
+      await health!.expectDeadline(null);
+      await health!.expectQuiet();
+    } finally { await health?.stop(); }
+  });
   it("activates an empty running worker from another service and drains committed inbound and outbound work", async () => {
     // Connector setup and tool calls construct short-lived service objects.
     // Their committed writes must wake the app-owned worker sharing this DB.

server/src/__tests__/github-connection-events.test.ts

@@ -1,3 +1,4 @@
+import { startToolHealthWorker } from "./helpers/tool-health-worker.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
@@ -253,6 +254,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
       credentialPolicy: "per_user",
       status: "active",
       enabled: true,
+      healthCheckedAt: new Date("2026-09-04T11:59:05.000Z"),
       config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } },
       transportConfig: {},
     });
@@ -318,11 +320,18 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
     let currentTime = new Date("2026-09-04T12:00:05.000Z");
     const service = githubConnectionEventService(db, { connector, now: () => currentTime });
 
-    const unsubscribe = subscribeCompanyLiveEvents(companyId, () => {
-      throw new Error("fixture live subscriber failed");
-    });
-    await expect(service.pollOnce()).resolves.toMatchObject({ processed: 1, duplicate: 0, failed: 0 });
-    unsubscribe();
+    const health = await startToolHealthWorker(db, companyId, currentTime);
+    try {
+      await health.expectDeadline(new Date("2026-09-04T12:14:05.000Z").getTime());
+      const unsubscribe = subscribeCompanyLiveEvents(companyId, () => {
+        throw new Error("fixture live subscriber failed");
+      });
+      try {
+        await expect(service.pollOnce()).resolves.toMatchObject({ processed: 1, duplicate: 0, failed: 0 });
+      } finally { unsubscribe(); }
+      await health.expectDeadline(currentTime.getTime() + 900_000 - (refreshed ? 60_000 : 0));
+      await health.expectQuiet();
+    } finally { await health.stop(); }
     let [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grantId));
     expect(grant?.providerTenant?.github).toMatchObject({ repositoryCount: refreshed ? 3 : 4, webhookHealth: "healthy" });
     if (refreshed) {

server/src/__tests__/helpers/tool-health-worker.ts

@@ -0,0 +1,52 @@
+import type { Db } from "@paperclipai/db";
+import { expect, vi } from "vitest";
+import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.js";
+import { connectionHealthSchedule } from "../../services/tool-access/catalog/health-schedule.js";
+import { registerToolConnectionHealthWork } from "../../services/tool-connection-health-work.js";
+import { idleWorkSnapshot } from "../../services/task-admission.js";
+
+export async function startToolHealthWorker(db: Db, companyId: string, now = new Date(), probe?: Parameters<typeof connectionHealthSchedule>[1]) {
+  vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+  vi.setSystemTime(now);
+  const errors: unknown[] = [];
+  const checkHealth = vi.fn(probe ?? (async () => { throw new Error("Unexpected due health probe"); }));
+  const schedule = connectionHealthSchedule(db, checkHealth);
+  const coordinator = createDeliveryWorkCoordinator({
+    owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error),
+  });
+  try {
+    await registerToolConnectionHealthWork(coordinator, {
+      load: scope => schedule.load(scope ?? companyId), check: schedule.check,
+    }, () => true).ready;
+  } catch (error) {
+    await coordinator.stop();
+    vi.useRealTimers();
+    throw error;
+  }
+  return {
+    checkHealth,
+    async expectDeadline(expected: number | null) {
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(idleWorkSnapshot().active).toBe(0);
+        expect(coordinator.nextWakeAt()).toBe(expected);
+      }, { timeout: 5_000 });
+      expect(errors).toEqual([]);
+      if (!probe) expect(checkHealth).not.toHaveBeenCalled();
+    },
+    async expectQuiet() {
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled();
+        expect(writes).not.toHaveBeenCalled();
+        expect(executes).not.toHaveBeenCalled();
+      } finally {
+        reads.mockRestore(); writes.mockRestore(); executes.mockRestore();
+      }
+    },
+    async stop() {
+      try { await coordinator.stop(); } finally { vi.useRealTimers(); }
+    },
+  };
+}