perf(chat): stop polling provider action and session sync queues (#15822)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Chat stores pending provider actions in durable database rows.
> - Three action queues still scan every second when empty.
> - These scans keep idle databases active.
> - This pull request wakes those queues after a commit or at their next
deadline.
> - Empty queues stop querying while existing recovery rules remain in
place.

## Linked Issues or Issue Description

Refs #15808, #15802, #15625. Related sleep work: #15750, #15757.

**What existing behavior does this improve?**

Provider messages and notices, accepted GitHub webhook receipts, and
Slack session status synchronization.

**Current behavior**

Chat maintenance scans these three queues every second. It also rechecks
paused Slack sync rows every 30 seconds.

**Proposed behavior**

Reuse the transaction-aware action workers and existing chat_actions
rows. Wake after commits. Restore unfinished work on startup. Use stored
retry and stale-claim deadlines. Stop scanning when empty. Park paused
Slack sync rows until an endpoint change.

## What Changed

- Add three topics to the existing action worker registration. No
tables, migrations, triggers, or external services.
- Signal producers, claims, retry updates, terminal settlements, and
operator resolution inside their owning transaction. Provider I/O
remains outside transactions.
- Share pending-state and deadline predicates between dispatch and
scheduling. Exclude terminal history and uncertain provider sends.
- Preserve GitHub receipt attempt fences and provider-send quarantine.
Preserve Slack's 30-minute refresh while a session is processing.
- Remove these workers from periodic app maintenance. Update the sleep
inventory and add regression coverage.

## Verification

- Passed: 13 focused tests across scheduler, app wiring, and real
PostgreSQL integration. Covers actual commit wakes with deferred fast
paths disabled, GitHub retries and startup recovery, stale-send
quarantine, Slack publication-to-sync delivery, paused deadlines, and
idle queues.
- Passed: pnpm typecheck.
- Passed: full chat integration and related coordinator suites, **1,274
tests across five files**.
- Passed: pnpm build.
- Greptile: **5/5**, no actionable findings, completed on
`dc284cd275e5f6df624d60b27954421f2bcdbee2`.
- Hosted CI: **54 checks passed**, two intentionally skipped Storybook
checks, no failures or pending checks on
`dc284cd275e5f6df624d60b27954421f2bcdbee2`.

Provider responses are simulated. PostgreSQL, transactions, service
dispatch, and scheduling are real. No live Slack or GitHub App was used.

## Risks

- Writers must emit the existing process-local transaction signal.
Independent database clients and direct SQL still need explicit
integration or startup recovery.
- Accepted GitHub receipts are covered. Finding callbacks GitHub never
delivered still uses the separate provider recovery lane.
- Other chat maintenance and persistent transports still run. This does
not establish whole-instance sleep.
- Existing claim, revision, credential, and authorization checks remain
authoritative during concurrent foreground and scheduled dispatch.

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

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

@@ -106,6 +106,12 @@ describe("createChatReconciliationCoordinator", () => {
     // The service integration tests hold real publication workers while this
     // scheduled method returns; app shutdown must also join those workers.
     expect(source).toContain("await chatChannels.shutdown()");
+    expect(source).not.toContain("chatChannels.processPendingSlackSessionSyncs()");
+    const service = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8");
+    const actionMaintenance = service.slice(service.indexOf("async function processPendingChatMaintenance("),
+      service.indexOf("async function processPendingDeliveries("));
+    expect(actionMaintenance).not.toContain("processPendingProviderEffects(");
+    expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress(");
   });
 
   it.each([false, true])(

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

@@ -19,6 +19,9 @@ function setup() {
   const coordinator = createDeliveryWorkCoordinator({ owner, canRun: enabled, canReconcile: enabled, onError });
   coordinators.push(coordinator);
   const service = {
+    processPendingProviderEffects: vi.fn(async () => {}),
+    processPendingGitHubWebhookIngress: vi.fn(async () => {}),
+    processPendingSlackSessionSyncs: vi.fn(async () => {}),
     processPendingSlackBoardMessages: vi.fn(async () => {}),
     processPendingSlackTaskStarts: vi.fn(async () => {}),
     processPendingReceiptReactions: vi.fn(async () => {}),
@@ -27,7 +30,7 @@ function setup() {
     nextChatActionAt: vi.fn(async (): Promise<number | null> => null),
     nextVerificationMessageAt: vi.fn(async (): Promise<number | null> => null),
   };
-  const runs = [service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
+  const runs = [service.processPendingProviderEffects, service.processPendingGitHubWebhookIngress, service.processPendingSlackSessionSyncs, service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
     service.processPendingReceiptReactions, service.processPendingSlackSessionStops, service.processPendingVerificationMessages];
   return { owner, service, runs, coordinator, enabled, onError, start: () => registerChatActionWork(coordinator, service, enabled) };
 }
@@ -37,7 +40,7 @@ describe("chat action scheduling", () => {
     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.nextChatActionAt).toHaveBeenCalledTimes(7);
     expect(s.service.nextVerificationMessageAt).toHaveBeenCalledTimes(1);
     expect(s.coordinator.nextWakeAt()).toBeNull();
     expect(vi.getTimerCount()).toBe(0);

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

@@ -13250,6 +13250,49 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     await service.shutdown();
   });
 
+  it("drains accepted GitHub webhooks from commit wakes and durable retry deadlines", async () => {
+    const fixture = await seedCompany();
+    const { endpoint, runtime, service, webhookSecret } = await configuredGitHubEndpoint(fixture, {
+      deferWebhookProcessing: true, scheduleDeferredWork: () => undefined,
+    });
+    const providerRuntime = runtime.endpoints.get(endpoint.id)!;
+    let attempts = 0;
+    providerRuntime.webhookHook = async () => {
+      providerRuntime.webhookResponse = new Response("result", { status: ++attempts === 1 ? 503 : 202 });
+    };
+    const delivery = randomUUID();
+    const send = () => service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({
+      delivery, event: "installation_repositories", payload: { action: "added", installation: { id: 2468 } }, webhookSecret,
+    }));
+    const onError = vi.fn();
+    let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    const receipt = () => db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id),
+      eq(chatActions.providerActionId, `github_webhook_ingress:${delivery}`))).then(rows => rows[0]);
+    try {
+      await registerChatActionWork(coordinator, service, () => true).ready;
+      expect((await send()).status).toBe(202);
+      await expect.poll(async () => (await receipt())?.status, { timeout: 10_000 }).toBe("processed");
+      expect(attempts).toBe(2);
+      expect((await receipt())?.result).toMatchObject({ attempts: 2, httpStatus: 202 });
+      // A duplicate callback wakes the lane but cannot repeat a settled effect.
+      await send();
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+      expect(attempts).toBe(2);
+      await coordinator.stop();
+      // Work committed while no worker exists is recovered by the next startup.
+      const restartDelivery = randomUUID();
+      await service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({
+        delivery: restartDelivery, event: "installation_repositories",
+        payload: { action: "added", installation: { id: 2468 } }, webhookSecret,
+      }));
+      expect(attempts).toBe(2);
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+      await registerChatActionWork(coordinator, service, () => true).ready;
+      await expect.poll(() => attempts, { timeout: 10_000 }).toBe(3);
+      expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); }
+  });
+
   it("safely re-arms an exact terminal GitHub delivery after an operator redelivery", async () => {
     const fixture = await seedCompany();
     const { endpoint, runtime, service, webhookSecret } =
@@ -32819,7 +32862,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         .update(chatEndpoints)
         .set({ status: "active" })
         .where(eq(chatEndpoints.id, endpoint.id));
-      const channelId = `C-LANE-${endpoint.id}`;
+      const channelId = `C${endpoint.id.replaceAll("-", "").toUpperCase()}`;
       await db.insert(chatEndpointResources).values({
         companyId: fixture.companyId,
         endpointId: endpoint.id,
@@ -32892,6 +32935,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     return { ...fixture, storage, runtime, service, lanes, enqueue, cleanup };
   }
 
+  it("wakes Slack session sync from publication commit and leaves settled action 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 });
+    try {
+      await registerChatActionWork(coordinator, f.service, () => true).ready;
+      await f.enqueue(0, "A completed answer");
+      await f.service.processPendingPublications();
+      await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions)
+        .where(and(eq(chatActions.endpointId, lane.endpoint.id), eq(chatActions.kind, "slack_session_sync"))),
+        { timeout: 10_000 }).toEqual([{ status: "processed" }]);
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+      const scans = vi.spyOn(db, "select");
+      try {
+        await new Promise(resolve => setTimeout(resolve, 1200));
+        expect(scans).not.toHaveBeenCalled();
+      } finally { scans.mockRestore(); }
+      expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); await f.cleanup(); }
+  });
+
+  it("recovers stale provider sends on startup without replaying uncertain effects", async () => {
+    const f = await publicationLaneFixture(1);
+    const lane = f.lanes[0]!;
+    const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+      kind: "provider_effect", providerActionId: randomUUID(), status: "processing", result: { attempts: 1 },
+      updatedAt: new Date(Date.now() - 120_000), payload: {} }).returning();
+    const before = lane.providerRuntime.posts.length;
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true });
+    try {
+      await registerChatActionWork(coordinator, f.service, () => true).ready;
+      await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions)
+        .where(eq(chatActions.id, action!.id)), { timeout: 10_000 }).toEqual([{ status: "delivery_unknown" }]);
+      expect(await f.service.nextChatActionAt("provider_effect")).toBeNull();
+      expect(lane.providerRuntime.posts).toHaveLength(before);
+    } finally { await coordinator.stop(); await f.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]!;
@@ -32929,6 +33011,21 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
           await set("delivery_unknown");
           expect(await f.service.nextChatActionAt(kind)).toBeNull();
           expect(await due()).toBe(false);
+        } else if (kind === "slack_session_sync") {
+          expect(await due()).toBe(true);
+          await set("received", { retryAt: later.toISOString() });
+          expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime());
+          expect(await due()).toBe(false);
+          await set("processing");
+          expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + 60_000);
+          for (const status of ["paused", "attention"]) {
+            await db.update(chatEndpoints).set({ status }).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));
+          await set("failed", { retryable: true });
+          expect(await f.service.nextChatActionAt(kind)).toBeNull();
         } else {
           expect(await due()).toBe(true);
           await set("processing");
@@ -53999,6 +54096,75 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     );
   });
 
+  it("delivers queued provider notices from commit wakes without the deferred fast path", async () => {
+    const fixture = await seedCompany();
+    const { callbacks, endpoint, service, runtime } =
+      await configuredSlackEndpoint(fixture, { scheduleDeferredWork: () => undefined });
+    if (!callbacks.onSlashCommand) {
+      throw new Error("Slack slash command callback was not registered");
+    }
+    await db.insert(chatEndpointResources).values({
+      companyId: fixture.companyId,
+      endpointId: endpoint.id,
+      type: "channel",
+      providerResourceId: "C-CONTROL-GUIDANCE",
+      label: "control-guidance",
+      availability: "available",
+      enabled: true,
+    });
+    const command = endpoint.setup.command;
+    if (!command) throw new Error("Slack endpoint did not expose its command");
+    const post = vi.fn(async () => ({
+      id: "unexpected-channel-control-post",
+      threadId: "slack:C-CONTROL-GUIDANCE:root",
+    }));
+    const postEphemeral = vi.fn(async () => ({
+      id: "channel-control-guidance",
+      threadId: "slack:C-CONTROL-GUIDANCE:root",
+    }));
+    const providerRuntime = runtime.endpoints.get(endpoint.id)!;
+    vi.spyOn(providerRuntime, "thread").mockReturnValue({ post, postEphemeral } as never);
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true });
+    onTestFinished(async () => { await coordinator.stop(); });
+    await registerChatActionWork(coordinator, service, () => true).ready;
+    for (const control of ["status"] as const) {
+      await callbacks.onSlashCommand({
+        endpointId: endpoint.id,
+        provider: "slack",
+        event: {
+          channel: {
+            id: "slack:C-CONTROL-GUIDANCE",
+            name: "control-guidance",
+            isDM: false,
+            post,
+            postEphemeral,
+          } as never,
+          command,
+          text: control,
+          triggerId: `channel-control-${control}`,
+          user: {
+            userId: "U-CHANNEL-CONTROLLER",
+            userName: "channel-controller",
+            fullName: "Channel Controller",
+            isBot: false,
+            isMe: false,
+            isSystem: false,
+          },
+          raw: { command, text: control },
+          adapter: {} as never,
+          openModal: async () => undefined,
+        },
+      });
+    }
+    await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions)
+      .where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "provider_effect"))),
+      { timeout: 10_000 }).toEqual([{ status: "processed" }]);
+    expect(postEphemeral).toHaveBeenCalledTimes(1);
+    expect(post).not.toHaveBeenCalled();
+    expect(await service.nextChatActionAt("provider_effect")).toBeNull();
+    await coordinator.stop();
+  });
+
   it("returns ephemeral guidance for exact Slack controls in channels without creating tasks or actions", async () => {
     const fixture = await seedCompany();
     const { callbacks, endpoint, service } =
@@ -54438,6 +54604,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const fixture = await seedCompany();
     const { callbacks, endpoint, runtime, service } =
       await configuredTelegramEndpoint(fixture);
+    const workWake = vi.fn();
+    const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatProviderEffects, workWake);
+    onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); });
     if (!callbacks.onSlashCommand) {
       throw new Error("Telegram slash command callback was not registered");
     }

server/src/app.ts

@@ -326,7 +326,7 @@ export function createChatReconciliationCoordinator(input: {
   projectRunMilestones: () => Promise<number>;
   flushPublications: () => Promise<unknown>;
   processPendingSlackFileUploadReceipts?: () => Promise<unknown>;
-  processPendingSlackSessionSyncs: () => Promise<unknown>;
+  processPendingSlackSessionSyncs?: () => Promise<unknown>;
   onError: (lane: ChatReconciliationLane, error: unknown) => void;
 }) {
   let stopped = false;
@@ -373,7 +373,7 @@ export function createChatReconciliationCoordinator(input: {
       milestoneReconciliation.poll();
       publicationReconciliation.poll();
       if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
-      start("Slack session status", input.processPendingSlackSessionSyncs);
+      if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs);
     },
     notifyPublications() {
       milestoneReconciliation.notify();
@@ -1271,8 +1271,6 @@ export async function createApp(
         publicBaseUrl: opts.authPublicBaseUrl,
       }),
     flushPublications: () => reconcileChatPublicationMaintenance(),
-    processPendingSlackSessionSyncs: () =>
-      chatChannels.processPendingSlackSessionSyncs(),
     onError: (lane, err) => {
       logger.error({ err, lane }, `Failed to reconcile chat ${lane}`);
     },

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

@@ -23,6 +23,10 @@ export function chatActionDeadline(kind: ScheduledChatAction) {
     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`;
+  if (kind === "slack_session_sync") return sql<number>`case
+    when ${chatEndpoints.status} in ('paused', 'attention') then null
+    when ${chatActions.status} = 'received' then ${retry}
+    when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS} end`;
   return sql<number>`case
     when ${chatActions.status} = 'received' then 0
     when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS}
@@ -30,13 +34,15 @@ export function chatActionDeadline(kind: ScheduledChatAction) {
 }
 
 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.
+  // Filter history before the endpoint join and aggregate. Failed actions
+  // 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"]),
+      : kind === "slack_session_sync"
+        ? inArray(chatActions.status, ["received", "processing"])
+        : or(inArray(chatActions.status, ["received", "processing"]),
         and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`));
   return and(eq(chatActions.kind, kind), status);
 }
@@ -54,6 +60,9 @@ export async function nextChatActionAt(db: Db, kind: ScheduledChatAction): Promi
 }
 
 type ChatActionService = {
+  processPendingProviderEffects(): Promise<unknown>;
+  processPendingGitHubWebhookIngress(): Promise<unknown>;
+  processPendingSlackSessionSyncs(): Promise<unknown>;
   processPendingSlackBoardMessages(): Promise<unknown>;
   processPendingSlackTaskStarts(): Promise<unknown>;
   processPendingReceiptReactions(): Promise<unknown>;
@@ -69,6 +78,9 @@ export function registerChatActionWork(
   canRun: () => boolean,
 ) {
   const actions = [
+    { kind: "provider_effect", run: service.processPendingProviderEffects },
+    { kind: "github_webhook_ingress", run: service.processPendingGitHubWebhookIngress },
+    { kind: "slack_session_sync", run: service.processPendingSlackSessionSyncs },
     { kind: "slack_board_message", run: service.processPendingSlackBoardMessages },
     { kind: "slash_task_start", run: service.processPendingSlackTaskStarts },
     { kind: "receipt_reaction", run: service.processPendingReceiptReactions },

server/src/services/chat-channels.ts

@@ -4581,6 +4581,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
       runtimeContext: LifecycleRuntimeFence;
     },
   ) {
+    await notifyChatActionWork(database, "provider_effect");
     const [inserted] = await database
       .insert(chatActions)
       .values({
@@ -4718,6 +4719,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
               ["verifying", "active"],
             )
           : null;
+      await notifyChatActionWork(tx, "provider_effect");
       const [ownedAction] = await tx
         .update(chatActions)
         .set({
@@ -4819,6 +4821,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     const failure =
       "Provider delivery could not be confirmed after the worker stopped. The effect will not be replayed automatically.";
     await db.transaction(async (tx) => {
+      await notifyChatActionWork(tx, "provider_effect");
       const [quarantined] = await tx
         .update(chatActions)
         .set({
@@ -5093,6 +5096,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     action: typeof chatActions.$inferSelect,
     attempt: number,
   ): Promise<boolean> {
+    await notifyChatActionWork(tx, "provider_effect");
     const [cancelled] = await tx
       .update(chatActions)
       .set({
@@ -5161,14 +5165,17 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
       if (!action) return "failed";
       const payload = providerEffectPayload(action.payload);
       if (!payload) {
-        await db
-          .update(chatActions)
-          .set({
-            status: "failed",
-            result: { code: "provider_effect_payload_invalid" },
-            updatedAt: new Date(),
-          })
-          .where(eq(chatActions.id, action.id));
+        await db.transaction(async (tx) => {
+          await notifyChatActionWork(tx, "provider_effect");
+          return tx
+            .update(chatActions)
+            .set({
+              status: "failed",
+              result: { code: "provider_effect_payload_invalid" },
+              updatedAt: new Date(),
+            })
+            .where(eq(chatActions.id, action.id));
+        });
         return "failed";
       }
       if (action.status === "processed") return "processed";
@@ -5238,6 +5245,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
                     payload,
                   )
                 : false;
+              await notifyChatActionWork(tx, "provider_effect");
               const [transportClaim] = await tx
                 .update(chatActions)
                 .set({
@@ -5355,6 +5363,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
                 },
                 ["verifying", "active"],
               );
+              await notifyChatActionWork(tx, "provider_effect");
               const [completed] = await tx
                 .update(chatActions)
                 .set({
@@ -5481,23 +5490,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     const actions = await db
       .select()
       .from(chatActions)
-      .where(
-        and(
-          eq(chatActions.kind, "provider_effect"),
-          or(
-            eq(chatActions.status, "received"),
-            and(
-              eq(chatActions.status, "processing"),
-              lte(chatActions.updatedAt, staleBefore),
-            ),
-            and(
-              eq(chatActions.status, "failed"),
-              sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
-              sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
-            ),
-          ),
-        ),
-      )
+      .where(dueChatAction("provider_effect", now.getTime()))
       .orderBy(asc(chatActions.createdAt))
       .limit(limit);
     for (const action of actions) {
@@ -25203,6 +25196,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
           runtimeGeneration: context.generation,
           credentialFingerprint: context.credentialFingerprint,
         };
+        await notifyChatActionWork(tx, "github_webhook_ingress");
         const inserted = await tx
           .insert(chatActions)
           .values({
@@ -25229,6 +25223,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
         // currently authenticated body after a terminal processing failure.
         // Successful, cancelled, retryable, and in-flight receipts remain
         // immutable deduplication fences.
+        await notifyChatActionWork(tx, "github_webhook_ingress");
         const [rearmed] = await tx
           .update(chatActions)
           .set({
@@ -25312,32 +25307,35 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
       Date.now() +
         Math.min(60_000, 1_000 * 2 ** Math.max(0, input.attempts - 1)),
     );
-    await db
-      .update(chatActions)
-      .set({
-        status: cancelled ? "cancelled" : processed ? "processed" : "failed",
-        payload: terminal
-          ? redactedGitHubWebhookIngressPayload(input.payload)
-          : input.payload,
-        result: {
-          attempts: input.attempts,
-          ...(input.cancelledCode ? { code: input.cancelledCode } : {}),
-          ...(input.response
-            ? { httpStatus: input.response.status }
-            : { error: redactError(input.error) }),
-          retryable: !terminal,
-          ...(!terminal ? { retryAt: retryAt.toISOString() } : {}),
-        },
-        updatedAt: new Date(),
-      })
-      .where(
-        and(
-          eq(chatActions.id, actionId),
-          eq(chatActions.kind, "github_webhook_ingress"),
-          eq(chatActions.status, "processing"),
-          sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`,
-        ),
-      );
+    await db.transaction(async (tx) => {
+      await notifyChatActionWork(tx, "github_webhook_ingress");
+      return tx
+        .update(chatActions)
+        .set({
+          status: cancelled ? "cancelled" : processed ? "processed" : "failed",
+          payload: terminal
+            ? redactedGitHubWebhookIngressPayload(input.payload)
+            : input.payload,
+          result: {
+            attempts: input.attempts,
+            ...(input.cancelledCode ? { code: input.cancelledCode } : {}),
+            ...(input.response
+              ? { httpStatus: input.response.status }
+              : { error: redactError(input.error) }),
+            retryable: !terminal,
+            ...(!terminal ? { retryAt: retryAt.toISOString() } : {}),
+          },
+          updatedAt: new Date(),
+        })
+        .where(
+          and(
+            eq(chatActions.id, actionId),
+            eq(chatActions.kind, "github_webhook_ingress"),
+            eq(chatActions.status, "processing"),
+            sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`,
+          ),
+        );
+    });
   }
 
   async function processGitHubWebhookIngress(
@@ -25361,18 +25359,21 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
       return githubIngressAcceptedResponse();
     const payload = githubWebhookIngressPayload(action.payload);
     if (!payload) {
-      await db
-        .update(chatActions)
-        .set({
-          status: "failed",
-          payload: {},
-          result: {
-            code: "github_webhook_ingress_payload_invalid",
-            retryable: false,
-          },
-          updatedAt: new Date(),
-        })
-        .where(eq(chatActions.id, action.id));
+      await db.transaction(async (tx) => {
+        await notifyChatActionWork(tx, "github_webhook_ingress");
+        return tx
+          .update(chatActions)
+          .set({
+            status: "failed",
+            payload: {},
+            result: {
+              code: "github_webhook_ingress_payload_invalid",
+              retryable: false,
+            },
+            updatedAt: new Date(),
+          })
+          .where(eq(chatActions.id, action.id));
+      });
       return githubIngressAcceptedResponse();
     }
     const now = new Date();
@@ -25407,21 +25408,24 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
               sql`coalesce((${chatActions.result}->>'attempts')::int, 0) = ${previousAttempts}`,
             )
           : eq(chatActions.status, "received");
-    const [claimed] = await db
-      .update(chatActions)
-      .set({
-        status: "processing",
-        result: { attempts },
-        updatedAt: now,
-      })
-      .where(
-        and(
-          eq(chatActions.id, action.id),
-          eq(chatActions.kind, "github_webhook_ingress"),
-          claimGuard,
-        ),
-      )
-      .returning();
+    const [claimed] = await db.transaction(async (tx) => {
+      await notifyChatActionWork(tx, "github_webhook_ingress");
+      return tx
+        .update(chatActions)
+        .set({
+          status: "processing",
+          result: { attempts },
+          updatedAt: now,
+        })
+        .where(
+          and(
+            eq(chatActions.id, action.id),
+            eq(chatActions.kind, "github_webhook_ingress"),
+            claimGuard,
+          ),
+        )
+        .returning();
+    });
     if (!claimed) return null;
     action = claimed;
 
@@ -26192,28 +26196,13 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     onlyActionId?: string,
   ) {
     const now = new Date();
-    const staleBefore = new Date(now.getTime() - PROVIDER_EFFECT_STALE_MS);
     const actions = await db
       .select({ id: chatActions.id })
       .from(chatActions)
-      .where(
-        and(
-          eq(chatActions.kind, "github_webhook_ingress"),
-          ...(onlyActionId ? [eq(chatActions.id, onlyActionId)] : []),
-          or(
-            eq(chatActions.status, "received"),
-            and(
-              eq(chatActions.status, "processing"),
-              lte(chatActions.updatedAt, staleBefore),
-            ),
-            and(
-              eq(chatActions.status, "failed"),
-              sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
-              sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
-            ),
-          ),
-        ),
-      )
+      .where(and(
+        onlyActionId ? eq(chatActions.id, onlyActionId) : undefined,
+        dueChatAction("github_webhook_ingress", now.getTime()),
+      ))
       .orderBy(asc(chatActions.createdAt))
       .limit(limit);
     for (const action of actions) await processGitHubWebhookIngress(action.id);
@@ -28092,8 +28081,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
 
   async function processPendingChatMaintenance(limit = 25) {
     return Promise.allSettled([
-      processPendingGitHubWebhookIngress(limit),
-      processPendingProviderEffects(limit),
       processPendingTelegramMaintenance(limit),
       processFailedChatRunRetries(limit),
     ]);
@@ -28106,7 +28093,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     // so a crashed processing claim is quarantined before the delivery worker
     // could otherwise replay it.
     const actionRecovery = onlyDeliveryId || !maintenance ? null : Promise.allSettled([
-      processPendingChatMaintenance(limit), processPendingSlackBoardMessages(limit),
+      processPendingChatMaintenance(limit), processPendingGitHubWebhookIngress(limit),
+      processPendingProviderEffects(limit), processPendingSlackBoardMessages(limit),
       processPendingSlackTaskStarts(limit), processPendingReceiptReactions(limit),
       processPendingSlackSessionStops(limit), slackRegistration.processPendingVerificationMessages(limit),
     ]);
@@ -30095,6 +30083,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
                 );
             }
             const now = new Date();
+            await notifyChatActionWork(tx, "provider_effect");
             const [claimed] = await tx
               .update(chatActions)
               .set({
@@ -30206,6 +30195,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
           const now = new Date();
           const nextStatus =
             resolution === "mark_delivered" ? "processed" : "cancelled";
+          await notifyChatActionWork(tx, "provider_effect");
           const [resolved] = await tx
             .update(chatActions)
             .set({
@@ -36931,43 +36921,36 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
   ) {
     const now = new Date();
     const actions = await db
-      .select()
+      .select({ action: chatActions })
       .from(chatActions)
-      .where(
-        and(
-          eq(chatActions.kind, "slack_session_sync"),
-          ...(onlyActionId ? [eq(chatActions.id, onlyActionId)] : []),
-          or(
-            and(
-              eq(chatActions.status, "received"),
-              sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
-            ),
-            and(
-              eq(chatActions.status, "processing"),
-              lte(chatActions.updatedAt, new Date(now.getTime() - 60_000)),
-            ),
-          ),
-        ),
-      )
+      .leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
+      .where(and(
+        onlyActionId ? eq(chatActions.id, onlyActionId) : undefined,
+        dueChatAction("slack_session_sync", now.getTime()),
+      ))
       .orderBy(asc(chatActions.updatedAt))
-      .limit(limit);
+      .limit(limit)
+      .then(rows => rows.map(row => row.action));
     await options.slackSessionSyncSelectionBarrier?.();
     for (const action of actions) {
       const payload = slackSessionSyncPayload(action.payload);
       if (!payload || !action.conversationId) {
-        await db
-          .update(chatActions)
-          .set({
-            status: "failed",
-            result: { code: "slack_session_payload_invalid" },
-            updatedAt: new Date(),
-          })
-          .where(
-            and(
-              eq(chatActions.id, action.id),
-              eq(chatActions.updatedAt, action.updatedAt),
-            ),
-          );
+        await db.transaction(async (tx) => {
+          await notifyChatActionWork(tx, "slack_session_sync");
+          return tx
+            .update(chatActions)
+            .set({
+              status: "failed",
+              result: { code: "slack_session_payload_invalid" },
+              updatedAt: new Date(),
+            })
+            .where(
+              and(
+                eq(chatActions.id, action.id),
+                eq(chatActions.updatedAt, action.updatedAt),
+              ),
+            );
+        });
         continue;
       }
       const revisionWhere = and(
@@ -36980,26 +36963,19 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
         eq(chatActions.updatedAt, action.updatedAt),
       );
       const record = await endpointRecord(action.endpointId);
-      if (record && ["paused", "attention"].includes(record.endpoint.status)) {
-        await db
-          .update(chatActions)
-          .set({
-            status: "received",
-            result: { retryAt: new Date(Date.now() + 30_000).toISOString() },
-            updatedAt: new Date(),
-          })
-          .where(selectedSnapshotWhere);
-        continue;
-      }
+      if (record && ["paused", "attention"].includes(record.endpoint.status)) continue;
       if (!record || record.endpoint.provider !== "slack") {
-        await db
-          .update(chatActions)
-          .set({
-            status: "cancelled",
-            result: { code: "slack_session_endpoint_unavailable" },
-            updatedAt: new Date(),
-          })
-          .where(selectedSnapshotWhere);
+        await db.transaction(async (tx) => {
+          await notifyChatActionWork(tx, "slack_session_sync");
+          return tx
+            .update(chatActions)
+            .set({
+              status: "cancelled",
+              result: { code: "slack_session_endpoint_unavailable" },
+              updatedAt: new Date(),
+            })
+            .where(selectedSnapshotWhere);
+        });
         continue;
       }
       const attempts =
@@ -37084,6 +37060,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
                 ? endpoint.allowDirectMessages
                 : nonDirectDestinationAllowed(endpoint, resource))
             ) {
+              await notifyChatActionWork(tx, "slack_session_sync");
               await tx
                 .update(chatActions)
                 .set({
@@ -37118,6 +37095,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
               .limit(1)
               .then((rows) => rows[0] ?? null);
             if (!publication) {
+              await notifyChatActionWork(tx, "slack_session_sync");
               await tx
                 .update(chatActions)
                 .set({
@@ -37150,6 +37128,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
               if (!run || !["queued", "running"].includes(run.status))
                 status = "active";
             }
+            await notifyChatActionWork(tx, "slack_session_sync");
             const claimed = await tx
               .update(chatActions)
               .set({
@@ -37195,6 +37174,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
               outcome === "updated" && claim.status === "processing"
                 ? 30 * 60_000
                 : null;
+            await notifyChatActionWork(tx, "slack_session_sync");
             await tx
               .update(chatActions)
               .set({
@@ -37232,35 +37212,38 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
           disposition.kind === "retry"
             ? disposition.retryAfterMs
             : Math.min(5 * 60_000, 1_000 * 2 ** Math.min(attempts, 8));
-        await db
-          .update(chatActions)
-          .set({
-            status: retryable ? "received" : "failed",
-            result: {
-              attempts,
-              code: retryable
-                ? "slack_session_sync_retry"
-                : "slack_session_sync_rejected",
-              ...(retryable
-                ? { retryAt: new Date(Date.now() + retryMs).toISOString() }
-                : {}),
-            },
-            updatedAt: new Date(),
-          })
-          .where(
-            and(
-              revisionWhere,
-              ownsAttempt
-                ? and(
-                    eq(chatActions.status, "processing"),
-                    sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
-                  )
-                : and(
-                    eq(chatActions.status, "received"),
-                    eq(chatActions.updatedAt, action.updatedAt),
-                  ),
-            ),
-          );
+        await db.transaction(async (tx) => {
+          await notifyChatActionWork(tx, "slack_session_sync");
+          return tx
+            .update(chatActions)
+            .set({
+              status: retryable ? "received" : "failed",
+              result: {
+                attempts,
+                code: retryable
+                  ? "slack_session_sync_retry"
+                  : "slack_session_sync_rejected",
+                ...(retryable
+                  ? { retryAt: new Date(Date.now() + retryMs).toISOString() }
+                  : {}),
+              },
+              updatedAt: new Date(),
+            })
+            .where(
+              and(
+                revisionWhere,
+                ownsAttempt
+                  ? and(
+                      eq(chatActions.status, "processing"),
+                      sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
+                    )
+                  : and(
+                      eq(chatActions.status, "received"),
+                      eq(chatActions.updatedAt, action.updatedAt),
+                    ),
+              ),
+            );
+        });
         logger.warn(
           {
             endpointId: action.endpointId,

server/src/services/chat-slack-session-outbox.ts

@@ -1,6 +1,8 @@
 import { sql } from "drizzle-orm";
 import { chatActions, type Db } from "@paperclipai/db";
 
+import { notifyChatActionWork } from "./chat-work-notifications.js";
+
 type DbOrTransaction = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 export interface SlackSessionSyncPayload {
@@ -47,6 +49,7 @@ export async function stageSlackSessionSync(
     runtimeGeneration: input.runtimeGeneration,
     credentialFingerprint: input.credentialFingerprint,
   };
+  await notifyChatActionWork(database, "slack_session_sync");
   await database
     .insert(chatActions)
     .values({

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

@@ -11,6 +11,9 @@ export async function notifyChatPublicationWork(tx: object): Promise<void> {
 }
 
 export const CHAT_ACTION_QUEUES = {
+  provider_effect: DELIVERY_QUEUES.chatProviderEffects,
+  github_webhook_ingress: DELIVERY_QUEUES.chatGitHubIngress,
+  slack_session_sync: DELIVERY_QUEUES.chatSessionSyncs,
   slack_board_message: DELIVERY_QUEUES.chatBoardMessages,
   slash_task_start: DELIVERY_QUEUES.chatTaskStarts,
   receipt_reaction: DELIVERY_QUEUES.chatReceiptReactions,

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

@@ -6,6 +6,9 @@ export const DELIVERY_QUEUES = {
   chatTaskStarts: "chat-task-starts",
   chatReceiptReactions: "chat-receipt-reactions",
   chatSessionStops: "chat-session-stops",
+  chatProviderEffects: "chat-provider-effects",
+  chatGitHubIngress: "chat-github-ingress",
+  chatSessionSyncs: "chat-session-syncs",
   chatVerification: "chat-verification",
   chatInbound: "chat-inbound",
   chatPublications: "chat-publications",