From b1ebe3de3c17610573de3e849fc245e2aace7742 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 10 Oct 2026 13:14:23 -0500 Subject: [PATCH] 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 --- ...-instance-sleep-and-scheduler-inventory.md | 14 +- .../app-private-hostname-gate.test.ts | 1 + server/src/__tests__/chat-action-work.test.ts | 109 +++ .../chat-channels.integration.test.ts | 145 +++- server/src/app.ts | 2 + server/src/services/chat-action-work.ts | 85 +++ server/src/services/chat-channels.ts | 692 +++++++++--------- server/src/services/chat-delivery-work.ts | 32 +- server/src/services/chat-receipt-reactions.ts | 2 + .../src/services/chat-slack-registration.ts | 34 +- .../src/services/chat-work-notifications.ts | 18 + .../services/delivery-work-notifications.ts | 5 + server/src/services/slack-board-messages.ts | 3 +- 13 files changed, 761 insertions(+), 381 deletions(-) create mode 100644 server/src/__tests__/chat-action-work.test.ts create mode 100644 server/src/services/chat-action-work.ts diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index fe1b460ff8a..d818caa41c6 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/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. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 11cb68ea391..61893417dfd 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/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 ="), diff --git a/server/src/__tests__/chat-action-work.test.ts b/server/src/__tests__/chat-action-work.test.ts new file mode 100644 index 00000000000..af541a60c07 --- /dev/null +++ b/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[] = []; +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) { + 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 => null), + nextVerificationMessageAt: vi.fn(async (): Promise => 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); + }); +}); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 5c15c718f6f..118b47a8a81 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/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 = {}) => 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(resolve => { release = resolve; }); + const staged = new Promise(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) { diff --git a/server/src/app.ts b/server/src/app.ts index 7fcff18b2d8..003d587c551 100644 --- a/server/src/app.ts +++ b/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(); diff --git a/server/src/services/chat-action-work.ts b/server/src/services/chat-action-work.ts new file mode 100644 index 00000000000..42993950548 --- /dev/null +++ b/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`case + when ${chatActions.status} = 'received' and ${chatEndpoints.status} not in ('paused', 'attention') then 0::numeric end`; + if (kind === "slash_task_start") return sql`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`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 { + const [row] = await db.select({ at: sql`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; + processPendingSlackTaskStarts(): Promise; + processPendingReceiptReactions(): Promise; + processPendingSlackSessionStops(): Promise; + processPendingVerificationMessages(): Promise; + nextChatActionAt(kind: ScheduledChatAction): Promise; + nextVerificationMessageAt(): Promise; +}; + +export function registerChatActionWork( + coordinator: ReturnType, + 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)) }; +} diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index d6b1d41b9ce..d81f4ea8744 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,4 +1,6 @@ -import { notifyChatDeliveryWork, notifyChatPublicationWork, notifyChatEndpointWork } from "./chat-work-notifications.js"; +import { dueChatAction, nextChatActionAt, SLACK_COMMAND_POST_STALE_MS, SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, SLACK_COMMAND_ADMISSION_STALE_MS, PROVIDER_EFFECT_STALE_MS } from "./chat-action-work.js"; +import type { ScheduledChatAction } from "./chat-work-notifications.js"; +import { notifyChatDeliveryWork, notifyChatPublicationWork, notifyChatEndpointWork, notifyChatActionWork, notifyChatVerificationWork } from "./chat-work-notifications.js"; import { syncSpekoVoiceTools } from "./voice/speko-agent-tools.js"; import { configureSpekoSessionTools } from "./voice/speko-tool-setup.js"; import { voiceSessionService } from "./voice/voice-session-service.js"; @@ -862,10 +864,6 @@ const REACTION_LINK_MAX_AGE_MS = 2 * 60_000; const DISCORD_ROOT_THREAD_CONFIRMATION_DELAY_MS = 5 * 60_000; const DELIVERY_LEASE_TTL_MS = 90_000; const DELIVERY_DRAIN_LIMIT = 100; -const SLACK_COMMAND_POST_STALE_MS = 60_000; -const SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS = 5 * 60_000; -const SLACK_COMMAND_ADMISSION_STALE_MS = 60_000; -const PROVIDER_EFFECT_STALE_MS = 60_000; const SLACK_FILE_RECEIPT_MAX_AGE_MS = 24 * 60 * 60_000; const SLACK_FILE_RECEIPT_MAX_ATTEMPTS = 12; const SLACK_FILE_RECEIPT_STALE_MS = 60_000; @@ -3472,27 +3470,30 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // endpoint again after a rotation and accidentally blessing an obsolete // adapter with the new generation. const context = input.runtimeContext ?? runtimeContextForRecord(record); - const [inserted] = await db - .insert(chatActions) - .values({ - companyId: input.endpoint.companyId, - endpointId: input.endpoint.id, - deliveryId: input.deliveryId, - kind: "receipt_reaction", - providerActionId: `receipt_reaction:${input.deliveryId}`, - payload: { - version: 1, - operation: "add", - threadId: input.thread.id, - messageId: input.message.id, - reaction: "eyes", - runtimeGeneration: context.generation, - credentialFingerprint: context.credentialFingerprint, - } satisfies ReceiptReactionPayload, - status: "received", - }) - .onConflictDoNothing() - .returning({ id: chatActions.id }); + const [inserted] = await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "receipt_reaction"); + return tx + .insert(chatActions) + .values({ + companyId: input.endpoint.companyId, + endpointId: input.endpoint.id, + deliveryId: input.deliveryId, + kind: "receipt_reaction", + providerActionId: `receipt_reaction:${input.deliveryId}`, + payload: { + version: 1, + operation: "add", + threadId: input.thread.id, + messageId: input.message.id, + reaction: "eyes", + runtimeGeneration: context.generation, + credentialFingerprint: context.credentialFingerprint, + } satisfies ReceiptReactionPayload, + status: "received", + }) + .onConflictDoNothing() + .returning({ id: chatActions.id }); + }); const actionId = inserted?.id ?? (await db @@ -3621,17 +3622,20 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { if (!action || ["processed", "cancelled"].includes(action.status)) return; const payload = receiptReactionPayload(action.payload); if (!payload) { - await db - .update(chatActions) - .set({ - status: "failed", - result: { - code: "receipt_reaction_payload_invalid", - retryable: false, - }, - updatedAt: new Date(), - }) - .where(eq(chatActions.id, action.id)); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "receipt_reaction"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + code: "receipt_reaction_payload_invalid", + retryable: false, + }, + updatedAt: new Date(), + }) + .where(eq(chatActions.id, action.id)); + }); return; } if (payload.operation === "add" && action.deliveryId) { @@ -3651,6 +3655,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { if (terminalRemoval) { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "receipt_reaction"); await tx .update(chatActions) .set({ @@ -3680,24 +3685,27 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // Adding or removing the same reaction is idempotent. A worker // interruption can be retried safely without an ambiguous // duplicate-message boundary. - await db - .update(chatActions) - .set({ - status: "failed", - result: { - ...(action.result ?? {}), - retryable: true, - retryAt: new Date().toISOString(), - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "processing"), - eq(chatActions.updatedAt, action.updatedAt), - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "receipt_reaction"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + ...(action.result ?? {}), + retryable: true, + retryAt: new Date().toISOString(), + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "processing"), + eq(chatActions.updatedAt, action.updatedAt), + ), + ); + }); action = await db .select() .from(chatActions) @@ -3723,6 +3731,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ); const [claimed] = await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "receipt_reaction"); const rows = await tx .update(chatActions) .set({ @@ -3760,6 +3769,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ) { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "receipt_reaction"); await tx .update(chatActions) .set({ @@ -3853,6 +3863,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // for an obsolete runtime. await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "receipt_reaction"); const processed = await tx .update(chatActions) .set({ @@ -3898,6 +3909,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ) { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "receipt_reaction"); const processed = await tx .update(chatActions) .set({ @@ -3957,6 +3969,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { : null; await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "receipt_reaction"); const [failed] = await tx .update(chatActions) .set({ @@ -4013,31 +4026,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { limit = 25, 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, "receipt_reaction"), - ...(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)`, - ), - ), - ), - ) - .orderBy(asc(chatActions.createdAt)) - .limit(limit); + const actions = await db.select({ id: chatActions.id }).from(chatActions) + .where(and(dueChatAction("receipt_reaction"), onlyActionId ? eq(chatActions.id, onlyActionId) : undefined)) + .orderBy(asc(chatActions.createdAt)).limit(limit); for (const action of actions) await processReceiptReaction(action.id); return actions.length; } @@ -22226,6 +22217,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ); const finalized = await db.transaction(async (tx) => { await input.credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); const [ownedAction] = await tx .update(chatActions) .set({ @@ -22321,28 +22313,31 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const staleAdmission = new Date( Date.now() - SLACK_COMMAND_ADMISSION_STALE_MS, ); - const [action] = await db - .update(chatActions) - .set({ - status: "admitting", - result: sql`coalesce(${chatActions.result}, '{}'::jsonb) - || jsonb_build_object('admissionClaimId', ${admissionClaimId}::text)`, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, actionId), - eq(chatActions.kind, "slash_task_start"), - or( - eq(chatActions.status, "provider_confirmed"), - and( - eq(chatActions.status, "admitting"), - lte(chatActions.updatedAt, staleAdmission), + const [action] = await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "admitting", + result: sql`coalesce(${chatActions.result}, '{}'::jsonb) + || jsonb_build_object('admissionClaimId', ${admissionClaimId}::text)`, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, actionId), + eq(chatActions.kind, "slash_task_start"), + or( + eq(chatActions.status, "provider_confirmed"), + and( + eq(chatActions.status, "admitting"), + lte(chatActions.updatedAt, staleAdmission), + ), ), ), - ), - ) - .returning(); + ) + .returning(); + }); if (!action) return false; await options.slackTaskAdmissionClaimBarrier?.({ actionId: action.id, @@ -22358,18 +22353,21 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { sql`${chatActions.result}->>'admissionClaimId' = ${admissionClaimId}`, ); const failAdmission = async (code: string, detail: string) => { - await db - .update(chatActions) - .set({ - status: "failed", - result: { - ...(confirmed ?? {}), - code, - redactedError: redactSensitiveText(detail).slice(0, MAX_ERROR_TEXT), - }, - updatedAt: new Date(), - }) - .where(ownsAdmission()); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + ...(confirmed ?? {}), + code, + redactedError: redactSensitiveText(detail).slice(0, MAX_ERROR_TEXT), + }, + updatedAt: new Date(), + }) + .where(ownsAdmission()); + }); }; if (!recovery || !confirmed || !action.principalId) { await failAdmission( @@ -22397,14 +22395,17 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // The provider-visible send is already confirmed. Wait for an // operator to repair/resume the endpoint, then continue Paperclip-only // admission without sending another Slack message. - await db - .update(chatActions) - .set({ - status: "provider_confirmed", - result: confirmed, - updatedAt: new Date(), - }) - .where(ownsAdmission()); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "provider_confirmed", + result: confirmed, + updatedAt: new Date(), + }) + .where(ownsAdmission()); + }); } return true; } @@ -22469,27 +22470,34 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { false, true, ); - await db - .update(chatActions) - .set({ status: "processed", result: confirmed, updatedAt: new Date() }) - .where(ownsAdmission()); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ status: "processed", result: confirmed, updatedAt: new Date() }) + .where(ownsAdmission()); + }); return true; } catch (error) { // The provider post is already confirmed, so this retry is strictly a // Paperclip admission retry. Preserve the root tuple and let the durable // reconciler resume without ever posting to Slack again. - await db - .update(chatActions) - .set({ - status: "provider_confirmed", - result: { - ...confirmed, - code: "slash_task_admission_retry", - redactedError: redactError(error), - }, - updatedAt: new Date(), - }) - .where(ownsAdmission()); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "provider_confirmed", + result: { + ...confirmed, + code: "slash_task_admission_retry", + retryAt: new Date(Date.now() + 1_000).toISOString(), + redactedError: redactError(error), + }, + updatedAt: new Date(), + }) + .where(ownsAdmission()); + }); throw error; } } @@ -22528,6 +22536,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async (credentialLease) => { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); await tx .update(chatActions) .set({ @@ -22570,6 +22579,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async (credentialLease) => { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); await tx .update(chatActions) .set({ status: "queued", updatedAt: new Date() }) @@ -22609,21 +22619,24 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? Math.max(0, action.result.attemptCount) : 0) + 1; - const [claimed] = await db - .update(chatActions) - .set({ - status: "validating", - result: { attemptCount: attempt }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.kind, "slash_task_start"), - eq(chatActions.status, action.status), - ), - ) - .returning(); + const [claimed] = await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "validating", + result: { attemptCount: attempt }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.kind, "slash_task_start"), + eq(chatActions.status, action.status), + ), + ) + .returning(); + }); if (!claimed) return false; action = claimed; @@ -22634,24 +22647,27 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { !initialRecord || initialRecord.endpoint.provider !== "slack" ) { - await db - .update(chatActions) - .set({ - status: "failed", - result: { - code: "slash_task_provider_context_missing", - retryable: false, - attemptCount: attempt, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "validating"), - sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + code: "slash_task_provider_context_missing", + retryable: false, + attemptCount: attempt, + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "validating"), + sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, + ), + ); + }); return true; } let providerPostStarted = false; @@ -22666,53 +22682,59 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async (credentialLease) => { const record = await endpointRecord(action.endpointId); if (!record || record.endpoint.provider !== "slack") { - await db - .update(chatActions) - .set({ - status: "failed", - result: { - code: "slash_task_provider_context_missing", - retryable: false, - attemptCount: attempt, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "validating"), - sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + code: "slash_task_provider_context_missing", + retryable: false, + attemptCount: attempt, + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "validating"), + sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, + ), + ); + }); return true; } failureRuntimeFence = runtimeContextForRecord(record); if (!["verifying", "active"].includes(record.endpoint.status)) { const terminallyArchived = record.endpoint.status === "archived"; - await db - .update(chatActions) - .set({ - status: terminallyArchived ? "cancelled" : "queued", - result: { - code: terminallyArchived - ? "slash_task_endpoint_archived" - : "slash_task_endpoint_waiting", - attemptCount: attempt - 1, - ...(terminallyArchived - ? {} - : { - retryAt: new Date(Date.now() + 5_000).toISOString(), - }), - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "validating"), - sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: terminallyArchived ? "cancelled" : "queued", + result: { + code: terminallyArchived + ? "slash_task_endpoint_archived" + : "slash_task_endpoint_waiting", + attemptCount: attempt - 1, + ...(terminallyArchived + ? {} + : { + retryAt: new Date(Date.now() + 5_000).toISOString(), + }), + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "validating"), + sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, + ), + ); + }); return false; } @@ -22787,23 +22809,26 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? record.endpoint.allowDirectMessages : nonDirectDestinationAllowed(record.endpoint, resource); if (!principalAllowed || !destinationAllowed) { - await db - .update(chatActions) - .set({ - status: "cancelled", - result: { - code: "slash_task_no_longer_authorized", - attemptCount: attempt, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "validating"), - sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + status: "cancelled", + result: { + code: "slash_task_no_longer_authorized", + attemptCount: attempt, + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "validating"), + sql`(${chatActions.result}->>'attemptCount')::int = ${attempt}`, + ), + ); + }); return true; } @@ -22855,6 +22880,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { !currentAuthorization?.allowed || !currentDestinationAllowed ) { + await notifyChatActionWork(tx, "slash_task_start"); await tx .update(chatActions) .set({ @@ -22876,6 +22902,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } failureResource = currentResource; + await notifyChatActionWork(tx, "slash_task_start"); const [transportClaim] = await tx .update(chatActions) .set({ @@ -22934,6 +22961,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const threadId = slackTaskStarterThreadId(rawChannelId, starter.id); const [providerConfirmed] = await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); return tx .update(chatActions) .set({ @@ -22997,62 +23025,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } } - async function processPendingSlackTaskStarts(limit: number) { - const now = new Date(); - const staleAdmission = new Date( - now.getTime() - SLACK_COMMAND_ADMISSION_STALE_MS, - ); - const staleProviderPost = new Date( - now.getTime() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, - ); - const staleValidation = new Date( - now.getTime() - SLACK_COMMAND_POST_STALE_MS, - ); - const actions = await db - .select({ id: chatActions.id }) - .from(chatActions) - .leftJoin(chatEndpoints, eq(chatEndpoints.id, chatActions.endpointId)) - .where( - // Provider-confirmed rows need no provider retry and have no due - // timestamp. Park them while an endpoint is repairable but paused - // or in attention; otherwise an old page can starve active work - // forever. Archived and missing endpoints remain eligible so their - // actions can be terminalized. - and( - eq(chatActions.kind, "slash_task_start"), - or( - and( - eq(chatActions.status, "queued"), - sql`(${chatActions.result}->>'retryAt' is null or ${chatActions.result}->>'retryAt' <= ${now.toISOString()})`, - ), - and( - eq(chatActions.status, "provider_confirmed"), - or( - isNull(chatEndpoints.id), - inArray(chatEndpoints.status, [ - "verifying", - "active", - "archived", - ]), - ), - ), - and( - eq(chatActions.status, "admitting"), - lte(chatActions.updatedAt, staleAdmission), - ), - and( - eq(chatActions.status, "resolving"), - lte(chatActions.updatedAt, staleProviderPost), - ), - and( - eq(chatActions.status, "validating"), - lte(chatActions.updatedAt, staleValidation), - ), - ), - ), - ) - .orderBy(asc(chatActions.createdAt)) - .limit(limit); + async function processPendingSlackTaskStarts(limit = 25) { + const actions = await db.select({ id: chatActions.id }).from(chatActions) + .leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId))) + .where(dueChatAction("slash_task_start")) + .orderBy(asc(chatActions.createdAt)).limit(limit); let processed = 0; for (let offset = 0; offset < actions.length; offset += 4) { const results = await Promise.all( @@ -23770,6 +23747,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { )) ) return null; + await notifyChatActionWork(tx, "slash_task_start"); return tx .insert(chatActions) .values({ @@ -26385,6 +26363,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { threadId: event.threadId, userId: event.userId, }; + await notifyChatActionWork(tx, "slack_session_stop"); const [inserted] = await tx .insert(chatActions) .values({ @@ -26453,18 +26432,21 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { : 0; const attempts = previousAttempts + 1; if (!payload) { - await db - .update(chatActions) - .set({ - status: "failed", - result: { - attempts, - code: "slack_session_stop_payload_invalid", - retryable: false, - }, - updatedAt: now, - }) - .where(eq(chatActions.id, action.id)); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slack_session_stop"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + attempts, + code: "slack_session_stop_payload_invalid", + retryable: false, + }, + updatedAt: now, + }) + .where(eq(chatActions.id, action.id)); + }); return; } @@ -26502,6 +26484,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { actorUserId: string | null, targetKind: "run" | "wakeup" | "none", ): Promise => { + await notifyChatActionWork(tx, "slack_session_stop"); await tx .update(chatActions) .set({ @@ -26741,6 +26724,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } runId = wakeup?.runId ?? null; if (!runId && wakeup?.status === "claimed") { + await notifyChatActionWork(tx, "slack_session_stop"); await tx .update(chatActions) .set({ @@ -26809,6 +26793,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { "run", ); } + await notifyChatActionWork(tx, "slack_session_stop"); await tx .update(chatActions) .set({ @@ -26858,6 +26843,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { .then((rows) => rows[0] ?? null); const cancellationWon = authoritativeRun?.status === "cancelled"; await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slack_session_stop"); const [settled] = await tx .update(chatActions) .set({ @@ -26903,32 +26889,35 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); } catch (error) { const retryable = attempts < 5; - await db - .update(chatActions) - .set({ - status: "failed", - result: { - attempts, - code: "slack_session_stop_cancellation_failed", - retryable, - ...(retryable - ? { - retryAt: new Date( - Date.now() + Math.min(60_000, 1_000 * 2 ** (attempts - 1)), - ).toISOString(), - } - : {}), - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "processing"), - sql`(${chatActions.result}->>'attempts')::int = ${attempts}`, - sql`${chatActions.result}->>'runId' = ${claim.runId}`, - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slack_session_stop"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { + attempts, + code: "slack_session_stop_cancellation_failed", + retryable, + ...(retryable + ? { + retryAt: new Date( + Date.now() + Math.min(60_000, 1_000 * 2 ** (attempts - 1)), + ).toISOString(), + } + : {}), + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "processing"), + sql`(${chatActions.result}->>'attempts')::int = ${attempts}`, + sql`${chatActions.result}->>'runId' = ${claim.runId}`, + ), + ); + }); logger.warn( { actionId: action.id, @@ -26945,31 +26934,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { limit = 25, 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, "slack_session_stop"), - ...(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)`, - ), - ), - ), - ) - .orderBy(asc(chatActions.createdAt)) - .limit(limit); + const actions = await db.select({ id: chatActions.id }).from(chatActions) + .where(and(dueChatAction("slack_session_stop"), onlyActionId ? eq(chatActions.id, onlyActionId) : undefined)) + .orderBy(asc(chatActions.createdAt)).limit(limit); for (const action of actions) await processSlackSessionStop(action.id); return actions.length; } @@ -27005,6 +26972,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const observedAt = new Date(); const setup = current.setup as InternalSetupState; const matchesCurrentUrl = slackCallbackMatchesPublicUrl(callback.url, `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${current.publicId}/slack`); + await notifyChatVerificationWork(tx); await tx.update(chatEndpoints).set({ setup: { ...setup, ...(matchesCurrentUrl ? { webhookVerifiedAt: setup.webhookVerifiedAt ?? observedAt.toISOString(), @@ -27622,6 +27590,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } } connectionProved = signingFingerprint !== null; + if (connectionProved) await notifyChatVerificationWork(tx); await tx .update(chatEndpoints) .set({ @@ -28045,7 +28014,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPendingSlackBoardMessages(limit = 25) { const pending = await db.select({ id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId }).from(chatActions) .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId))) - .where(and(eq(chatActions.kind, "slack_board_message"), eq(chatActions.status, "received"), notInArray(chatEndpoints.status, ["paused", "attention"]))) + .where(dueChatAction("slack_board_message")) .orderBy(asc(chatActions.createdAt)).limit(limit); for (const candidate of pending) { await withSlackBoardLease(db, { ...candidate, actionId: candidate.id }, fetchImpl, async (lease) => { @@ -28057,6 +28026,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const commentId = String(action.payload.commentId); async function cancelWork(message: string) { await lease.commit(async (tx) => { + await notifyChatActionWork(tx, "slack_board_message"); await tx.update(chatActions).set({ status: "cancelled", result: { code: "slack_board_work_not_authorized", message }, updatedAt: new Date() }).where(eq(chatActions.id, action.id)); await notifyChatPublicationWork(tx); await tx.update(chatPublications).set({ state: "cancelled", redactedError: message, nextAttemptAt: null, updatedAt: new Date() }).where(and( @@ -28091,6 +28061,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { await issueService(tx as unknown as Db).update(current.id, { status: "todo", actorUserId: userId }); await logActivity(tx as unknown as Db, { companyId: current.companyId, actorType: "user", actorId: userId, action: "issue.updated", entityType: "issue", entityId: current.id, details: { status: "todo", source: "slack_board_message" } }); } + await notifyChatActionWork(tx, "slack_board_message"); await tx.update(chatActions).set({ result: { resumeApplied: true }, updatedAt: new Date() }).where(eq(chatActions.id, action.id)); } return current; @@ -28105,6 +28076,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); await lease.commit(async (tx) => { await notifyChatPublicationWork(tx); + await notifyChatActionWork(tx, "slack_board_message"); await tx.update(chatActions).set({ status: "processed", updatedAt: new Date() }).where(eq(chatActions.id, action.id)); }); } catch (error) { @@ -28120,17 +28092,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPendingChatMaintenance(limit = 25) { return Promise.allSettled([ - processPendingSlackBoardMessages(limit), processPendingGitHubWebhookIngress(limit), processPendingProviderEffects(limit), - slackRegistration.processPendingVerificationMessages(limit), - processPendingReceiptReactions(limit), - processPendingSlackSessionStops(limit), processPendingTelegramMaintenance(limit), - // Slack task starts are an action-backed outbox. Queued work may - // post once; provider-confirmed rows perform Paperclip-only - // admission; a stale in-flight post is quarantined as unknown. - processPendingSlackTaskStarts(limit), processFailedChatRunRetries(limit), ]); } @@ -28141,7 +28105,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // chat_actions as their outbox. Reconcile them before inbound deliveries // so a crashed processing claim is quarantined before the delivery worker // could otherwise replay it. - const actionRecovery = onlyDeliveryId || !maintenance ? null : processPendingChatMaintenance(limit); + const actionRecovery = onlyDeliveryId || !maintenance ? null : Promise.allSettled([ + processPendingChatMaintenance(limit), processPendingSlackBoardMessages(limit), + processPendingSlackTaskStarts(limit), processPendingReceiptReactions(limit), + processPendingSlackSessionStops(limit), slackRegistration.processPendingVerificationMessages(limit), + ]); const reactionRecovery = processPendingReactionDeliveries( limit, onlyDeliveryId, @@ -30373,6 +30341,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async (credentialLease) => { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); await tx .update(chatActions) .set({ @@ -30428,6 +30397,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ); } const now = new Date(); + await notifyChatActionWork(tx, "slash_task_start"); await tx .update(chatActions) .set({ @@ -30612,6 +30582,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { : 0) + 1; const [claimed] = await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); await tx .update(chatActions) .set({ @@ -30717,6 +30688,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }, ); } + await notifyChatActionWork(tx, "slash_task_start"); const rows = await tx .update(chatActions) .set({ @@ -30778,6 +30750,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } const [completed] = await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); return tx .update(chatActions) .set({ @@ -39117,6 +39090,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { processQueuedDeliveries: () => processPendingDeliveries(25, undefined, false), scheduleQueuedPublications: () => processPendingPublications(25, { waitForCompletion: false, maintenance: false }), processPendingChatMaintenance, + processPendingSlackBoardMessages, + processPendingSlackTaskStarts, + processPendingVerificationMessages: slackRegistration.processPendingVerificationMessages, + nextVerificationMessageAt: slackRegistration.nextVerificationMessageAt, + nextChatActionAt: (kind: ScheduledChatAction) => nextChatActionAt(db, kind), processPublicationMaintenance, nextInboundDeliveryAt, nextPublicationAt, diff --git a/server/src/services/chat-delivery-work.ts b/server/src/services/chat-delivery-work.ts index 1872e3b6899..ed790894d30 100644 --- a/server/src/services/chat-delivery-work.ts +++ b/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; @@ -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, + task: { queue: DeliveryQueue; run(): Promise; next(): Promise }, + 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; +} diff --git a/server/src/services/chat-receipt-reactions.ts b/server/src/services/chat-receipt-reactions.ts index c550649fba5..3a0487de660 100644 --- a/server/src/services/chat-receipt-reactions.ts +++ b/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), ); diff --git a/server/src/services/chat-slack-registration.ts b/server/src/services/chat-slack-registration.ts index a42909e7ae2..905c7ab0346 100644 --- a/server/src/services/chat-slack-registration.ts +++ b/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 { + 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)) }; } diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index f6caa18e0b9..50da6074d55 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -10,7 +10,25 @@ export async function notifyChatPublicationWork(tx: object): Promise { 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 { + await notifyDeliveryWork(tx, CHAT_ACTION_QUEUES[kind]); +} + +export async function notifyChatVerificationWork(tx: object): Promise { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatVerification); +} + export async function notifyChatEndpointWork(tx: object): Promise { await notifyChatDeliveryWork(tx); await notifyChatPublicationWork(tx); + for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue); + await notifyChatVerificationWork(tx); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 98dfb361d4f..686d860f02c 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/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", diff --git a/server/src/services/slack-board-messages.ts b/server/src/services/slack-board-messages.ts index b8fc4ddd8f3..a00a8f0f759 100644 --- a/server/src/services/slack-board-messages.ts +++ b/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({