From c23fd46f1013db9c35c19bc6dbce62455e3975da Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 01:27:26 -0500 Subject: [PATCH] test(server): verify real health scheduling producers Co-Authored-By: Paperclip --- doc/idle-tool-connection-health.md | 27 ++++++++++ ...2026-10-24-recurring-server-sleep-stack.md | 12 +++++ server/src/__tests__/access-service.test.ts | 14 +++-- .../email-channels.integration.test.ts | 27 +++++++++- .../github-connection-events.test.ts | 19 +++++-- .../__tests__/helpers/tool-health-worker.ts | 52 +++++++++++++++++++ 6 files changed, 141 insertions(+), 10 deletions(-) create mode 100644 server/src/__tests__/helpers/tool-health-worker.ts diff --git a/doc/idle-tool-connection-health.md b/doc/idle-tool-connection-health.md index 289399b47ef..148aa7ed808 100644 --- a/doc/idle-tool-connection-health.md +++ b/doc/idle-tool-connection-health.md @@ -125,3 +125,30 @@ log company/connection identifiers and a fixed message, without provider prose. The saved 15-minute deadline still controls the next check. All 64 health/startup cases and server typechecking pass after repair; total selected coverage is now 771 unique tests. Fresh remote gates are required for this repair. + + +Review follow-up: fair round-robin company batching replaces the global oldest-25 +selection. The two-company regression failed before the repair; 19 health-worker +and three shared batching cases pass afterward. Each probe's pre/post eligibility +query already includes the connection ID, so those indexed current-state checks +remain. A scheduled observation still emits the normal transaction signal and +may cause one coalesced company refresh; this deliberately preserves uncertainty +and concurrent-writer recovery rather than introducing a silent-write exception. + +The real-service producer suite now proves AgentMail activation triggers its first +health probe, pause cancels the next deadline, GitHub installation events update +the cached deadline, and member removal cancels a personal connection's deadline. +The GitHub newer-access branch retains its old health deadline, as intended. +These tests use the actual coordinator, transaction notifications, schedule query, +and service mutations; AgentMail also runs the actual health service against its +controlled provider. Subsequent idle windows assert no SQL. + +The first run exposed two incorrect fixture expectations: AgentMail setup leaves +healthCheckedAt null until its first probe, and a stale GitHub installation event +does not overwrite a newer access refresh. The tests now preserve those policies. +A targeted rerun reused AgentMail's retained fixture user and failed setup with a +duplicate key; a fresh disposable database fixes that isolation issue without +changing application behavior or weakening assertions. All 66 access/email/GitHub +cases pass on the final run. Server typecheck/build pass. Cumulative selected +coverage is 776 unique tests; full-suite and fresh remote review/checks remain +acceptance gates. No whole-instance sleep or provider suspension is claimed. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 45e83ec38a1..161012742d6 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1050,3 +1050,15 @@ tests pass (`/tmp/sleep18-fairness-after.log`), after the recorded failing basel AgentMail activation/pause, GitHub installation event, and access-removal producer tests are complete. Then push the owning PR, request fresh review, and merge it into the Discord branch. Full original inventory and earlier unresolved CI remain. + + +Health producer review coverage is now implemented and verified. The actual +AgentMail setup/health/pause, GitHub installation-event, and member-removal paths +move or cancel the live coordinator deadline. All 66 owner-suite cases pass on +isolated PostgreSQL databases, alongside the prior 22 fairness/health cases and +fresh server typecheck/build. Initial fixture expectations and the retained-user +DB setup failure are preserved in the health document. Publish these changes with +the fair company selector, request fresh review, and propagate to Discord #17557. +The Discord initial head has Greptile 5/5; its CI and Paperclip review were still +running when inspected. The remaining original inventory and #17490 failure +remain open; no stack PR is merged or declared fully accepted. diff --git a/server/src/__tests__/access-service.test.ts b/server/src/__tests__/access-service.test.ts index 5a340cbffcf..3eb93daa681 100644 --- a/server/src/__tests__/access-service.test.ts +++ b/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); diff --git a/server/src/__tests__/email-channels.integration.test.ts b/server/src/__tests__/email-channels.integration.test.ts index b22003f5d23..0c430b75266 100644 --- a/server/src/__tests__/email-channels.integration.test.ts +++ b/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 }) => Promise, + beforeSetup?: (input: { companyId: string; fetcher: typeof fetch; wakeup: ReturnType }) => Promise, ) { 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> | 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. diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 7ed7d130232..695ea9be5fa 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/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) { diff --git a/server/src/__tests__/helpers/tool-health-worker.ts b/server/src/__tests__/helpers/tool-health-worker.ts new file mode 100644 index 00000000000..7247d5566ab --- /dev/null +++ b/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[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(); } + }, + }; +}