Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-10-14 17:40:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(server): move Slack recovery into channel modules

Browse Source 
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-13 09:23:21 -05:00
1 parent 3f7b00a8d0
commit 1c1bf5d019
3 files changed
+603 -547
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/services/chat-channels.ts
+11 -547
View File
@@ -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(
server/src/services/chat-channels/slack/board-messages.ts
+104
View File
@@ -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<EndpointRecord | null>;
resolveCredentials: ReturnType<typeof createChatCredentials>["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 };
}
server/src/services/chat-channels/slack/task-start-resolution.ts
+488
View File
@@ -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<Parameters<typeof createSlackTaskStart>[1], "endpointRecord" | "withCredentialMutationLease" | "runtimeFor" | "runtimeContextForRecord" | "runtimeCallbackEndpoint" | "lockCurrentPrincipalAuthorization" | "sponsorAllowsGuest" | "nonDirectDestinationAllowed"> & Pick<ReturnType<typeof createSlackTaskStart>, "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 };
}