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({