perf(server): wake email work from commits and deadlines (#15750)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Email connects agent tasks to durable incoming and outgoing
messages.
> - The email worker scans the database every second, even with no
inboxes.
> - These queries prevent idle database connections from closing.
> - This pull request wakes email work from committed writes and known
deadlines.
> - Empty email installations stop polling. Existing delivery and
catch-up guarantees remain.

## Linked Issues or Issue Description

Refs #15625. Related standby work: #15314.

**What existing behavior does this improve?**

The AgentMail background worker and its database traffic when idle.

**Current behavior**

Every instance queries active email endpoints each second. Active
inboxes also scan delivery queues each second even when no work is due.

**Proposed behavior**

Scan once at startup. Wake after endpoint and queue transactions settle.
Restore the next retry or catch-up deadline from existing rows. An
instance with no active inboxes has no recurring email queries or timer.

**Reason and benefit**

Thousands of empty instances should not keep their databases awake to
discover that email is still unconfigured.

## What Changed

- Use the shared transaction-aware coordinator for email work, including
changes made by separate service objects on the same database owner.
- Signal endpoint activation and control changes, incoming events, and
outgoing sends inside their transactions.
- Add an optional deadline to the coordinator. Email uses persisted
retry times, mailbox catch-up times, and socket reconnect times. Only
the first pending send in each conversation determines its send
deadline.
- Preserve startup recovery, lease fences, idempotency, standby
suppression, idle drain, and shutdown.
- Update the sleep inventory and add quiet-idle, commit, activation,
retry, and lifecycle tests.

## Verification

- Focused coordinator, email idle, and warm standby tests: 20 passed
locally.
- Database-backed email integration suite: 41 passed locally with real
PostgreSQL and a simulated AgentMail provider.
- After review, the updated delayed-reply regression passed separately:
1 selected test passed, 40 intentionally filtered out. It verifies no
scans before the first reply deadline and delivery of both replies in
order.
- `pnpm -r typecheck` and `pnpm build` passed locally.
- The local `pnpm test:run` was interrupted after its checkout was
removed externally. Completed local checks above are preserved. The
[final-commit hosted CI
matrix](https://github.com/paperclipai/paperclip/actions/runs/37997072212)
passed: 54 successful checks, 2 expected Storybook skips, no failures.
It includes the full general and serialized test matrix, runner checks,
browser tests, build, typecheck, and canary packaging.
- Greptile: 5/5 on `ec98f3b86ee476d75e0edf3366b357377ea09d69`, with no
actionable findings.

## Risks

- New email producers must register work inside their write transaction.
No new tables or migrations are required.
- Signals are local to one database owner and its dedicated pools.
Independent processes and direct SQL do not emit them. Startup recovers
existing work; configured inboxes also retain their recovery deadline.
- Active inboxes still catch up each minute. WebSocket mode still renews
its lease every 20 seconds. This change does not provide external waking
or make active sockets sleep.
- Provider calls were simulated in tests. No live email was sent.

## Model Used

OpenAI Codex, GPT-6, with reasoning, repository inspection, code
execution, and test tools. The exact deployment model ID and
context-window size are not exposed in this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/__tests__/cloud-warm-standby.test.ts

@@ -3,6 +3,7 @@ import express from "express";
 import request from "supertest";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { Db } from "@paperclipai/db";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
 import { cloudWarmStandbyMiddleware, cloudWarmStandbyServerOptions } from "../middleware/cloud-warm-standby.js";
 import { healthRoutes } from "../routes/health.js";
 import { emailChannelService } from "../services/email-channels.js";
@@ -124,7 +125,7 @@ describe("unclaimed Cloud background work", () => {
     vi.useFakeTimers();
     let enabled = false;
     const select = vi.fn(() => { throw new Error("SQL probe"); });
-    const db = { select } as unknown as Db;
+    const db = installDatabaseWorkSignals({ select, execute: vi.fn(), transaction: vi.fn() }) as unknown as Db;
     const email = emailChannelService(db, { heartbeat: { wakeup: vi.fn() }, isBackgroundWorkEnabled: () => enabled });
     const scheduler = createPluginJobScheduler({
       db,

server/src/__tests__/delivery-work-coordinator.test.ts

@@ -27,7 +27,7 @@ function setup() {
 function task() { return { retryMs: 5000, run: vi.fn(async () => {}), hasPending: vi.fn(async () => false) }; }
 
 describe("delivery work coordinator", () => {
-  it("leaves five empty queues with no deadlines, timers, or periodic queries", async () => {
+  it("leaves empty queues with no deadlines, timers, or periodic queries", async () => {
     const s = setup();
     const tasks = Object.values(DELIVERY_QUEUES).map(queue => {
       const t = task(); return { queue, t, worker: s.coordinator.register(queue, t) };
@@ -45,6 +45,21 @@ describe("delivery work coordinator", () => {
     expect(vi.getTimerCount()).toBe(0);
     expect(idleWorkSnapshot().active).toBe(0);
   });
+  it("schedules maintenance without holding idle intent, and lets new work preempt it", async () => {
+    const s = setup(), t = task();
+    let next: number | null = Date.now() + 60_000;
+    await s.coordinator.register(DELIVERY_QUEUES.email, { ...t, nextRunAt: () => next }).ready;
+    await vi.advanceTimersByTimeAsync(59_000);
+    expect(t.run).toHaveBeenCalledTimes(1);
+    await s.enqueue(DELIVERY_QUEUES.email);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(t.run).toHaveBeenCalledTimes(2);
+    expect(idleWorkSnapshot().active).toBe(0);
+    next = null;
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(t.run).toHaveBeenCalledTimes(3);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+  });
   it("does not postpone an already committed wake when another producer starts writing", async () => {
     const s = setup(), t = task();
     await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;

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

@@ -199,7 +199,9 @@ describe("AgentMail durable email pipeline", () => {
     expect(await db.select().from(authUsers).where(eq(authUsers.id, "email-board"))).toHaveLength(1);
   });
 
-  async function fixture(mode: "websocket" | "webhook" = "webhook", storage?: StorageService) {
+  async function fixture(mode: "websocket" | "webhook" = "webhook", storage?: StorageService,
+    beforeSetup?: (input: { fetcher: typeof fetch; wakeup: ReturnType<typeof vi.fn> }) => Promise<void>,
+  ) {
     const companyId = randomUUID(),
       agentId = randomUUID(),
       endpointId = randomUUID();
@@ -312,6 +314,7 @@ describe("AgentMail durable email pipeline", () => {
       storage,
     });
     services.push(service);
+    await beforeSetup?.({ fetcher, wakeup });
     await service.setup(
       companyId,
       {
@@ -374,6 +377,73 @@ describe("AgentMail durable email pipeline", () => {
       setWebhookError: (status: number) => { webhookError = status; },
     };
   }
+  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.
+    let running!: EmailChannelService;
+    const f = await fixture("webhook", undefined, async ({ fetcher, wakeup }) => {
+      running = emailChannelService(db, { heartbeat: { wakeup }, fetch: fetcher });
+      services.push(running);
+      await running.start();
+      const select = vi.spyOn(db, "select");
+      await new Promise(resolve => setTimeout(resolve, 1200));
+      expect(select).not.toHaveBeenCalled();
+      select.mockRestore();
+    });
+    await vi.waitFor(async () => {
+      const [config] = await db.select().from(emailEndpoints).where(eq(emailEndpoints.endpointId, f.endpointId));
+      expect(config.lastSyncAt).not.toBeNull();
+    }, { timeout: 5000 });
+    await running.tick();
+    await f.service.control(f.endpointId, "pause", { userId: "email-board" });
+    await f.service.control(f.endpointId, "resume", { userId: "email-board" });
+    const message = f.message();
+    f.messages.set(message.message_id, message);
+    await f.service.admit(await f.service.getEndpoint(f.endpointId), {
+      event_type: "message.received", message,
+    });
+    await vi.waitFor(() => expect(f.wakeup).toHaveBeenCalledTimes(1), { timeout: 5000 });
+    const [parent] = await db.select().from(issues).where(eq(issues.companyId, f.companyId));
+    const input = emailSendSchema.parse({ endpointId: f.endpointId, parentIssueId: parent.id,
+      to: ["recipient@example.test"], subject: "Wake on send", text: "Hello", idempotencyKey: randomUUID() });
+    await f.service.queueSend(f.companyId, input, { userId: "email-board" });
+    await vi.waitFor(async () => expect((await f.service.publication(input.idempotencyKey, f.companyId)).outcome).toBe("sent"), { timeout: 5000 });
+    expect(f.sends).toHaveLength(1);
+    await running.tick();
+    await f.service.control(f.endpointId, "pause", { userId: "email-board" });
+    await new Promise(resolve => setTimeout(resolve, 100));
+    const select = vi.spyOn(db, "select");
+    await new Promise(resolve => setTimeout(resolve, 1200));
+    expect(select).not.toHaveBeenCalled();
+  });
+
+  it("restores a retry deadline without polling replies blocked behind that send", async () => {
+    const f = await fixture();
+    const message = f.message("received", "outbound-thread");
+    await f.receive(message);
+    const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.endpointId, f.endpointId));
+    const input = emailSendSchema.parse({ endpointId: f.endpointId, conversationId: conversation.id,
+      replyToMessageId: message.message_id, text: "First reply", idempotencyKey: randomUUID() });
+    await f.service.queueSend(f.companyId, input, { userId: "email-board" });
+    const second = { ...input, text: "Second reply", idempotencyKey: randomUUID() };
+    await f.service.queueSend(f.companyId, second, { userId: "email-board" });
+    // Simulate a previous process persisting a provider retry deadline. The
+    // second reply has no deadline but cannot overtake this first reply.
+    await db.update(chatPublications).set({ state: "retry", nextAttemptAt: new Date(Date.now() + 5000) })
+      .where(eq(chatPublications.id, input.idempotencyKey));
+    await f.service.start();
+    expect(f.sends).toHaveLength(0);
+    const select = vi.spyOn(db, "select");
+    await new Promise(resolve => setTimeout(resolve, 1100));
+    expect(select).not.toHaveBeenCalled();
+    expect(f.sends).toHaveLength(0);
+    await vi.waitFor(() => expect(f.sends).toHaveLength(2), { timeout: 8000 });
+    await f.service.tick();
+    expect(f.sends.map(send => send.key)).toEqual([input.idempotencyKey, second.idempotencyKey]);
+    for (const id of [input.idempotencyKey, second.idempotencyKey])
+      expect((await f.service.publication(id, f.companyId)).outcome).toBe("sent");
+  });
+
   it("admits signed webhooks through the durable queue and rejects a valid signature for another inbox", async () => {
     const f = await fixture();
     const endpoint = await f.service.getEndpoint(f.endpointId);

server/src/__tests__/email-idle-work.test.ts

@@ -0,0 +1,81 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import type { Db } from "@paperclipai/db";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { emailChannelService } from "../services/email-channels.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const services: ReturnType<typeof emailChannelService>[] = [];
+afterEach(async () => {
+  for (const service of services.splice(0)) await service.shutdown();
+  vi.useRealTimers();
+});
+function fixture() {
+  vi.useFakeTimers();
+  const select = vi.fn(() => ({ from: () => ({ where: async () => [] }) }));
+  const execute = vi.fn(async () => [{ xid: "42", status: "aborted" }]);
+  const db = installDatabaseWorkSignals({ select, execute,
+    async transaction(callback: (tx: any) => Promise<any>) {
+      return callback({ execute, transaction: db.transaction });
+    },
+  });
+  let enabled = true;
+  const service = emailChannelService(db as unknown as Db, {
+    heartbeat: { wakeup: vi.fn() },
+    isBackgroundWorkEnabled: () => enabled,
+    isReconciliationEnabled: () => true,
+  });
+  services.push(service);
+  return { db, select, service, enable: (value: boolean) => { enabled = value; } };
+}
+
+describe("email idle work", () => {
+  it("scans at startup, then leaves an unconfigured stack with zero timers and queries", async () => {
+    const f = fixture();
+    await f.service.start();
+    expect(f.select).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(3_600_000);
+    expect(f.select).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("wakes on an outer transaction commit from a separate producer, never an inner savepoint", async () => {
+    const f = fixture();
+    await f.service.start();
+    await f.db.transaction(async tx => {
+      await tx.transaction(async (inner: object) => notifyDeliveryWork(inner, DELIVERY_QUEUES.email));
+      expect(f.select).toHaveBeenCalledTimes(1);
+    });
+    await vi.advanceTimersByTimeAsync(1);
+    expect(f.select).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("retains startup and committed work across idle drain without SQL or dispatch", async () => {
+    const f = fixture();
+    f.enable(false);
+    await f.service.start();
+    await f.db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.email));
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(f.select).not.toHaveBeenCalled();
+    expect(idleWorkSnapshot().active).toBe(0);
+    f.enable(true);
+    await vi.advanceTimersByTimeAsync(1001);
+    expect(f.select).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("retries a failed startup scan and unsubscribes on shutdown", async () => {
+    const f = fixture();
+    const warning = vi.spyOn(process, "emitWarning").mockImplementation(() => {});
+    f.select.mockImplementationOnce(() => { throw new Error("database unavailable"); });
+    await f.service.start();
+    await vi.advanceTimersByTimeAsync(1001);
+    expect(f.select).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+    await f.service.shutdown();
+    await f.db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.email));
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(f.select).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+    warning.mockRestore();
+  });
+});

server/src/app.ts

@@ -646,6 +646,7 @@ export async function createApp(
   // mutation guard. The Chat SDK adapter verifies the provider signature
   // before Paperclip persists or acts on any event.
   const emailChannels = emailChannelService(db, {
+    isReconciliationEnabled: () => !isWarmStandby(),
     isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive(),
     heartbeat: connectionIntentHeartbeat,
     storage: opts.storageService,

server/src/services/delivery-work-coordinator.ts

@@ -22,6 +22,8 @@ export function createDeliveryWorkCoordinator(input: {
       retryMs: number;
       run: (signal: AbortSignal) => Promise<unknown>;
       hasPending: () => Promise<boolean>;
+      // Optional maintenance/retry deadline restored from durable state.
+      nextRunAt?: () => number | null;
     }) {
       if (stopped) throw new Error("Delivery coordinator is stopped");
       if (workers.has(queue)) throw new Error(`Delivery worker already registered: ${queue}`);
@@ -51,6 +53,8 @@ export function createDeliveryWorkCoordinator(input: {
           .then(() => workerStopped || !input.canRun() ? undefined : task.run(attempt.signal)).then(() => workerStopped ? false : task.hasPending()).then(pending => {
           if (pending || databaseWorkPending(input.owner, queue)) schedule(task.retryMs);
           else { finishIntent?.(); finishIntent = undefined; }
+          const next = task.nextRunAt?.();
+          if (next != null) schedule(Math.max(0, next - Date.now()));
         }).catch(error => {
           schedule(task.retryMs);
           // Logging must never turn a recoverable sweep failure into an

server/src/services/delivery-work-notifications.ts

@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  email: "email-channels",
   feedback: "feedback-exports",
   chatCompletion: "chat-completions",
   connection: "connection-continuations",

server/src/services/email-channels.ts

@@ -1,3 +1,5 @@
+import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { HttpError } from "../errors.js";
 import { createHash, randomUUID } from "node:crypto";
 import WebSocket from "ws";
@@ -77,6 +79,8 @@ type Tx = Parameters<Parameters<Db["transaction"]>[0]>[0];
 export interface EmailChannelOptions {
   /** Suppress periodic database work while an unclaimed Cloud app stands by. */
   isBackgroundWorkEnabled?: () => boolean;
+  /** Allow transaction outcome checks during idle drain, but not warm standby. */
+  isReconciliationEnabled?: () => boolean;
   heartbeat: Pick<ReturnType<typeof heartbeatService>, "wakeup">;
   storage?: StorageService;
   publicBaseUrl?: string;
@@ -130,7 +134,12 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
   let stopped = false;
   let ticking = false;
   let activeTick: Promise<void> | null = null;
-  let timer: ReturnType<typeof setInterval> | undefined;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let worker: ReturnType<ReturnType<typeof createDeliveryWorkCoordinator>["register"]> | undefined;
+  let nextWorkAt: number | null = null;
+  function needWorkAt(at: number) {
+    nextWorkAt = Math.min(nextWorkAt ?? Infinity, Math.max(Date.now() + 1000, at));
+  }
 
   async function getEndpoint(id: string) {
     const [row] = await db
@@ -881,6 +890,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
       }
       const now = new Date();
       await db.transaction(async (tx) => {
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.email);
         await tx
           .update(emailEndpoints)
           .set({
@@ -909,7 +919,6 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
       return summary(await getEndpoint(endpoint.id));
     });
     if (!result) throw conflict("Inbox setup is already running");
-    if (timer) void tick().catch(() => {});
     return result;
   }
 
@@ -919,22 +928,24 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
     if (!event) return;
     if (event.inbox_id !== endpoint.botExternalId)
       throw forbidden("Email event belongs to a different inbox");
-    await db
-      .insert(chatDeliveries)
-      .values({
-        companyId: endpoint.companyId,
-        endpointId: endpoint.id,
-        providerEventId: event.eventId,
-        deduplicationKey: `${event.kind}:${event.message_id}`,
-        eventKind: "message",
-        normalizedEvent: event,
-      })
-      .onConflictDoNothing();
-    await db
-      .update(chatEndpoints)
-      .set({ lastEventAt: new Date() })
-      .where(eq(chatEndpoints.id, endpoint.id));
-    if (timer) void tick().catch(() => {});
+    await db.transaction(async (tx) => {
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.email);
+      await tx
+        .insert(chatDeliveries)
+        .values({
+          companyId: endpoint.companyId,
+          endpointId: endpoint.id,
+          providerEventId: event.eventId,
+          deduplicationKey: `${event.kind}:${event.message_id}`,
+          eventKind: "message",
+          normalizedEvent: event,
+        })
+        .onConflictDoNothing();
+      await tx
+        .update(chatEndpoints)
+        .set({ lastEventAt: new Date() })
+        .where(eq(chatEndpoints.id, endpoint.id));
+    });
   }
   async function webhook(
     publicId: string,
@@ -1550,6 +1561,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
         if (!reply)
           throw badRequest("Reply target is not in this inbox conversation");
       }
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.email);
       await tx.insert(chatPublications).values({
         id: input.idempotencyKey,
         companyId,
@@ -1571,7 +1583,6 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
     await audit(endpoint, "email.queued", actor, {
       publicationId: input.idempotencyKey,
     });
-    if (timer) void tick().catch(() => {});
     return publication(input.idempotencyKey, companyId);
   }
   async function publication(
@@ -1968,6 +1979,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
             .update(emailEndpoints)
             .set({ lastSyncAt: null })
             .where(eq(emailEndpoints.endpointId, endpoint.id));
+          worker?.wake();
         } else await admit(endpoint, value);
       })().catch(async () => {
         await markError(
@@ -1988,6 +2000,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
       const delay = Math.min(60_000, (backoff.get(endpoint.id) ?? 1000) * 2);
       backoff.set(endpoint.id, delay);
       reconnectAt.set(endpoint.id, Date.now() + delay);
+      worker?.wake();
       void db
         .delete(chatEndpointLeases)
         .where(
@@ -2056,6 +2069,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
   async function runTick() {
     if (ticking || stopped) return;
     ticking = true;
+    nextWorkAt = null;
     try {
       const endpoints = await db
         .select()
@@ -2096,7 +2110,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
                   ne(chatPublications.state, "delivery_unknown"),
                 ),
               )
-              .orderBy(asc(chatPublications.createdAt))
+              .orderBy(asc(chatPublications.createdAt), asc(chatPublications.id))
               .limit(25);
             // Each conversation is serial, while independent inbox threads can make progress.
             const firstSends = [
@@ -2199,7 +2213,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
             );
             if (
               !config.lastSyncAt ||
-              Date.now() - config.lastSyncAt.getTime() > 60_000
+              Date.now() - config.lastSyncAt.getTime() >= 60_000
             )
               await withLease(
                 endpoint,
@@ -2208,7 +2222,40 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
                 },
                 "email-catchup",
               );
+            // Catch-up is a real scheduled obligation; an empty queue is not.
+            const refreshed = await getConfig(endpoint.id);
+            needWorkAt((refreshed.lastSyncAt?.getTime() ?? 0) + 60_000);
+            if (refreshed.receiveMode === "websocket" && !sockets.has(endpoint.id))
+              needWorkAt(reconnectAt.get(endpoint.id) ?? Date.now() + 1000);
+            const [inbound] = await db
+              .select({ at: sql<Date | null>`min(coalesce(${chatDeliveries.nextAttemptAt}, now()))` })
+              .from(chatDeliveries)
+              .where(and(
+                eq(chatDeliveries.endpointId, endpoint.id),
+                inArray(chatDeliveries.state, ["received", "processing", "retry"]),
+              ));
+            // A later reply cannot run ahead of its conversation's first send.
+            // Its null deadline must not turn a delayed retry into 1s polling.
+            const sendHeads = db
+              .selectDistinctOn([chatPublications.conversationId], {
+                nextAttemptAt: chatPublications.nextAttemptAt,
+              })
+              .from(emailSends)
+              .innerJoin(chatPublications, eq(chatPublications.id, emailSends.publicationId))
+              .where(and(
+                eq(emailSends.endpointId, endpoint.id),
+                inArray(emailSends.outcome, ["queued", "uncertain"]),
+                ne(chatPublications.state, "delivery_unknown"),
+              ))
+              .orderBy(asc(chatPublications.conversationId), asc(chatPublications.createdAt), asc(chatPublications.id))
+              .as("email_send_heads");
+            const [outbound] = await db
+              .select({ at: sql<Date | null>`min(coalesce(${sendHeads.nextAttemptAt}, now()))` })
+              .from(sendHeads);
+            for (const pending of [inbound, outbound])
+              if (pending?.at) needWorkAt(new Date(pending.at).getTime());
           } catch (e) {
+            needWorkAt(Date.now() + 1000);
             await markError(endpoint, diagnostic(e));
           }
         }),
@@ -2219,6 +2266,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
   }
   async function stopEndpoint(endpoint: Endpoint) {
     await db.transaction(async (tx) => {
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.email);
       await tx
         .select()
         .from(chatEndpoints)
@@ -2319,6 +2367,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
           await removeUnusedSecret(secretId);
       }
       await db.transaction(async (tx) => {
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.email);
         await tx
           .update(chatEndpoints)
           .set({
@@ -2642,17 +2691,28 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) {
       };
     },
     start: () => {
-      if (!timer) {
-        timer = setInterval(() => {
-          void tick().catch(() => {});
-        }, 1000);
-        timer.unref();
-        void tick().catch(() => {});
-      }
+      if (coordinator || stopped) return;
+      coordinator = createDeliveryWorkCoordinator({
+        owner: db,
+        canRun: () => options.isBackgroundWorkEnabled?.() !== false,
+        canReconcile: () => options.isReconciliationEnabled?.() ?? (options.isBackgroundWorkEnabled?.() !== false),
+        onError: (error) => process.emitWarning(`Email background work failed: ${diagnostic(error)}`),
+      });
+      worker = coordinator.register(DELIVERY_QUEUES.email, {
+        retryMs: 1000,
+        run: tick,
+        // Retries and maintenance have deadlines; only unresolved writes need
+        // the coordinator's pending-intent hold, not a configured idle inbox.
+        hasPending: async () => false,
+        nextRunAt: () => options.isBackgroundWorkEnabled?.() === false
+          ? Date.now() + 1000
+          : nextWorkAt,
+      });
+      return worker.ready;
     },
     shutdown: async () => {
       stopped = true;
-      clearInterval(timer);
+      await coordinator?.stop();
       await activeTick;
       const heldSockets = [...sockets.entries()];
       sockets.clear();