Merge health scheduling review repairs into Discord scheduling

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

doc/idle-tool-connection-health.md

@@ -21,9 +21,13 @@ company status are not new filters: this migration preserves the old policy.
 An unchecked row is due immediately. Otherwise its deadline is `healthCheckedAt`
 plus 15 minutes. A company hint refreshes that company's cache. Startup and an
 instance feature-settings change can refresh the whole instance. Refreshes are
-coalesced with a 30-second minimum spacing. There is no scheduled refresh when
+coalesced with a 30-second minimum spacing when no due backlog remains.
+Batches retain at most 25 checks and continue after one second while already due
+rows remain, so the batch limit does not cap dispatch at 50 checks per minute. There is no scheduled refresh when
 there are no eligible rows or unresolved changes. Future checks cause no SQL
-before their deadline. Each pass checks at most 25 due connections.
+before their deadline. Each pass checks at most 25 due connections, taking turns
+across companies and preserving due order within each company. The company cursor
+rotates across passes when more companies are due than fit in one batch.
 
 Before a provider check, the service reloads current eligibility and deadline.
 A saved health result, including a failed health observation, owns the next
@@ -123,3 +127,40 @@ 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.
+
+
+The Discord child review also exposed fixed-batch throughput as a risk for health.
+A probe-timestamp regression measured a 30.7-second gap before the 26th already
+owed check. The worker now schedules a one-second continuation only when the due
+selection exceeds its bounded batch. Failed work retains backoff; an empty or
+fully future-dated inventory has no continuation. This is scheduling concrete
+backlog, not a periodic idle scan. The initial deadline-only assertion passed
+incorrectly because a pending transaction notification temporarily reported an
+immediate coordinator wake; acceptance measures actual probe timestamps instead.

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -1069,3 +1069,53 @@ follow-ups for cross-company backlog fairness and direct service-producer tests.
 These remain owning-PR work despite green Actions and Greptile 5/5. The same
 fairness defect was reproduced and repaired in the Discord worker. Shared helper
 commit `e77979d4fc` is ready to carry into #17554. No PR is newly accepted or merged.
+
+
+## Discord publication and health fairness repair
+
+Discord registration is [#17557](https://github.com/paperclipai/paperclip/pull/17557),
+head `3755d7caced67589de474e0ca3e1b8b5b4e659c4`, based on #17554 and attached.
+It has 639 selected passing tests, workspace typecheck/build, and final server
+verification. CI run **40168371442** and review run **40168371311** are confirmed
+live on that head. Fresh CI and review remain required; no merge is authorized.
+The source and detailed producer audit are on the Discord child branch.
+
+The health parent Paperclip Review completed neutral with 4/5 at `da40899606`.
+It requests fair company batching and direct real-service producer coverage.
+The existing probes are already scoped by connection ID, so the review's concern
+about full inventory scans around each probe needs clarification rather than
+removing current-state eligibility checks. Self-generated observation hints still
+cause a bounded coalesced company refresh; avoid muting correctness signals merely
+to reduce that bounded work without proving equivalent recovery.
+
+Fairness was reproduced with 30 unchecked connections in one company and one due
+connection in another. The quiet company was omitted from the first batch. The
+shared selector from Discord commit `e77979d4fc` was applied to the health branch
+as `097f5c8aff`; the health worker now rotates companies. All 22 health/selection
+tests pass (`/tmp/sleep18-fairness-after.log`), after the recorded failing baseline
+(`/tmp/sleep18-fairness-before.log`). The fix is local until the additional
+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.
+
+
+Discord's first Paperclip review (4/5) identified a dispatch-capacity issue shared
+by health: 25 checks per 30 seconds delays a large already-due backlog. Health's
+actual-probe timestamp regression failed with a 30.7-second gap, then passed when
+only a remaining due batch receives a one-second continuation. All 22 health and
+batching cases pass; server types pass. Propagate the parent and apply the same
+bounded continuation to Discord, with its own regression. The Discord review also
+requests preserving queued scopes after malformed hints and broader producer
+coverage. Its initial head has Greptile 5/5 but these findings remain open.

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(); }
+    },
+  };
+}

server/src/__tests__/tool-connection-health-work.test.ts

@@ -136,7 +136,10 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   });
   it("drains more than one due page without repeatedly checking completed rows", async () => {
     for (let i = 0; i < 30; i++) await connection();
+    const probedAt: number[] = [], performProbe = probe.getMockImplementation()!;
+    probe.mockImplementation(async (id: string) => { probedAt.push(Date.now()); return performProbe(id); });
     await start(); await checked(30);
+    expect(probedAt[25] - probedAt[24]).toBeLessThan(5_000);
     expect(new Set(probe.mock.calls.map(([id]) => id)).size).toBe(30);
   });
   it("checks the first connection created by the real API service after setup commits", async () => {
@@ -164,6 +167,25 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await tools.archiveConnection(row.id, companyId); await quiet();
   });
 
+  it("checks a quiet company in the first batch despite another company's unchecked backlog", async () => {
+    for (let index = 0; index < 30; index++) await connection({}, false);
+    const otherId = randomUUID();
+    await db.insert(companies).values({ id: otherId, name: "Quiet health", issuePrefix: `Q${otherId.slice(0, 6)}` });
+    const [app] = await db.insert(toolApplications).values({ companyId: otherId, name: "Quiet health", type: "mcp_http" }).returning();
+    try {
+      const [other] = await db.insert(toolConnections).values({
+        companyId: otherId, applicationId: app.id, name: "Quiet", uid: randomUUID(), transport: "mcp_remote",
+        status: "active", enabled: true, healthCheckedAt: new Date(Date.now() - 901_000),
+      }).returning();
+      await start();
+      expect(probe).toHaveBeenCalledTimes(25);
+      expect(probe.mock.calls.map(([id]) => id)).toContain(other.id);
+    } finally {
+      await coordinator?.stop(); coordinator = undefined;
+      await db.delete(companies).where(eq(companies.id, otherId));
+    }
+  });
+
   it("continues another company's checks when one scope cannot refresh", async () => {
     const service = connectionHealthSchedule(db, probe), load = service.load;
     await start(service); await quiet();

server/src/services/tool-connection-health-work.ts

@@ -1,3 +1,4 @@
+import { takeCompanyWorkBatch } from "./company-work-batch.js";
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import type { ConnectionHealthDeadline } from "./tool-access/catalog/health-schedule.js";
@@ -15,6 +16,7 @@ export function registerToolConnectionHealthWork(
   const failures = new Map<string, { count: number; retryAt: number }>();
   const retryMs = 30_000;
   let earliest = 0;
+  let lastCompany: string | null = null;
   function request(hints?: ReadonlySet<string>) {
     for (const hint of hints ?? ["*"]) dirty.add(hint);
   }
@@ -52,8 +54,10 @@ export function registerToolConnectionHealthWork(
       }
       const due = [...deadlines.values()].filter(entry => !unavailable.has("*") && !unavailable.has(entry.companyId)
         && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())
-        .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, 25);
-      for (const entry of due) {
+        .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id));
+      const batch = takeCompanyWorkBatch(due, 25, lastCompany);
+      lastCompany = batch.lastCompany;
+      for (const entry of batch.entries) {
         if (signal.aborted || !canRun()) break;
         try {
           const current = await service.check(entry.id);
@@ -65,6 +69,7 @@ export function registerToolConnectionHealthWork(
           errors.push(error);
         }
       }
+      if (due.length > batch.entries.length) earliest = Date.now() + 1_000;
       if (errors.length) throw new AggregateError(errors, "Tool connection health checks could not settle");
     },
     hasPending: async () => dirty.size > 0,