From 1c1bf5d0191a080f3f11c1d24b87dac4a2becfe9 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 13 Oct 2026 09:23:21 -0500 Subject: [PATCH] refactor(server): move Slack recovery into channel modules Co-Authored-By: Paperclip --- server/src/services/chat-channels.ts | 558 +----------------- .../chat-channels/slack/board-messages.ts | 104 ++++ .../slack/task-start-resolution.ts | 488 +++++++++++++++ 3 files changed, 603 insertions(+), 547 deletions(-) create mode 100644 server/src/services/chat-channels/slack/board-messages.ts create mode 100644 server/src/services/chat-channels/slack/task-start-resolution.ts diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 729af208be..b83550d95e 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,5 @@ +import { createSlackTaskStartResolution } from "./chat-channels/slack/task-start-resolution.js"; +import { createSlackBoardMessages } from "./chat-channels/slack/board-messages.js"; import { PROVIDER_LABELS, CAPABILITIES } from "./chat-channels/provider-catalog.js"; import { createMessageResourceAdmission } from "./chat-channels/inbound/resource-admission.js"; import type { DiscordGatewayOwnership } from "./chat-channels/runtime-types.js"; @@ -73,7 +75,7 @@ import { createTelegramPublicationDrafts } from "./chat-channels/telegram/public import { createSlackFileReceipts } from "./chat-channels/slack/file-receipts.js"; import { createSlackSessionStatus } from "./chat-channels/slack/session-status.js"; import { createDiscordCommands } from "./chat-channels/discord/commands.js"; -import { createSlackTaskStart, slackSlashTaskRecoveryPayload, slackTaskStarterThreadId } from "./chat-channels/slack/task-start.js"; +import { createSlackTaskStart } from "./chat-channels/slack/task-start.js"; import { createGitHubWebhookIngress, type GitHubIngressStageResult } from "./chat-channels/github/webhook-ingress.js"; import { createSlackSessionStop } from "./chat-channels/slack/session-stop.js"; import { createTelegramMaintenance } from "./chat-channels/telegram/maintenance.js"; @@ -106,7 +108,7 @@ import { type GitHubLifecycleEvent, githubLifecycleEventFromRequest, } from "./chat-channels/github/messages.js"; -import { dueChatAction, nextChatActionAt, SLACK_COMMAND_POST_STALE_MS, SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS } from "./chat-action-work.js"; +import { nextChatActionAt } from "./chat-action-work.js"; import type { ScheduledChatAction } from "./chat-work-notifications.js"; import { notifyChatDeliveryWork, notifyChatPublicationWork, notifyChatEndpointWork, notifyChatActionWork, notifyChatVerificationWork } from "./chat-work-notifications.js"; @@ -127,9 +129,9 @@ import { slackChatRegistrationService } from "./chat-slack-registration.js"; import { chatSlackRegistrations } from "@paperclipai/db"; import { canActorReadIssuePrivacy, canPublishIssueToChatAudience } from "./authorization.js"; -import { withSlackBoardLease } from "./slack-board-lease.js"; -import { mirrorSlackBoardComment, slackBoardReplyBindings } from "./slack-board-messages.js"; -import { authorizeSlackBoardPublication } from "./slack-board-authority.js"; + +import { mirrorSlackBoardComment } from "./slack-board-messages.js"; + import { assertSlackBoardWorkAllowed } from "./slack-board-resume.js"; import { rememberVerifiedSlackSearchEvent } from "./connectors/slack-search-context.js"; @@ -1371,6 +1373,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { processPendingSlackSessionSyncs } = createSlackSessionStatus(db, { endpointRecord, withCredentialMutationLease, runtimeCallbackEndpoint, nonDirectDestinationAllowed, runIdFromMilestonePublication, resolveCredentials, fetchImpl, redactError, get slackSessionSyncSelectionBarrier() { return options.slackSessionSyncSelectionBarrier; } }); const { reconcileDiscordCommands, handleDiscordNativeCommand } = createDiscordCommands(db, { endpointRecord, runtimeContextForRecord, runtimeGeneration, credentialFingerprint, resolveCredentialRefs, fetchImpl, withCredentialMutationLease, runtimeCallbackRecord, ensurePrincipal, runtimeCallbackEndpoint, nonDirectDestinationAllowed, lockCurrentPrincipalAuthorization, getDiscordGatewayOwnership: (endpointId) => discordGatewayOwnerships.get(endpointId), discordGatewayRuntimeIsCurrent, stageAuthorizedTaskControlPublication, scheduleMessageProcessing, processPendingPublications }); const { latestSlackDmControlThreadId, finalizeSlackTaskStartFailure, admitConfirmedSlackTaskStart, processSlackTaskStart, processPendingSlackTaskStarts } = createSlackTaskStart(db, { endpointRecord, runtimeGeneration, credentialFingerprint, invalidateRuntime, runtimeFor, processMessage, runtimeContextForRecord, redactError, withCredentialMutationLease, sponsorAllowsGuest, nonDirectDestinationAllowed, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, get slackTaskAdmissionClaimBarrier() { return options.slackTaskAdmissionClaimBarrier; }, maxErrorText: MAX_ERROR_TEXT }); + const { resolveSlackTaskStart } = createSlackTaskStartResolution(db, { endpointRecord, withCredentialMutationLease, runtimeFor, runtimeContextForRecord, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, sponsorAllowsGuest, nonDirectDestinationAllowed, finalizeSlackTaskStartFailure, admitConfirmedSlackTaskStart, get deferWebhookProcessing() { return options.deferWebhookProcessing; } }); + const { processPendingSlackBoardMessages } = createSlackBoardMessages(db, { endpointRecord, resolveCredentials, fetchImpl, get heartbeat() { return options.heartbeat; } }); const { handleSlackSlashCommand } = createSlackCommands(db, { runtime, runtimeFor, handleSdkMessage, stageProviderEffect, scheduleProviderEffect, ensurePrincipal, runtimeCallbackEndpoint, getPublicBaseUrl, canonicalProviderResourceId, nonDirectDestinationAllowed, @@ -5459,84 +5463,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { * authenticated attachment downloads through the endpoint runtime; unsafe * or no-longer-available files are omitted without losing the text turn. */ - 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(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) => { - const [action] = await db.select().from(chatActions).where(and(eq(chatActions.id, candidate.id), eq(chatActions.status, "received"))); - if (!action) return; - const issueId = String(action.payload.issueId); - const userId = String(action.payload.userId); - const agentId = String(action.payload.agentId); - 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( - eq(chatPublications.companyId, action.companyId), eq(chatPublications.endpointId, action.endpointId), - eq(chatPublications.conversationId, action.conversationId!), eq(chatPublications.commentId, commentId), - inArray(chatPublications.state, ["pending", "retry"]), - )); - }); - } - const currentEndpoint = await endpointRecord(action.endpointId); - if (currentEndpoint && ["paused", "attention"].includes(currentEndpoint.endpoint.status)) return; - const bindings = await slackBoardReplyBindings(db, { companyId: action.companyId, issueId, userId, agentId, commentIds: [commentId] }); - if (!bindings.some(binding => binding.endpointId === action.endpointId && binding.conversationId === action.conversationId)) { - await cancelWork("The Slack message author no longer has access to this conversation"); - return; - } - try { - const [publication] = await db.select().from(chatPublications).where(and(eq(chatPublications.companyId, action.companyId), eq(chatPublications.endpointId, action.endpointId), eq(chatPublications.commentId, commentId))).limit(1); - const [conversation] = await db.select().from(chatConversations).where(and(eq(chatConversations.companyId, action.companyId), eq(chatConversations.id, action.conversationId!))); - const credentials = currentEndpoint ? await resolveCredentials(currentEndpoint.endpoint) : null; - if (!publication || !conversation || !currentEndpoint || !await authorizeSlackBoardPublication(db, currentEndpoint.endpoint, conversation, publication, credentials?.botToken ?? "", lease.fetch)) { - throw forbidden("The Slack message author no longer has access to this conversation"); - } - // Persist the resume and its marker together before waking. If a - // scheduler response is lost, retry cannot reopen work a second time. - const issue = await lease.commit(async (tx) => { - const [current] = await tx.select().from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, action.companyId), eq(issues.assigneeAgentId, agentId))).for("update"); - if (!current) throw forbidden("The task is no longer assigned to the Slack agent"); - await assertSlackBoardWorkAllowed(tx as unknown as Db, current); - if (action.result?.resumeApplied !== true) { - if (["done", "blocked"].includes(current.status)) { - 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; - }); - await lease.commit(async () => {}); - await options.heartbeat.wakeup(agentId, { - source: "automation", triggerDetail: "system", reason: "issue_commented", - idempotencyKey: `slack-board-comment:${action.id}`, allowRunCoalescing: false, - requestedByActorType: "user", requestedByActorId: userId, - payload: { issueId, commentId, resumeIntent: true, followUpRequested: true }, - contextSnapshot: { issueId, taskId: issueId, taskKey: issue.identifier ?? issueId, wakeCommentId: commentId, source: "issue.comment", resumeIntent: true, followUpRequested: true }, - }); - 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) { - if (error instanceof HttpError && [400, 403, 404, 409].includes(error.status)) { - await cancelWork(error.message); - return; - } - logger.warn({ actionId: action.id, error: error instanceof Error ? error.message : "Wakeup failed" }, "Slack Board message wakeup will retry"); - } - }); - } - } + async function processPendingChatMaintenance(limit = 25) { return Promise.allSettled([ @@ -6591,470 +6518,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { await resolveProviderEffect(endpointId, actionId, resolution, userId); return; } - if (initialRecord.endpoint.provider !== "slack") { - throw conflict("This provider action cannot be resolved here", { - code: "chat_action_resolution_unsupported", - }); - } - if (resolution === "mark_delivered") { - throw conflict("A Slack task start cannot be marked delivered here", { - code: "chat_action_resolution_unsupported", - }); - } - - if (resolution === "cancel") { - await withCredentialMutationLease( - initialRecord.endpoint, - async (credentialLease) => { - await db.transaction(async (tx) => { - await credentialLease.assertOwned(tx); - await notifyChatActionWork(tx, "slash_task_start"); - await tx - .update(chatActions) - .set({ - status: "delivery_unknown", - result: { code: "slash_task_delivery_unknown" }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.endpointId, endpointId), - eq(chatActions.id, actionId), - eq(chatActions.kind, "slash_task_start"), - or( - and( - eq(chatActions.status, "received"), - lte( - chatActions.updatedAt, - new Date(Date.now() - SLACK_COMMAND_POST_STALE_MS), - ), - ), - and( - eq(chatActions.status, "resolving"), - lte( - chatActions.updatedAt, - new Date( - Date.now() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, - ), - ), - ), - ), - ), - ); - await credentialLease.assertOwned(tx); - const action = await tx - .select() - .from(chatActions) - .where( - and( - eq(chatActions.endpointId, endpointId), - eq(chatActions.id, actionId), - eq(chatActions.kind, "slash_task_start"), - ), - ) - .for("update") - .then((rows) => rows[0] ?? null); - if (!action) throw notFound("Provider action not found"); - if (action.status !== "delivery_unknown") { - throw conflict( - "Only an unconfirmed provider action needs operator resolution", - { - code: "chat_action_resolution_not_required", - }, - ); - } - const now = new Date(); - await notifyChatActionWork(tx, "slash_task_start"); - await tx - .update(chatActions) - .set({ - status: "cancelled", - result: { code: "slash_task_cancelled_by_operator" }, - updatedAt: now, - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "delivery_unknown"), - ), - ); - await logActivity(tx as unknown as Db, { - companyId: action.companyId, - actorType: "user", - actorId: userId, - action: "chat.slack_command_cancel", - entityType: "chat_action", - entityId: action.id, - details: { - endpointId, - previousState: "delivery_unknown", - nextState: "cancelled", - duplicateRiskAccepted: false, - }, - }); - await credentialLease.assertOwned(tx); - }); - }, - ); - return; - } - - let confirmedActionId: string | null = null; - await withCredentialMutationLease( - initialRecord.endpoint, - async (credentialLease) => { - const record = await endpointRecord(endpointId); - if (!record) throw notFound("Chat endpoint not found"); - if ( - record.endpoint.provider !== "slack" || - !["verifying", "active"].includes(record.endpoint.status) - ) { - throw conflict( - "The Slack connection must be active before retrying this action", - { - code: "chat_action_endpoint_unavailable", - }, - ); - } - const action = await db - .select() - .from(chatActions) - .where( - and( - eq(chatActions.endpointId, endpointId), - eq(chatActions.id, actionId), - eq(chatActions.kind, "slash_task_start"), - ), - ) - .then((rows) => rows[0] ?? null); - if (!action) throw notFound("Provider action not found"); - const staleActionCanBeNormalized = - (action.status === "received" && - action.updatedAt.getTime() <= - Date.now() - SLACK_COMMAND_POST_STALE_MS) || - (action.status === "resolving" && - action.updatedAt.getTime() <= - Date.now() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS); - if ( - action.status !== "delivery_unknown" && - !staleActionCanBeNormalized - ) { - throw conflict( - "Only an unconfirmed provider action needs operator resolution", - { - code: "chat_action_resolution_not_required", - }, - ); - } - const recovery = slackSlashTaskRecoveryPayload(action.payload); - if (!recovery) { - throw conflict( - "This action does not contain enough context for an explicit retry", - { - code: "chat_action_retry_context_missing", - }, - ); - } - if (!action.principalId) { - throw conflict("The original Slack account is no longer available", { - code: "chat_action_principal_missing", - }); - } - const principal = await db - .select() - .from(chatExternalPrincipals) - .where( - and( - eq(chatExternalPrincipals.id, action.principalId), - eq(chatExternalPrincipals.companyId, record.endpoint.companyId), - eq(chatExternalPrincipals.provider, "slack"), - eq( - chatExternalPrincipals.providerAccountId, - record.endpoint.providerAccountId ?? "unknown", - ), - ), - ) - .then((rows) => rows[0] ?? null); - if (!principal || principal.isBot || principal.kind !== "user") { - throw conflict("The original Slack account is no longer available", { - code: "chat_action_principal_missing", - }); - } - const link = await db - .select({ - paperclipUserId: chatIdentityLinks.paperclipUserId, - status: chatIdentityLinks.status, - }) - .from(chatIdentityLinks) - .where( - and( - eq(chatIdentityLinks.endpointId, endpointId), - eq(chatIdentityLinks.principalId, principal.id), - ), - ) - .then((rows) => rows[0] ?? null); - let principalAllowed = false; - if (link?.status === "linked" && link.paperclipUserId) { - principalAllowed = await db - .select({ membershipRole: companyMemberships.membershipRole }) - .from(companyMemberships) - .where( - and( - eq(companyMemberships.companyId, record.endpoint.companyId), - eq(companyMemberships.principalType, "user"), - eq(companyMemberships.principalId, link.paperclipUserId), - eq(companyMemberships.status, "active"), - ne(companyMemberships.membershipRole, "viewer"), - ), - ) - .then((rows) => rows.length > 0); - } else if (record.endpoint.allowUnlinkedPeople) { - principalAllowed = await sponsorAllowsGuest(record.endpoint); - } - - const rawChannelId = recovery.channelId.replace(/^slack:/, ""); - const isDirectMessage = /^D[A-Z0-9]+$/i.test(rawChannelId); - const resource = await db - .select() - .from(chatEndpointResources) - .where( - and( - eq(chatEndpointResources.endpointId, endpointId), - eq(chatEndpointResources.providerResourceId, rawChannelId), - ), - ) - .then((rows) => rows[0] ?? null); - const destinationAllowed = isDirectMessage - ? record.endpoint.allowDirectMessages - : nonDirectDestinationAllowed(record.endpoint, resource); - if (!principalAllowed || !destinationAllowed) { - throw conflict( - "The original account or destination is no longer allowed", - { - code: "chat_action_no_longer_authorized", - }, - ); - } - - // Resolve and validate the runtime before claiming the action. A local - // setup failure at this point cannot have sent anything to Slack and - // therefore leaves the original ambiguous action untouched. - const endpointRuntime = await runtimeFor(record.endpoint); - const baseThread = endpointRuntime.thread(`slack:${rawChannelId}:`); - const retryAttempt = - (action.status === "delivery_unknown" && - typeof action.result?.attemptCount === "number" && - Number.isSafeInteger(action.result.attemptCount) - ? Math.max(0, action.result.attemptCount) - : 0) + 1; - const [claimed] = await db.transaction(async (tx) => { - await credentialLease.assertOwned(tx); - await notifyChatActionWork(tx, "slash_task_start"); - await tx - .update(chatActions) - .set({ - status: "delivery_unknown", - result: { code: "slash_task_delivery_unknown" }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.endpointId, endpointId), - eq(chatActions.id, actionId), - eq(chatActions.kind, "slash_task_start"), - or( - and( - eq(chatActions.status, "received"), - lte( - chatActions.updatedAt, - new Date(Date.now() - SLACK_COMMAND_POST_STALE_MS), - ), - ), - and( - eq(chatActions.status, "resolving"), - lte( - chatActions.updatedAt, - new Date( - Date.now() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, - ), - ), - ), - ), - ), - ); - await credentialLease.assertOwned(tx); - const current = await tx - .select() - .from(chatActions) - .where( - and( - eq(chatActions.endpointId, endpointId), - eq(chatActions.id, actionId), - eq(chatActions.kind, "slash_task_start"), - ), - ) - .for("update") - .then((rows) => rows[0] ?? null); - if (!current) throw notFound("Provider action not found"); - if (current.status !== "delivery_unknown") { - throw conflict( - "Only an unconfirmed provider action needs operator resolution", - { - code: "chat_action_resolution_not_required", - }, - ); - } - const currentEndpoint = await runtimeCallbackEndpoint( - tx, - record.endpoint.id, - runtimeContextForRecord(record), - ["verifying", "active"], - ); - const currentResource = - currentEndpoint && !isDirectMessage - ? await tx - .select() - .from(chatEndpointResources) - .where( - and( - eq( - chatEndpointResources.companyId, - currentEndpoint.companyId, - ), - eq(chatEndpointResources.endpointId, currentEndpoint.id), - eq( - chatEndpointResources.providerResourceId, - rawChannelId, - ), - ), - ) - .for("update") - .then((rows) => rows[0] ?? null) - : null; - const currentAuthorization = currentEndpoint - ? await lockCurrentPrincipalAuthorization( - tx, - currentEndpoint, - action.principalId!, - ) - : null; - const currentDestinationAllowed = currentEndpoint - ? isDirectMessage - ? currentEndpoint.allowDirectMessages - : nonDirectDestinationAllowed(currentEndpoint, currentResource) - : false; - if ( - !currentEndpoint || - !currentAuthorization?.allowed || - !currentDestinationAllowed - ) { - throw conflict( - "The original account or destination is no longer allowed", - { - code: "chat_action_no_longer_authorized", - }, - ); - } - await notifyChatActionWork(tx, "slash_task_start"); - const rows = await tx - .update(chatActions) - .set({ - status: "resolving", - result: { - code: "slash_task_retry_requested", - attemptCount: retryAttempt, - authorizedUserId: currentAuthorization.userId, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, current.id), - eq(chatActions.status, "delivery_unknown"), - ), - ) - .returning(); - await logActivity(tx as unknown as Db, { - companyId: current.companyId, - actorType: "user", - actorId: userId, - action: "chat.slack_command_retry_anyway", - entityType: "chat_action", - entityId: current.id, - details: { - endpointId, - previousState: "delivery_unknown", - nextState: "resolving", - duplicateRiskAccepted: true, - }, - }); - await credentialLease.assertOwned(tx); - return rows; - }); - if (!claimed) { - throw conflict("This provider action is already being resolved", { - code: "chat_action_resolution_conflict", - }); - } - - let starter: { id: string; threadId: string }; - try { - starter = await baseThread.post("Starting a task…"); - } catch (error) { - await finalizeSlackTaskStartFailure({ - actionId: action.id, - actionStatus: "resolving", - attempt: retryAttempt, - connectionId: record.endpoint.connectionId, - credentialLease, - endpoint: record.endpoint, - error, - providerAttempted: true, - resource, - runtimeFence: runtimeContextForRecord(record), - }); - throw error; - } - const [completed] = await db.transaction(async (tx) => { - await credentialLease.assertOwned(tx); - await notifyChatActionWork(tx, "slash_task_start"); - return tx - .update(chatActions) - .set({ - // The provider send is now confirmed, but task admission is a - // separate durable phase. Persist this fence before constructing the - // Paperclip delivery so a crash cannot turn into another Slack post. - status: "provider_confirmed", - result: { - attemptCount: retryAttempt, - authorizedUserId: claimed.result?.authorizedUserId ?? null, - threadId: slackTaskStarterThreadId(rawChannelId, starter.id), - providerMessageId: starter.id, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.status, "resolving"), - sql`(${chatActions.result}->>'attemptCount')::int = ${retryAttempt}`, - ), - ) - .returning({ id: chatActions.id }); - }); - // A deliberately cancelled action wins over a provider response that - // arrived only after the operator's stale-action window. Keep the Slack - // starter as an orphan rather than creating work the operator cancelled. - if (!completed) return; - confirmedActionId = action.id; - }, - ); - if (typeof confirmedActionId === "string") { - await admitConfirmedSlackTaskStart( - confirmedActionId, - options.deferWebhookProcessing === true, - ); - } + return resolveSlackTaskStart(endpointId, actionId, resolution, userId, initialRecord); } async function getPublicationBatchStatus( diff --git a/server/src/services/chat-channels/slack/board-messages.ts b/server/src/services/chat-channels/slack/board-messages.ts new file mode 100644 index 0000000000..4d1bdef2fc --- /dev/null +++ b/server/src/services/chat-channels/slack/board-messages.ts @@ -0,0 +1,104 @@ +import { dueChatAction } from "../../chat-action-work.js"; +import { notifyChatPublicationWork, notifyChatActionWork } from "../../chat-work-notifications.js"; +import { withSlackBoardLease } from "../../slack-board-lease.js"; +import { slackBoardReplyBindings } from "../../slack-board-messages.js"; +import { authorizeSlackBoardPublication } from "../../slack-board-authority.js"; +import { assertSlackBoardWorkAllowed } from "../../slack-board-resume.js"; +import { and, asc, eq, inArray } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { chatActions, chatConversations, chatEndpoints, chatPublications, issues } from "@paperclipai/db"; +import { HttpError, forbidden } from "../../../errors.js"; +import { logger } from "../../../middleware/logger.js"; +import { logActivity } from "../../activity-log.js"; +import { issueService } from "../../issues.js"; +import type { EndpointRecord } from "../runtime-types.js"; +import type { createChatCredentials } from "../credentials.js"; +import type { IssueAssignmentWakeupDeps } from "../../issue-assignment-wakeup.js"; +type BoardMessageOptions = { + endpointRecord: (endpointId: string) => Promise; + resolveCredentials: ReturnType["resolveCredentials"]; + fetchImpl: typeof globalThis.fetch; + readonly heartbeat: IssueAssignmentWakeupDeps; +}; +export function createSlackBoardMessages(db: Db, options: BoardMessageOptions) { + const { endpointRecord, resolveCredentials, fetchImpl } = options; + 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(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) => { + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.id, candidate.id), eq(chatActions.status, "received"))); + if (!action) return; + const issueId = String(action.payload.issueId); + const userId = String(action.payload.userId); + const agentId = String(action.payload.agentId); + 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( + eq(chatPublications.companyId, action.companyId), eq(chatPublications.endpointId, action.endpointId), + eq(chatPublications.conversationId, action.conversationId!), eq(chatPublications.commentId, commentId), + inArray(chatPublications.state, ["pending", "retry"]), + )); + }); + } + const currentEndpoint = await endpointRecord(action.endpointId); + if (currentEndpoint && ["paused", "attention"].includes(currentEndpoint.endpoint.status)) return; + const bindings = await slackBoardReplyBindings(db, { companyId: action.companyId, issueId, userId, agentId, commentIds: [commentId] }); + if (!bindings.some(binding => binding.endpointId === action.endpointId && binding.conversationId === action.conversationId)) { + await cancelWork("The Slack message author no longer has access to this conversation"); + return; + } + try { + const [publication] = await db.select().from(chatPublications).where(and(eq(chatPublications.companyId, action.companyId), eq(chatPublications.endpointId, action.endpointId), eq(chatPublications.commentId, commentId))).limit(1); + const [conversation] = await db.select().from(chatConversations).where(and(eq(chatConversations.companyId, action.companyId), eq(chatConversations.id, action.conversationId!))); + const credentials = currentEndpoint ? await resolveCredentials(currentEndpoint.endpoint) : null; + if (!publication || !conversation || !currentEndpoint || !await authorizeSlackBoardPublication(db, currentEndpoint.endpoint, conversation, publication, credentials?.botToken ?? "", lease.fetch)) { + throw forbidden("The Slack message author no longer has access to this conversation"); + } + // Persist the resume and its marker together before waking. If a + // scheduler response is lost, retry cannot reopen work a second time. + const issue = await lease.commit(async (tx) => { + const [current] = await tx.select().from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, action.companyId), eq(issues.assigneeAgentId, agentId))).for("update"); + if (!current) throw forbidden("The task is no longer assigned to the Slack agent"); + await assertSlackBoardWorkAllowed(tx as unknown as Db, current); + if (action.result?.resumeApplied !== true) { + if (["done", "blocked"].includes(current.status)) { + 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; + }); + await lease.commit(async () => {}); + await options.heartbeat.wakeup(agentId, { + source: "automation", triggerDetail: "system", reason: "issue_commented", + idempotencyKey: `slack-board-comment:${action.id}`, allowRunCoalescing: false, + requestedByActorType: "user", requestedByActorId: userId, + payload: { issueId, commentId, resumeIntent: true, followUpRequested: true }, + contextSnapshot: { issueId, taskId: issueId, taskKey: issue.identifier ?? issueId, wakeCommentId: commentId, source: "issue.comment", resumeIntent: true, followUpRequested: true }, + }); + 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) { + if (error instanceof HttpError && [400, 403, 404, 409].includes(error.status)) { + await cancelWork(error.message); + return; + } + logger.warn({ actionId: action.id, error: error instanceof Error ? error.message : "Wakeup failed" }, "Slack Board message wakeup will retry"); + } + }); + } + } + return { processPendingSlackBoardMessages }; +} diff --git a/server/src/services/chat-channels/slack/task-start-resolution.ts b/server/src/services/chat-channels/slack/task-start-resolution.ts new file mode 100644 index 0000000000..959d5b0ea1 --- /dev/null +++ b/server/src/services/chat-channels/slack/task-start-resolution.ts @@ -0,0 +1,488 @@ +import type { EndpointRecord } from "../runtime-types.js"; +import { slackSlashTaskRecoveryPayload, slackTaskStarterThreadId } from "./task-start.js"; +import { SLACK_COMMAND_POST_STALE_MS, SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS } from "../../chat-action-work.js"; +import { notifyChatActionWork } from "../../chat-work-notifications.js"; +import { and, eq, lte, ne, or, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { chatActions, chatEndpointResources, chatExternalPrincipals, chatIdentityLinks, companyMemberships } from "@paperclipai/db"; +import { conflict, notFound } from "../../../errors.js"; +import { logActivity } from "../../activity-log.js"; +import type { createSlackTaskStart } from "./task-start.js"; +type ResolutionOptions = Pick[1], "endpointRecord" | "withCredentialMutationLease" | "runtimeFor" | "runtimeContextForRecord" | "runtimeCallbackEndpoint" | "lockCurrentPrincipalAuthorization" | "sponsorAllowsGuest" | "nonDirectDestinationAllowed"> & Pick, "finalizeSlackTaskStartFailure" | "admitConfirmedSlackTaskStart"> & { readonly deferWebhookProcessing: boolean | undefined }; + +export function createSlackTaskStartResolution(db: Db, options: ResolutionOptions) { + const { endpointRecord, withCredentialMutationLease, runtimeFor, runtimeContextForRecord, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, sponsorAllowsGuest, nonDirectDestinationAllowed, finalizeSlackTaskStartFailure, admitConfirmedSlackTaskStart } = options; + async function resolveSlackTaskStart( + endpointId: string, + actionId: string, + resolution: "mark_delivered" | "retry_anyway" | "cancel", + userId: string, + initialRecord: EndpointRecord, + ) { + if (initialRecord.endpoint.provider !== "slack") { + throw conflict("This provider action cannot be resolved here", { + code: "chat_action_resolution_unsupported", + }); + } + if (resolution === "mark_delivered") { + throw conflict("A Slack task start cannot be marked delivered here", { + code: "chat_action_resolution_unsupported", + }); + } + + if (resolution === "cancel") { + await withCredentialMutationLease( + initialRecord.endpoint, + async (credentialLease) => { + await db.transaction(async (tx) => { + await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); + await tx + .update(chatActions) + .set({ + status: "delivery_unknown", + result: { code: "slash_task_delivery_unknown" }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.endpointId, endpointId), + eq(chatActions.id, actionId), + eq(chatActions.kind, "slash_task_start"), + or( + and( + eq(chatActions.status, "received"), + lte( + chatActions.updatedAt, + new Date(Date.now() - SLACK_COMMAND_POST_STALE_MS), + ), + ), + and( + eq(chatActions.status, "resolving"), + lte( + chatActions.updatedAt, + new Date( + Date.now() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, + ), + ), + ), + ), + ), + ); + await credentialLease.assertOwned(tx); + const action = await tx + .select() + .from(chatActions) + .where( + and( + eq(chatActions.endpointId, endpointId), + eq(chatActions.id, actionId), + eq(chatActions.kind, "slash_task_start"), + ), + ) + .for("update") + .then((rows) => rows[0] ?? null); + if (!action) throw notFound("Provider action not found"); + if (action.status !== "delivery_unknown") { + throw conflict( + "Only an unconfirmed provider action needs operator resolution", + { + code: "chat_action_resolution_not_required", + }, + ); + } + const now = new Date(); + await notifyChatActionWork(tx, "slash_task_start"); + await tx + .update(chatActions) + .set({ + status: "cancelled", + result: { code: "slash_task_cancelled_by_operator" }, + updatedAt: now, + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "delivery_unknown"), + ), + ); + await logActivity(tx as unknown as Db, { + companyId: action.companyId, + actorType: "user", + actorId: userId, + action: "chat.slack_command_cancel", + entityType: "chat_action", + entityId: action.id, + details: { + endpointId, + previousState: "delivery_unknown", + nextState: "cancelled", + duplicateRiskAccepted: false, + }, + }); + await credentialLease.assertOwned(tx); + }); + }, + ); + return; + } + + let confirmedActionId: string | null = null; + await withCredentialMutationLease( + initialRecord.endpoint, + async (credentialLease) => { + const record = await endpointRecord(endpointId); + if (!record) throw notFound("Chat endpoint not found"); + if ( + record.endpoint.provider !== "slack" || + !["verifying", "active"].includes(record.endpoint.status) + ) { + throw conflict( + "The Slack connection must be active before retrying this action", + { + code: "chat_action_endpoint_unavailable", + }, + ); + } + const action = await db + .select() + .from(chatActions) + .where( + and( + eq(chatActions.endpointId, endpointId), + eq(chatActions.id, actionId), + eq(chatActions.kind, "slash_task_start"), + ), + ) + .then((rows) => rows[0] ?? null); + if (!action) throw notFound("Provider action not found"); + const staleActionCanBeNormalized = + (action.status === "received" && + action.updatedAt.getTime() <= + Date.now() - SLACK_COMMAND_POST_STALE_MS) || + (action.status === "resolving" && + action.updatedAt.getTime() <= + Date.now() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS); + if ( + action.status !== "delivery_unknown" && + !staleActionCanBeNormalized + ) { + throw conflict( + "Only an unconfirmed provider action needs operator resolution", + { + code: "chat_action_resolution_not_required", + }, + ); + } + const recovery = slackSlashTaskRecoveryPayload(action.payload); + if (!recovery) { + throw conflict( + "This action does not contain enough context for an explicit retry", + { + code: "chat_action_retry_context_missing", + }, + ); + } + if (!action.principalId) { + throw conflict("The original Slack account is no longer available", { + code: "chat_action_principal_missing", + }); + } + const principal = await db + .select() + .from(chatExternalPrincipals) + .where( + and( + eq(chatExternalPrincipals.id, action.principalId), + eq(chatExternalPrincipals.companyId, record.endpoint.companyId), + eq(chatExternalPrincipals.provider, "slack"), + eq( + chatExternalPrincipals.providerAccountId, + record.endpoint.providerAccountId ?? "unknown", + ), + ), + ) + .then((rows) => rows[0] ?? null); + if (!principal || principal.isBot || principal.kind !== "user") { + throw conflict("The original Slack account is no longer available", { + code: "chat_action_principal_missing", + }); + } + const link = await db + .select({ + paperclipUserId: chatIdentityLinks.paperclipUserId, + status: chatIdentityLinks.status, + }) + .from(chatIdentityLinks) + .where( + and( + eq(chatIdentityLinks.endpointId, endpointId), + eq(chatIdentityLinks.principalId, principal.id), + ), + ) + .then((rows) => rows[0] ?? null); + let principalAllowed = false; + if (link?.status === "linked" && link.paperclipUserId) { + principalAllowed = await db + .select({ membershipRole: companyMemberships.membershipRole }) + .from(companyMemberships) + .where( + and( + eq(companyMemberships.companyId, record.endpoint.companyId), + eq(companyMemberships.principalType, "user"), + eq(companyMemberships.principalId, link.paperclipUserId), + eq(companyMemberships.status, "active"), + ne(companyMemberships.membershipRole, "viewer"), + ), + ) + .then((rows) => rows.length > 0); + } else if (record.endpoint.allowUnlinkedPeople) { + principalAllowed = await sponsorAllowsGuest(record.endpoint); + } + + const rawChannelId = recovery.channelId.replace(/^slack:/, ""); + const isDirectMessage = /^D[A-Z0-9]+$/i.test(rawChannelId); + const resource = await db + .select() + .from(chatEndpointResources) + .where( + and( + eq(chatEndpointResources.endpointId, endpointId), + eq(chatEndpointResources.providerResourceId, rawChannelId), + ), + ) + .then((rows) => rows[0] ?? null); + const destinationAllowed = isDirectMessage + ? record.endpoint.allowDirectMessages + : nonDirectDestinationAllowed(record.endpoint, resource); + if (!principalAllowed || !destinationAllowed) { + throw conflict( + "The original account or destination is no longer allowed", + { + code: "chat_action_no_longer_authorized", + }, + ); + } + + // Resolve and validate the runtime before claiming the action. A local + // setup failure at this point cannot have sent anything to Slack and + // therefore leaves the original ambiguous action untouched. + const endpointRuntime = await runtimeFor(record.endpoint); + const baseThread = endpointRuntime.thread(`slack:${rawChannelId}:`); + const retryAttempt = + (action.status === "delivery_unknown" && + typeof action.result?.attemptCount === "number" && + Number.isSafeInteger(action.result.attemptCount) + ? Math.max(0, action.result.attemptCount) + : 0) + 1; + const [claimed] = await db.transaction(async (tx) => { + await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); + await tx + .update(chatActions) + .set({ + status: "delivery_unknown", + result: { code: "slash_task_delivery_unknown" }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.endpointId, endpointId), + eq(chatActions.id, actionId), + eq(chatActions.kind, "slash_task_start"), + or( + and( + eq(chatActions.status, "received"), + lte( + chatActions.updatedAt, + new Date(Date.now() - SLACK_COMMAND_POST_STALE_MS), + ), + ), + and( + eq(chatActions.status, "resolving"), + lte( + chatActions.updatedAt, + new Date( + Date.now() - SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, + ), + ), + ), + ), + ), + ); + await credentialLease.assertOwned(tx); + const current = await tx + .select() + .from(chatActions) + .where( + and( + eq(chatActions.endpointId, endpointId), + eq(chatActions.id, actionId), + eq(chatActions.kind, "slash_task_start"), + ), + ) + .for("update") + .then((rows) => rows[0] ?? null); + if (!current) throw notFound("Provider action not found"); + if (current.status !== "delivery_unknown") { + throw conflict( + "Only an unconfirmed provider action needs operator resolution", + { + code: "chat_action_resolution_not_required", + }, + ); + } + const currentEndpoint = await runtimeCallbackEndpoint( + tx, + record.endpoint.id, + runtimeContextForRecord(record), + ["verifying", "active"], + ); + const currentResource = + currentEndpoint && !isDirectMessage + ? await tx + .select() + .from(chatEndpointResources) + .where( + and( + eq( + chatEndpointResources.companyId, + currentEndpoint.companyId, + ), + eq(chatEndpointResources.endpointId, currentEndpoint.id), + eq( + chatEndpointResources.providerResourceId, + rawChannelId, + ), + ), + ) + .for("update") + .then((rows) => rows[0] ?? null) + : null; + const currentAuthorization = currentEndpoint + ? await lockCurrentPrincipalAuthorization( + tx, + currentEndpoint, + action.principalId!, + ) + : null; + const currentDestinationAllowed = currentEndpoint + ? isDirectMessage + ? currentEndpoint.allowDirectMessages + : nonDirectDestinationAllowed(currentEndpoint, currentResource) + : false; + if ( + !currentEndpoint || + !currentAuthorization?.allowed || + !currentDestinationAllowed + ) { + throw conflict( + "The original account or destination is no longer allowed", + { + code: "chat_action_no_longer_authorized", + }, + ); + } + await notifyChatActionWork(tx, "slash_task_start"); + const rows = await tx + .update(chatActions) + .set({ + status: "resolving", + result: { + code: "slash_task_retry_requested", + attemptCount: retryAttempt, + authorizedUserId: currentAuthorization.userId, + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, current.id), + eq(chatActions.status, "delivery_unknown"), + ), + ) + .returning(); + await logActivity(tx as unknown as Db, { + companyId: current.companyId, + actorType: "user", + actorId: userId, + action: "chat.slack_command_retry_anyway", + entityType: "chat_action", + entityId: current.id, + details: { + endpointId, + previousState: "delivery_unknown", + nextState: "resolving", + duplicateRiskAccepted: true, + }, + }); + await credentialLease.assertOwned(tx); + return rows; + }); + if (!claimed) { + throw conflict("This provider action is already being resolved", { + code: "chat_action_resolution_conflict", + }); + } + + let starter: { id: string; threadId: string }; + try { + starter = await baseThread.post("Starting a task…"); + } catch (error) { + await finalizeSlackTaskStartFailure({ + actionId: action.id, + actionStatus: "resolving", + attempt: retryAttempt, + connectionId: record.endpoint.connectionId, + credentialLease, + endpoint: record.endpoint, + error, + providerAttempted: true, + resource, + runtimeFence: runtimeContextForRecord(record), + }); + throw error; + } + const [completed] = await db.transaction(async (tx) => { + await credentialLease.assertOwned(tx); + await notifyChatActionWork(tx, "slash_task_start"); + return tx + .update(chatActions) + .set({ + // The provider send is now confirmed, but task admission is a + // separate durable phase. Persist this fence before constructing the + // Paperclip delivery so a crash cannot turn into another Slack post. + status: "provider_confirmed", + result: { + attemptCount: retryAttempt, + authorizedUserId: claimed.result?.authorizedUserId ?? null, + threadId: slackTaskStarterThreadId(rawChannelId, starter.id), + providerMessageId: starter.id, + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.status, "resolving"), + sql`(${chatActions.result}->>'attemptCount')::int = ${retryAttempt}`, + ), + ) + .returning({ id: chatActions.id }); + }); + // A deliberately cancelled action wins over a provider response that + // arrived only after the operator's stale-action window. Keep the Slack + // starter as an orphan rather than creating work the operator cancelled. + if (!completed) return; + confirmedActionId = action.id; + }, + ); + if (typeof confirmedActionId === "string") { + await admitConfirmedSlackTaskStart( + confirmedActionId, + options.deferWebhookProcessing === true, + ); + } + } + return { resolveSlackTaskStart }; +}