perf(chat): wake action queues from commits and deadlines (#15808)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Chat keeps durable rows for provider actions and setup messages.
> - Five action queues still scan the database every second when empty.
> - Empty scans add idle database traffic and prevent cost savings from
sleep.
> - This pull request wakes these queues when their writes commit or
their next deadline arrives.
> - Existing rows retain retry and restart recovery. Empty queues stop
scanning.

## Linked Issues or Issue Description

Refs #15802, #15625, #15750, #15757. Related allocation work: #15153.

**What existing behavior does this improve?**

Idle chat reconciliation for Slack Board messages, Slack slash-command
task starts, Slack setup verification messages, receipt reactions, and
Slack session stops.

**Current behavior**

The app runs these five queue scans through chat maintenance every
second, including when no work exists.

**Proposed behavior**

Each queue has an independent worker on the existing scheduler.
Producers and foreground claim/retry/settlement writes announce work
inside their database transaction. The worker wakes after the outer
transaction settles. Startup restores durable work. Shared dispatch and
deadline predicates select real retries and abandoned claims. Empty
queues disarm.

**Subsystem affected**

Server chat services. No schema or migration changes.

## What Changed

- Register five action workers alongside the three delivery workers from
#15802. Remove their scans from periodic app maintenance.
- Share action eligibility between dispatch and deadline selection. Park
paused Board messages and task starts until endpoint changes. Preserve
uncertain-send quarantine.
- Wake setup verification after account, welcome, registration, or
authenticated callback changes. Exclude prerequisites that need another
event.
- Persist a retry time for failed provider-confirmed task admission. Its
own commit notification cannot cause an immediate retry loop.
- Reuse the existing transaction signals and scheduler registration
code. Provider calls remain outside database transactions.
- Update the instance-sleep inventory and add scheduler and PostgreSQL
regression coverage.

## Verification

- Passed: 37 scheduler, coordinator, and app-wiring tests, including an
idle hour with no further scans or timers.
- Passed: seven focused PostgreSQL chat cases. These cover deadline
eligibility, startup recovery, outer-commit wakes, actual producer
signals, and signed-callback verification without maintenance polling.
- Passed: `pnpm typecheck`.
- Passed: full chat integration suite and related
publication/coordinator tests, **1,278 tests** across six files,
including **1,228 chat integration cases**.
- Passed: `pnpm build`.
- After the review fix, seven focused PostgreSQL integration cases and
the server typecheck passed again. Deadline and dispatch queries now
exclude completed action history.
- Greptile: **5/5** on `35a979fce6bb9ab910fab3373ec8e537fc2d01a9`, with
the prior finding resolved and no new findings.
- Fixed a CI timing failure in the GitHub mention integration test:
asynchronous admission now has an explicit ten-second assertion deadline
instead of the default one second. The failure reproduced locally; all
five mention variants and all 398 cases in the matching CI shard pass
after the fix.
- Hosted CI passed on `35a979fce6bb9ab910fab3373ec8e537fc2d01a9`: **54
successful checks**, two intentionally skipped Storybook checks, no
failures or pending checks.

Provider responses are simulated in the integration suite. PostgreSQL,
transactions, service dispatch, and scheduling are real. No live Slack
workspace was used.

## Risks

- Every producer that changes runnable state must emit a transaction
signal. This change covers the five queues and their foreground
dispatch/retry paths. Direct SQL and independent DB owners still need
explicit integration or startup recovery.
- Missing retry or claim deadlines can strand work. Tests compare the
dispatch and deadline behavior, including paused and uncertain states.
- Other provider maintenance, sockets, and scheduled jobs remain. This
change does not establish full instance sleep.
- No new tables, migrations, or provider credentials are required.

## Model Used

OpenAI Codex (GPT-6), with repository inspection, code execution, and
automated tests.

## 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>

doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md

@@ -422,7 +422,19 @@ Longer polling intervals alone do not complete an item.
   eligibility predicate so paused endpoints, unknown outcomes, blocked followers,
   and receipts owned by a live sender do not cause empty scans. Completion wakes
   refill free endpoint slots without waiting for slow peers. No new tables.
-  Remaining **1s** maintenance: provider runtimes, provider-specific action
+  Five additional queues now use independent commit/deadline workers:
+  Slack Board messages, slash-command task starts, setup verification messages,
+  receipt reactions, and Slack session stops. Their producers and foreground
+  claim/retry/settlement writes signal inside the owning transaction. The same
+  eligibility predicates drive dispatch and deadline selection. Paused task
+  starts/Board messages and uncertain sends have no recurring deadline.
+  Verification waits for account linking, welcome completion, and a verified
+  callback; those state changes wake it. Failed confirmed task admission saves
+  a retry deadline so its commit cannot cause an immediate retry loop.
+  Startup restores existing rows, and empty workers disarm. No new tables.
+  Tests cover an idle hour, real producer notifications, PostgreSQL deadline
+  selection, stale claims, outer-transaction wakes, and callback-driven setup.
+  Remaining **1s** maintenance: provider runtimes, other provider action
   outboxes, run milestone/wake-notice projection, GitHub operation recovery,
   Teams protocol maintenance, and Slack session sync. GitHub pre-ingress recovery
   must still inspect provider history: a lost callback has no local queue row.

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -93,6 +93,7 @@ describe("createChatReconciliationCoordinator", () => {
   it("wires bounded publication refill to commit workers and isolates periodic maintenance", () => {
     const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
     expect(source).toContain("registerChatDeliveryWork(deliveryWork, chatChannels,");
+    expect(source).toContain("registerChatActionWork(deliveryWork, chatChannels,");
     const maintenance = source.slice(
       source.indexOf("const reconcileChatPublicationMaintenance ="),
       source.indexOf("const chatReconciliation ="),

server/src/__tests__/chat-action-work.test.ts

@@ -0,0 +1,109 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { registerChatActionWork } from "../services/chat-action-work.js";
+import { CHAT_ACTION_QUEUES, notifyChatActionWork, notifyChatEndpointWork, notifyChatVerificationWork } from "../services/chat-work-notifications.js";
+
+const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
+afterEach(async () => {
+  for (const coordinator of coordinators.splice(0)) await coordinator.stop();
+  vi.useRealTimers();
+});
+function setup() {
+  vi.useFakeTimers();
+  const execute = vi.fn(async () => [{ xid: "42", status: "committed" }]);
+  const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>) {
+    return callback({ execute });
+  } });
+  const enabled = vi.fn(() => true), onError = vi.fn();
+  const coordinator = createDeliveryWorkCoordinator({ owner, canRun: enabled, canReconcile: enabled, onError });
+  coordinators.push(coordinator);
+  const service = {
+    processPendingSlackBoardMessages: vi.fn(async () => {}),
+    processPendingSlackTaskStarts: vi.fn(async () => {}),
+    processPendingReceiptReactions: vi.fn(async () => {}),
+    processPendingSlackSessionStops: vi.fn(async () => {}),
+    processPendingVerificationMessages: vi.fn(async () => {}),
+    nextChatActionAt: vi.fn(async (): Promise<number | null> => null),
+    nextVerificationMessageAt: vi.fn(async (): Promise<number | null> => null),
+  };
+  const runs = [service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
+    service.processPendingReceiptReactions, service.processPendingSlackSessionStops, service.processPendingVerificationMessages];
+  return { owner, service, runs, coordinator, enabled, onError, start: () => registerChatActionWork(coordinator, service, enabled) };
+}
+describe("chat action scheduling", () => {
+  it("does one startup scan per lane and no more SQL or timers during an idle hour", async () => {
+    const s = setup();
+    await s.start().ready;
+    await vi.advanceTimersByTimeAsync(3_600_000);
+    for (const run of s.runs) expect(run).toHaveBeenCalledTimes(1);
+    expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(4);
+    expect(s.service.nextVerificationMessageAt).toHaveBeenCalledTimes(1);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("wakes each lane after its write settles and wakes parked work when an endpoint changes", async () => {
+    const s = setup();
+    await s.start().ready;
+    for (const kind of Object.keys(CHAT_ACTION_QUEUES) as (keyof typeof CHAT_ACTION_QUEUES)[]) {
+      await s.owner.transaction(async tx => {
+        await notifyChatActionWork(tx, kind);
+        expect(s.runs.every(run => run.mock.calls.length === 1)).toBe(true);
+      });
+    }
+    await s.owner.transaction(tx => notifyChatVerificationWork(tx));
+    await vi.advanceTimersByTimeAsync(1);
+    for (const run of s.runs) expect(run).toHaveBeenCalledTimes(2);
+    await s.owner.transaction(tx => notifyChatEndpointWork(tx));
+    await vi.advanceTimersByTimeAsync(1);
+    for (const run of s.runs) expect(run).toHaveBeenCalledTimes(3);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("restores durable deadlines and reschedules when a foreground send moves a retry earlier", async () => {
+    const s = setup();
+    let due: number | null = Date.now() + 120_000;
+    s.service.nextChatActionAt.mockImplementation(async () => due);
+    await s.start().ready;
+    expect(s.coordinator.nextWakeAt()).toBe(due);
+    await vi.advanceTimersByTimeAsync(1000);
+    due = Date.now() + 30_000;
+    await s.owner.transaction(tx => notifyChatActionWork(tx, "receipt_reaction"));
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.coordinator.nextWakeAt()).toBe(due);
+    expect(s.service.processPendingReceiptReactions).toHaveBeenCalledTimes(2);
+    due = null;
+    await vi.advanceTimersByTimeAsync(30_000);
+    expect(s.service.processPendingReceiptReactions).toHaveBeenCalledTimes(3);
+    await vi.advanceTimersByTimeAsync(120_000);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("keeps lanes independent and retains a commit received during a drain", async () => {
+    const s = setup();
+    let release!: () => void;
+    s.service.processPendingSlackBoardMessages.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
+    const work = s.start();
+    await vi.advanceTimersByTimeAsync(1);
+    await s.owner.transaction(tx => notifyChatActionWork(tx, "slack_board_message"));
+    await s.owner.transaction(tx => notifyChatActionWork(tx, "slack_session_stop"));
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.processPendingSlackSessionStops).toHaveBeenCalledTimes(2);
+    release(); await work.ready;
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.processPendingSlackBoardMessages).toHaveBeenCalledTimes(2);
+  });
+  it("retries failed recovery and does no dispatch or deadline queries during standby", async () => {
+    const s = setup();
+    s.enabled.mockReturnValue(false);
+    await s.start().ready;
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(s.service.nextChatActionAt).not.toHaveBeenCalled();
+    for (const run of s.runs) expect(run).not.toHaveBeenCalled();
+    s.enabled.mockReturnValue(true);
+    s.service.processPendingSlackTaskStarts.mockRejectedValueOnce(new Error("DB unavailable"));
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(s.onError).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(s.service.processPendingSlackTaskStarts).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+});

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

@@ -1,6 +1,7 @@
+import { registerChatActionWork, dueChatAction } from "../services/chat-action-work.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
-import { notifyChatPublicationWork } from "../services/chat-work-notifications.js";
+import { notifyChatPublicationWork, notifyChatActionWork, CHAT_ACTION_QUEUES } from "../services/chat-work-notifications.js";
 import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
 import { aiConnectionService } from "../services/ai-connections.js";
 import { fastResponseService } from "../services/fast-responses.js";
@@ -4492,7 +4493,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         const delivery = randomUUID();
         const send = () => f.service.handleWebhook(f.endpoint.publicId, "github", signedGitHubWebhookRequest({ event, delivery, payload, webhookSecret: f.webhookSecret }));
         expect((await send()).status).toBeLessThan(300);
-        await expect.poll(async () => (await db.select().from(chatConversations).where(eq(chatConversations.endpointId, f.endpoint.id))).length).toBe(1);
+        // Webhook acknowledgement precedes the asynchronous admission transaction.
+        await expect.poll(async () => (await db.select().from(chatConversations).where(eq(chatConversations.endpointId, f.endpoint.id))).length, { timeout: 10_000 }).toBe(1);
         await send();
         const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.endpointId, f.endpoint.id));
         expect(conversation.externalThreadId).toBe(`github:PaperclipAI/Paperclip:${event === "issues" || event === "issue_comment" ? "issue:" : ""}83${event === "pull_request_review_comment" ? ":rc:83000" : ""}`);
@@ -6553,6 +6555,27 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
       await expect(restarted.slackRegistration.install(f.endpoint.id, actor)).resolves.toHaveProperty("authorizationUrl");
       expect((await restarted.get(f.endpoint.id)).setup.webhookVerifiedAt).toBeFalsy();
     });
+    it("wakes verification from a signed callback with no maintenance polling", async () => {
+      const f = await fixture({ scheduleDeferredWork: () => undefined });
+      await f.create(); await f.service.slackRegistration.complete(await f.install(), "code", null, actor);
+      const onError = vi.fn();
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+      try {
+        await registerChatActionWork(coordinator, f.service, () => true).ready;
+        await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+        expect(await f.service.nextVerificationMessageAt()).toBeNull();
+        await f.service.handleWebhook(f.endpoint.publicId, "slack", signedSlackWebhookRequest({
+          url: f.endpoint.setup.webhookUrl!, contentType: "application/json", signingSecret,
+          body: JSON.stringify({ type: "url_verification", challenge: "scheduler" }),
+        }));
+        await vi.waitFor(async () => {
+          expect((await f.service.get(f.endpoint.id)).setup.slackAccount?.verificationStatus).toBe("sent");
+          expect(coordinator.nextWakeAt()).toBeNull();
+        }, { timeout: 10_000 });
+        expect(f.provider.mock.calls.filter(([url]) => String(url).endsWith("chat.postMessage"))).toHaveLength(2);
+        expect(onError).not.toHaveBeenCalled();
+      } finally { await coordinator.stop(); }
+    });
     it("acknowledges verification before sending one DM across concurrent callbacks and restart recovery", async () => {
       const f = await fixture({ scheduleDeferredWork: () => undefined });
       await f.create(); await f.service.slackRegistration.complete(await f.install(), "code", null, actor);
@@ -6979,6 +7002,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
     const fixture = await seedCompany();
     const { callbacks, endpoint, runtime, service, wakeup } = await configuredSlackEndpoint(fixture);
+    const workWake = vi.fn();
+    const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatBoardMessages, workWake);
+    onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); });
     const channel = makeThread({ channelId: "CBOARD", id: "slack:CBOARD:8000.1", name: "board" });
     await deliverMessage({ callbacks, endpointId: endpoint.id, thread: channel.thread,
       message: makeMessage({ id: "8000.1", text: "@maya start here", mentioned: true }), trigger: "mention" });
@@ -7013,6 +7039,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
       "**Owner User (via Paperclip)**\n\nPlease make the plan", "The plan is ready.",
     ]);
     // The explicit Board composer uses the same mirror and a durable wakeup.
+    workWake.mockClear();
     const key = randomUUID();
     await service.publishBoardMessage(endpoint.id, conversation.id, "Now assign the follow-ups", key, "owner-user");
     await service.publishBoardMessage(endpoint.id, conversation.id, "Now assign the follow-ups", key, "owner-user");
@@ -7025,9 +7052,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     wakeup.mockRejectedValueOnce(new Error("temporary scheduler failure"));
     await service.publishBoardMessage(endpoint.id, conversation.id, "Retry this work", randomUUID(), "owner-user");
     expect(await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "slack_board_message"), eq(chatActions.status, "received")))).toHaveLength(1);
-    await service.processPendingDeliveries();
+    const recovery = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
+    try {
+      await registerChatActionWork(recovery, service, () => true).ready;
+      await vi.waitFor(async () => {
+        expect(await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "slack_board_message"), eq(chatActions.status, "received")))).toHaveLength(0);
+      });
+    } finally { await recovery.stop(); }
     await service.processPendingPublications();
-    expect(await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "slack_board_message"), eq(chatActions.status, "received")))).toHaveLength(0);
     expect(runtime.endpoints.get(endpoint.id)!.posts.filter(post => post.text.includes("Retry this work"))).toHaveLength(1);
   });
 
@@ -18376,6 +18408,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     });
     const { callbacks, endpoint, runtime, service } =
       await configuredSlackEndpoint(fixture, { cancelRun });
+    const workWake = vi.fn();
+    const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatSessionStops, workWake);
+    onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); });
     const externalUserId = "USTOPPER1";
     const channelId = "CSTOPSESSION1";
     const threadTs = `${Math.floor(Date.now() / 1_000) - 5}.100000`;
@@ -18482,6 +18517,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
       },
     });
     const webhookUrl = `https://paperclip.example/api/chat-webhooks/${endpoint.publicId}/slack`;
+    workWake.mockClear();
     const first = await service.handleWebhook(
       endpoint.publicId,
       "slack",
@@ -27664,6 +27700,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const fixture = await seedCompany();
     const { callbacks, endpoint, runtime, service } =
       await configuredSlackEndpoint(fixture);
+    const workWake = vi.fn();
+    const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatReceiptReactions, workWake);
+    onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); });
     const endpointRuntime = runtime.endpoints.get(endpoint.id);
     if (!endpointRuntime) throw new Error("Expected Slack runtime");
     endpointRuntime.reactionErrors.push(
@@ -32853,6 +32892,101 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     return { ...fixture, storage, runtime, service, lanes, enqueue, cleanup };
   }
 
+  it("restores action retry and stale-claim deadlines and parks paused or uncertain work", async () => {
+    const f = await publicationLaneFixture(1);
+    const lane = f.lanes[0]!;
+    const at = new Date();
+    const later = new Date(at.getTime() + 120_000);
+    try {
+      for (const kind of Object.keys(CHAT_ACTION_QUEUES) as (keyof typeof CHAT_ACTION_QUEUES)[]) {
+        const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+          kind, providerActionId: randomUUID(), status: "received", updatedAt: at }).returning();
+        const set = (status: string, result: Record<string, unknown> = {}) => db.update(chatActions)
+          .set({ status, result, updatedAt: at }).where(eq(chatActions.id, action!.id));
+        const due = async () => db.select({ id: chatActions.id }).from(chatActions)
+          .leftJoin(chatEndpoints, eq(chatEndpoints.id, chatActions.endpointId))
+          .where(and(eq(chatActions.id, action!.id), dueChatAction(kind))).then(rows => rows.length > 0);
+        if (kind === "slack_board_message") {
+          expect(await due()).toBe(true);
+          await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+          expect(await f.service.nextChatActionAt(kind)).toBeNull();
+          expect(await due()).toBe(false);
+          await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+        } else if (kind === "slash_task_start") {
+          for (const [status, delay] of [["validating", 60_000], ["resolving", 300_000], ["admitting", 60_000]] as const) {
+            await set(status);
+            expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + delay);
+            expect(await due()).toBe(false);
+          }
+          for (const status of ["queued", "provider_confirmed"]) {
+            await set(status, { retryAt: later.toISOString() });
+            expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime());
+            expect(await due()).toBe(false);
+            await db.update(chatEndpoints).set({ status: "attention" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+            expect(await f.service.nextChatActionAt(kind)).toBeNull();
+            await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+          }
+          await set("delivery_unknown");
+          expect(await f.service.nextChatActionAt(kind)).toBeNull();
+          expect(await due()).toBe(false);
+        } else {
+          expect(await due()).toBe(true);
+          await set("processing");
+          expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + 60_000);
+          expect(await due()).toBe(false);
+          await set("failed", { retryable: true, retryAt: later.toISOString() });
+          expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime());
+          expect(await due()).toBe(false);
+          await set("failed", { retryable: false });
+          expect(await f.service.nextChatActionAt(kind)).toBeNull();
+        }
+        await set("processed");
+        expect(await f.service.nextChatActionAt(kind)).toBeNull();
+      }
+    } finally { await f.cleanup(); }
+  });
+
+  it("recovers action rows on startup, wakes after outer commit, and leaves empty queues quiet", async () => {
+    const f = await publicationLaneFixture(1);
+    const lane = f.lanes[0]!;
+    const onError = vi.fn();
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    const action = (kind: "receipt_reaction" | "slack_session_stop") => ({ companyId: f.companyId,
+      endpointId: lane.endpoint.id, kind, providerActionId: randomUUID(), status: "processing",
+      updatedAt: new Date(Date.now() - 120_000), payload: {} });
+    try {
+      const [stale] = await db.insert(chatActions).values(action("receipt_reaction")).returning();
+      await registerChatActionWork(coordinator, f.service, () => true).ready;
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+      expect((await db.select().from(chatActions).where(eq(chatActions.id, stale!.id)))[0]?.status).toBe("failed");
+      const scans = vi.spyOn(db, "select");
+      try {
+        await new Promise(resolve => setTimeout(resolve, 1200));
+        expect(scans).not.toHaveBeenCalled();
+      } finally { scans.mockRestore(); }
+      let release!: () => void, inserted!: () => void;
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      const staged = new Promise<void>(resolve => { inserted = resolve; });
+      const writing = db.transaction(async tx => {
+        await tx.transaction(async nested => {
+          await notifyChatActionWork(nested, "slack_session_stop");
+          await nested.insert(chatActions).values(action("slack_session_stop"));
+        });
+        inserted(); await gate;
+      });
+      await staged;
+      expect(coordinator.nextWakeAt()).not.toBeNull();
+      release(); await writing;
+      await vi.waitFor(async () => {
+        const rows = await db.select().from(chatActions).where(and(eq(chatActions.companyId, f.companyId), eq(chatActions.kind, "slack_session_stop")));
+        expect(rows[0]?.status).toBe("failed");
+        expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(lane.providerRuntime.posts).toHaveLength(0);
+      expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); await f.cleanup(); }
+  });
+
   it("restores chat queue deadlines without polling blocked followers or receipt owners", async () => {
     const f = await publicationLaneFixture(1);
     const { service, lanes, enqueue } = f;
@@ -53054,6 +53188,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const fixture = await seedCompany();
     const { callbacks, endpoint, runtime, service } =
       await configuredSlackEndpoint(fixture);
+    const workWake = vi.fn();
+    const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatTaskStarts, workWake);
+    onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); });
     const command = endpoint.setup.command;
     if (!command) throw new Error("Slack endpoint did not expose its command");
     if (!callbacks.onSlashCommand) {

server/src/app.ts

@@ -13,6 +13,7 @@ import { customerSuccessRoutes } from "./routes/customer-success.js";
 import { cloudWarmStandbyMiddleware } from "./middleware/cloud-warm-standby.js";
 import type { CloudWarmStandby } from "./services/cloud-warm-standby.js";
 import { browserUseRoutes } from "./routes/browser-use.js";
+import { registerChatActionWork } from "./services/chat-action-work.js";
 import { registerChatDeliveryWork } from "./services/chat-delivery-work.js";
 import { registerBrowserUseCleanup } from "./services/browser-use-work.js";
 import { browserUseService } from "./services/browser-use.js";
@@ -1252,6 +1253,7 @@ export async function createApp(
     });
   }
   registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
+  registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
   emailChannels.start();
   const reconcileChatPublicationMaintenance = async () => {
     await chatChannels.processPublicationMaintenance();

server/src/services/chat-action-work.ts

@@ -0,0 +1,85 @@
+import { and, eq, inArray, lte, or, sql } from "drizzle-orm";
+import { chatActions, chatEndpoints, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { CHAT_ACTION_QUEUES, type ScheduledChatAction } from "./chat-work-notifications.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { registerChatQueueWork } from "./chat-delivery-work.js";
+
+export const SLACK_COMMAND_POST_STALE_MS = 60_000;
+export const SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS = 5 * 60_000;
+export const SLACK_COMMAND_ADMISSION_STALE_MS = 60_000;
+export const PROVIDER_EFFECT_STALE_MS = 60_000;
+
+/** The dispatch predicate and scheduler use the same eligibility and deadlines.
+ * Zero means ready now; null means terminal or waiting for an external change. */
+export function chatActionDeadline(kind: ScheduledChatAction) {
+  const updated = sql`extract(epoch from ${chatActions.updatedAt}) * 1000`;
+  const retry = sql`coalesce(extract(epoch from (${chatActions.result}->>'retryAt')::timestamptz) * 1000, 0)`;
+  if (kind === "slack_board_message") return sql<number>`case
+    when ${chatActions.status} = 'received' and ${chatEndpoints.status} not in ('paused', 'attention') then 0::numeric end`;
+  if (kind === "slash_task_start") return sql<number>`case
+    when ${chatActions.status} = 'queued' and (${chatEndpoints.id} is null or ${chatEndpoints.status} in ('verifying', 'active', 'archived')) then ${retry}
+    when ${chatActions.status} = 'provider_confirmed' and (${chatEndpoints.id} is null or ${chatEndpoints.status} in ('verifying', 'active', 'archived')) then ${retry}
+    when ${chatActions.status} = 'admitting' then ${updated} + ${SLACK_COMMAND_ADMISSION_STALE_MS}
+    when ${chatActions.status} = 'resolving' then ${updated} + ${SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS}
+    when ${chatActions.status} = 'validating' then ${updated} + ${SLACK_COMMAND_POST_STALE_MS} end`;
+  return sql<number>`case
+    when ${chatActions.status} = 'received' then 0
+    when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS}
+    when ${chatActions.status} = 'failed' and ${chatActions.result}->>'retryable' = 'true' then ${retry} end`;
+}
+
+function pendingChatAction(kind: ScheduledChatAction) {
+  // Filter history before the endpoint join and aggregate. Failed reactions
+  // and stops only remain live when the saved result permits another attempt.
+  const status = kind === "slack_board_message"
+    ? eq(chatActions.status, "received")
+    : kind === "slash_task_start"
+      ? inArray(chatActions.status, ["queued", "provider_confirmed", "admitting", "resolving", "validating"])
+      : or(inArray(chatActions.status, ["received", "processing"]),
+        and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`));
+  return and(eq(chatActions.kind, kind), status);
+}
+
+export function dueChatAction(kind: ScheduledChatAction, now = Date.now()) {
+  return and(pendingChatAction(kind), lte(chatActionDeadline(kind), now));
+}
+
+export async function nextChatActionAt(db: Db, kind: ScheduledChatAction): Promise<number | null> {
+  const [row] = await db.select({ at: sql<string | null>`min(${chatActionDeadline(kind)})` })
+    .from(chatActions)
+    .leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
+    .where(pendingChatAction(kind));
+  return row?.at == null ? null : Number(row.at);
+}
+
+type ChatActionService = {
+  processPendingSlackBoardMessages(): Promise<unknown>;
+  processPendingSlackTaskStarts(): Promise<unknown>;
+  processPendingReceiptReactions(): Promise<unknown>;
+  processPendingSlackSessionStops(): Promise<unknown>;
+  processPendingVerificationMessages(): Promise<unknown>;
+  nextChatActionAt(kind: ScheduledChatAction): Promise<number | null>;
+  nextVerificationMessageAt(): Promise<number | null>;
+};
+
+export function registerChatActionWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ChatActionService,
+  canRun: () => boolean,
+) {
+  const actions = [
+    { kind: "slack_board_message", run: service.processPendingSlackBoardMessages },
+    { kind: "slash_task_start", run: service.processPendingSlackTaskStarts },
+    { kind: "receipt_reaction", run: service.processPendingReceiptReactions },
+    { kind: "slack_session_stop", run: service.processPendingSlackSessionStops },
+  ] as const;
+  const workers = actions.map(({ kind, run }) => registerChatQueueWork(coordinator, {
+    queue: CHAT_ACTION_QUEUES[kind], run, next: () => service.nextChatActionAt(kind),
+  }, canRun));
+  workers.push(registerChatQueueWork(coordinator, {
+    queue: DELIVERY_QUEUES.chatVerification,
+    run: service.processPendingVerificationMessages, next: service.nextVerificationMessageAt,
+  }, canRun));
+  return { ready: Promise.all(workers.map(worker => worker.ready)) };
+}

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

@@ -1,5 +1,5 @@
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
-import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { DELIVERY_QUEUES, type DeliveryQueue } from "./delivery-work-notifications.js";
 
 type ChatDeliveryService = {
   processQueuedDeliveries(): Promise<unknown>;
@@ -23,16 +23,7 @@ export function registerChatDeliveryWork(
     { queue: DELIVERY_QUEUES.chatReceipts, run: service.processPendingSlackFileUploadReceipts, next: service.nextSlackReceiptAt },
   ];
   const workers = tasks.map(task => {
-    let next: number | null = null;
-    const worker = coordinator.register(task.queue, {
-      retryMs: 1000,
-      run: () => task.run(),
-      hasPending: async () => {
-        next = canRun() ? await task.next() : Date.now() + 1000;
-        return false;
-      },
-      nextRunAt: () => next === null ? null : Math.max(Date.now() + 1000, next),
-    });
+    const worker = registerChatQueueWork(coordinator, task, canRun);
     if (task.queue === DELIVERY_QUEUES.chatPublications) {
       // Dispatch returns after reserving endpoint slots. Refill them when a
       // provider task ends, including failures before its final durable write.
@@ -42,3 +33,22 @@ export function registerChatDeliveryWork(
   });
   return { ready: Promise.all(workers.map(worker => worker.ready)) };
 }
+
+/** Empty queues disarm; only known work, uncertain writes, and failures retry. */
+export function registerChatQueueWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  task: { queue: DeliveryQueue; run(): Promise<unknown>; next(): Promise<number | null> },
+  canRun: () => boolean,
+) {
+  let next: number | null = null;
+  const worker = coordinator.register(task.queue, {
+    retryMs: 1000,
+    run: () => task.run(),
+    hasPending: async () => {
+      next = canRun() ? await task.next() : Date.now() + 1000;
+      return false;
+    },
+    nextRunAt: () => next === null ? null : Math.max(Date.now() + 1000, next),
+  });
+  return worker;
+}

server/src/services/chat-receipt-reactions.ts

@@ -1,3 +1,4 @@
+import { notifyChatActionWork } from "./chat-work-notifications.js";
 import { and, eq, inArray, or, sql } from "drizzle-orm";
 import {
   chatActions,
@@ -205,6 +206,7 @@ export async function stageReceiptReactionRemovals(
     ];
   });
   if (removals.length === 0) return [];
+  await notifyChatActionWork(tx, "receipt_reaction");
   const removalDeliveryIds = new Set(
     removals.map((removal) => removal.deliveryId),
   );

server/src/services/chat-slack-registration.ts

@@ -1,3 +1,4 @@
+import { notifyChatVerificationWork } from "./chat-work-notifications.js";
 import { createAgentAvatarPool } from "./agent-avatar-pool.js";
 import type { AgentAvatarRequest } from "./agent-avatars.js";
 import { removeSlackRegistration } from "./chat-slack-registration-cleanup.js";
@@ -111,6 +112,7 @@ export function slackChatRegistrationService(db: Db, options: {
           ...(ids[key] ? { existingRef: { secretId: ids[key], configPath: `slack_registration.${key}` } } : {}) });
         ids[key] = result.secret.id;
       }
+      await notifyChatVerificationWork(tx);
       await tx.update(chatSlackRegistrations).set({ ...patch, secretIds: ids, updatedAt: new Date() }).where(eq(chatSlackRegistrations.endpointId, row.endpointId));
       if (account) {
         const saved = (await tx.select().from(chatEndpoints).where(and(eq(chatEndpoints.id, row.endpointId), eq(chatEndpoints.companyId, row.companyId))).for("update"))[0];
@@ -334,6 +336,7 @@ export function slackChatRegistrationService(db: Db, options: {
       const saved = (await tx.select().from(chatEndpoints).where(and(eq(chatEndpoints.id, row.endpointId), eq(chatEndpoints.companyId, row.companyId))).for("update"))[0];
       const registrationNow = (await tx.select().from(chatSlackRegistrations).where(eq(chatSlackRegistrations.endpointId, row.endpointId)))[0];
       if (!saved || saved.status === "archived" || registrationNow?.status === "removed" || registrationNow?.requestId !== row.requestId) throw conflict("Slack setup changed");
+      await notifyChatVerificationWork(tx);
       await tx.update(chatEndpoints).set({ setup: { ...saved.setup, slackAccount: account }, updatedAt: new Date() }).where(eq(chatEndpoints.id, row.endpointId));
       await lease.assertOwned(tx);
     });
@@ -397,12 +400,26 @@ export function slackChatRegistrationService(db: Db, options: {
       await sendSetupMessage(row, { userId: account.paperclipUserId, sessionId: null, bypassPermissionCheck: false }, lease, "verification");
     });
   }
-  async function processPendingVerificationMessages(limit = 25) {
-    const pending = await db.select({ endpointId: chatEndpoints.id }).from(chatEndpoints)
+  function pendingVerificationMessage() {
+    // A pending verification cannot progress until linking, welcome, and an
+    // authenticated callback finish. Their commits wake this lane.
+    return and(eq(chatSlackRegistrations.status, "configured"),
+      inArray(chatEndpoints.status, ["verifying", "active"]),
+      sql`${chatEndpoints.setup}->'slackAccount'->>'status' = 'linked'`,
+      sql`${chatEndpoints.setup}->>'webhookVerifiedAt' is not null`,
+      sql`coalesce(${chatEndpoints.setup}->'slackAccount'->>'welcomeStatus', '') not in ('pending', 'sending')`,
+      sql`${chatEndpoints.setup}->'slackAccount'->>'verificationStatus' in ('pending', 'sending')`);
+  }
+  function pendingVerificationMessages(limit: number) {
+    return db.select({ endpointId: chatEndpoints.id }).from(chatEndpoints)
       .innerJoin(chatSlackRegistrations, and(eq(chatSlackRegistrations.endpointId, chatEndpoints.id), eq(chatSlackRegistrations.companyId, chatEndpoints.companyId)))
-      .where(and(eq(chatSlackRegistrations.status, "configured"), inArray(chatEndpoints.status, ["verifying", "active"]),
-        sql`${chatEndpoints.setup}->'slackAccount'->>'verificationStatus' in ('pending', 'sending')`)).limit(limit);
-    for (const row of pending) {
+      .where(pendingVerificationMessage()).limit(limit);
+  }
+  async function nextVerificationMessageAt(): Promise<number | null> {
+    return (await pendingVerificationMessages(1)).length ? Date.now() : null;
+  }
+  async function processPendingVerificationMessages(limit = 25) {
+    for (const row of await pendingVerificationMessages(limit)) {
       try { await notifyVerified(row.endpointId); }
       catch { logger.warn({ endpointId: row.endpointId }, "Slack verification message could not be processed"); }
     }
@@ -437,7 +454,10 @@ export function slackChatRegistrationService(db: Db, options: {
       return;
     }
     await lease.assertOwned();
-    await db.update(chatSlackRegistrations).set({ status: "configured", errorCode: null, updatedAt: new Date() }).where(eq(chatSlackRegistrations.endpointId, endpointId));
+    await db.transaction(async tx => {
+      await notifyChatVerificationWork(tx);
+      await tx.update(chatSlackRegistrations).set({ status: "configured", errorCode: null, updatedAt: new Date() }).where(eq(chatSlackRegistrations.endpointId, endpointId));
+    });
     // Runtime now owns a separate vaulted copy. Keep the registration IDs until
     // deletion succeeds so cleanup itself is retryable.
     await cleanupStaged(endpointId, lease);
@@ -505,7 +525,7 @@ export function slackChatRegistrationService(db: Db, options: {
     if (!row) throw notFound("Slack connection not found");
     return `/${encodeURIComponent(row.prefix)}/apps/chat/connect?provider=slack&resume=${encodeURIComponent(endpointId)}`;
   }
-  return { create, install, pending, expiredReturn, complete, cleanup, returnPath, registration, notifyVerified, processPendingVerificationMessages,
+  return { create, install, pending, expiredReturn, complete, cleanup, returnPath, registration, notifyVerified, processPendingVerificationMessages, nextVerificationMessageAt,
     close: async () => { await avatarPool?.close(); },
     resume: (endpointId: string, actor: SlackSetupActor) => options.withLock(endpointId, lease => resumeLocked(endpointId, actor, lease)) };
 }

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

@@ -10,7 +10,25 @@ export async function notifyChatPublicationWork(tx: object): Promise<void> {
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceipts);
 }
 
+export const CHAT_ACTION_QUEUES = {
+  slack_board_message: DELIVERY_QUEUES.chatBoardMessages,
+  slash_task_start: DELIVERY_QUEUES.chatTaskStarts,
+  receipt_reaction: DELIVERY_QUEUES.chatReceiptReactions,
+  slack_session_stop: DELIVERY_QUEUES.chatSessionStops,
+} as const;
+export type ScheduledChatAction = keyof typeof CHAT_ACTION_QUEUES;
+
+export async function notifyChatActionWork(tx: object, kind: ScheduledChatAction): Promise<void> {
+  await notifyDeliveryWork(tx, CHAT_ACTION_QUEUES[kind]);
+}
+
+export async function notifyChatVerificationWork(tx: object): Promise<void> {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatVerification);
+}
+
 export async function notifyChatEndpointWork(tx: object): Promise<void> {
   await notifyChatDeliveryWork(tx);
   await notifyChatPublicationWork(tx);
+  for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue);
+  await notifyChatVerificationWork(tx);
 }

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

@@ -2,6 +2,11 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  chatBoardMessages: "chat-board-messages",
+  chatTaskStarts: "chat-task-starts",
+  chatReceiptReactions: "chat-receipt-reactions",
+  chatSessionStops: "chat-session-stops",
+  chatVerification: "chat-verification",
   chatInbound: "chat-inbound",
   chatPublications: "chat-publications",
   chatReceipts: "chat-slack-receipts",

server/src/services/slack-board-messages.ts

@@ -1,4 +1,4 @@
-import { notifyChatPublicationWork } from "./chat-work-notifications.js";
+import { notifyChatPublicationWork, notifyChatActionWork } from "./chat-work-notifications.js";
 import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
 import {
   assets,
@@ -102,6 +102,7 @@ export async function mirrorSlackBoardComment(
     if (!principal) {
       throw forbidden("Link your Slack account to this connection before sending a message to its Slack thread");
     }
+    if (options.wakeAgent) await notifyChatActionWork(db, "slack_board_message");
     const [receipt] = await db
       .insert(chatActions)
       .values({