diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index fc554723f06..68a53b082f1 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,6 +1,8 @@ +import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; +import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js"; import { createGitHubMaintenance } from "./chat-channels/github/maintenance.js"; -import { createSpekoChannelService, authorizeSpekoPublicVoiceTask, replaceSpekoChannelRuntime } from "./chat-channels/speko/runtime.js"; +import { createSpekoChannelService, authorizeSpekoPublicVoiceTask, replaceSpekoChannelRuntime, SPEKO_CONVERSATION_IS_LINEAR } from "./chat-channels/speko/runtime.js"; import { prepareDiscordRuntimeCommand } from "./chat-channels/discord/runtime-commands.js"; import { createSlackSetupServices } from "./chat-channels/slack/setup-services.js"; import { isSupportedTelegramWebhookBaseUrl } from "./chat-channels/telegram/setup.js"; @@ -9,7 +11,7 @@ import { githubInboundTaskTitle, githubInboundCommentText, githubAssignedAgentRe import { applyChatTaskReviewContainment } from "./chat-channels/inbound/task-trust.js"; import { photonConversationBindingIsComplete, photonControlTargetsConversation, photonReplyMetadata, recordPhotonInboundActivity } from "./chat-channels/imessage-photon/messages.js"; import { deferTeamsSetupReply } from "./chat-channels/microsoft-teams/setup-replies.js"; -import { slackOrphanFollowUp, stripSlackTitleMentions } from "./chat-channels/slack/messages.js"; +import { slackOrphanFollowUp, stripSlackTitleMentions, slackMessageMeetsMentionRequirement } from "./chat-channels/slack/messages.js"; import { createChatPublicationResolution } from "./chat-channels/publications/resolution.js"; import { createChatPublicationRequests } from "./chat-channels/publications/requests.js"; import { createTeamsBoardSendPreparation } from "./chat-channels/microsoft-teams/board-send-preparation.js"; @@ -17,7 +19,7 @@ import { createInboundWakeAuthority, inboundWakeScope } from "./chat-channels/in import { createInboundWakeups } from "./chat-channels/inbound/wakeups.js"; import { createInboundWakeNotices } from "./chat-channels/inbound/wakeup-notices.js"; import { createDiscordRuntimeReconciliation } from "./chat-channels/discord/runtime-reconciliation.js"; -import { createPhotonRuntimeCallbacks } from "./chat-channels/imessage-photon/runtime-callbacks.js"; +import { createPhotonRuntimeCallbacks, discardPhotonInboundMessages, photonRuntimeStatusMayAdvance, PHOTON_MESSAGE_INGRESS_ONLY } from "./chat-channels/imessage-photon/runtime-callbacks.js"; import { createChatReceiverOwnership } from "./chat-channels/runtime/ownership.js"; import { createGitHubWebhookScope, githubWebhookCanRecoverRevokedInstallation } from "./chat-channels/github/webhook-scope.js"; import { createGitHubWebhookAuthentication, retryableGitHubWebhookResponse } from "./chat-channels/github/webhook-authentication.js"; @@ -110,7 +112,7 @@ import { import { telegramLifecycleEventFromPayload, telegramIgnoresInboundMessage } from "./chat-channels/telegram/messages.js"; import { - teamsWebhookLifecycleEventFromPayload, teamsOrphanFollowUp, teamsAttachmentReferenceText, + teamsWebhookLifecycleEventFromPayload, teamsRootMentionCanEnableSetupDestination, teamsOrphanFollowUp, teamsAttachmentReferenceText, } from "./chat-channels/microsoft-teams/messages.js"; import { githubSupplementalLifecycleFromRequest, @@ -562,6 +564,12 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const discordGatewayStopTasks = new Set>(); let shuttingDown = false; const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } }); + const { receiveSdkMessages } = createChatSdkMessageReceiver({ + endpointRecord, processMessage, + ingressOnly: (provider) => provider === "imessage-photon" + ? PHOTON_MESSAGE_INGRESS_ONLY + : options.deferWebhookProcessing === true, + }); function localRuntimeEpoch(endpointId: string): number { return runtimeLocalEpochs.get(endpointId) ?? 0; @@ -582,9 +590,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } async function invalidateRuntime(endpointId: string): Promise { - for (const [deliveryId, live] of liveInboundMessages) { - if (live.runtimeContext?.endpointRuntime === runtime.get(endpointId) && live.thread.id.startsWith("imessage-photon:")) liveInboundMessages.delete(deliveryId); - } + discardPhotonInboundMessages(liveInboundMessages, () => runtime.get(endpointId)); runtimeLocalEpochs.set(endpointId, localRuntimeEpoch(endpointId) + 1); runtimeVersions.delete(endpointId); const discordOwnership = discordGatewayOwnerships.get(endpointId); @@ -1065,7 +1071,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const stillRegistered = runtime.get(endpoint.id) === instance; const stillCurrent = latest !== null && - (latest.endpoint.status === record.endpoint.status || record.endpoint.provider === "imessage-photon" && record.endpoint.status === "attention" && ["active", "verifying"].includes(latest.endpoint.status)) && + (latest.endpoint.status === record.endpoint.status || record.endpoint.provider === "imessage-photon" && photonRuntimeStatusMayAdvance(record.endpoint, latest.endpoint)) && runtimeContextForRecord(latest).version === context.version; if (!stillRegistered || !stillCurrent) { if (stillRegistered) { @@ -1684,7 +1690,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { let { conversation: existingConversation, issue: existingIssue, latestConversation, isLinear } = await resolveInboundConversationBinding(db, { endpoint, thread, - isLinear: () => endpoint.provider !== "speko" && surfaceKind !== "native_thread", + isLinear: () => endpoint.provider === "speko" ? SPEKO_CONVERSATION_IS_LINEAR : surfaceKind !== "native_thread", lookupConversation: endpoint.provider === "microsoft-teams" ? () => conversationForThread(endpoint.id, thread.id) : undefined, @@ -1735,7 +1741,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { (endpoint.provider === "github" || endpoint.provider === "discord" || (endpoint.provider === "microsoft-teams" && - resource?.type === "channel")) && + teamsRootMentionCanEnableSetupDestination(resource))) && !thread.isDM && endpoint.status === "verifying" && enabledResourceCount === 0 && @@ -2364,10 +2370,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { : nonDirectDestinationAllowed(currentEndpoint, currentResource); const mentionStillAllowed = currentEndpoint.provider !== "slack" || - !currentEndpoint.requireAtMention || - thread.isDM || - trigger === "mention" || - message.isMention === true; + slackMessageMeetsMentionRequirement(currentEndpoint, thread, trigger, message); if ( !endpointStillAllowed || !mentionStillAllowed || @@ -2578,40 +2581,15 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { slackSlashControl?: boolean; } = {}, ) { - if ( - event.provider === "discord" && - (!runtimeContext || - !(await ensureDiscordGatewayRuntimeIsCurrent( - event.endpointId, - runtimeContext, - ))) - ) { - return; - } - if (event.provider === "discord") { - await options.discordGatewayMessageAdmissionBarrier?.(); - } - const record = await endpointRecord(event.endpointId); - if (!record) throw notFound("Chat endpoint not found"); - const messages = [...(event.context?.skipped ?? []), event.message]; - for (const message of messages) - await processMessage( - record.endpoint, - event.thread, - message, - message === event.message ? event.trigger : "subscribed_message", - event.provider === "imessage-photon" || - options.deferWebhookProcessing === true, - null, - runtimeContext, - event.providerUpdateId, - null, - messageOptions.receiptReactionSupported !== false, - undefined, - false, - false, - messageOptions.slackSlashControl === true, - ); + if (event.provider === "discord" && !(await admitDiscordSdkMessage( + event.endpointId, + runtimeContext, + { + ensureRuntimeIsCurrent: ensureDiscordGatewayRuntimeIsCurrent, + messageAdmissionBarrier: () => options.discordGatewayMessageAdmissionBarrier?.(), + }, + ))) return; + return receiveSdkMessages(event, runtimeContext, messageOptions); } async function handleMessageUpdated(...args: Parameters["handleMessageUpdated"]>) { return messageLifecycle.handleMessageUpdated(...args); diff --git a/server/src/services/chat-channels/discord/messages.ts b/server/src/services/chat-channels/discord/messages.ts index 61582abed26..fadfadb5274 100644 --- a/server/src/services/chat-channels/discord/messages.ts +++ b/server/src/services/chat-channels/discord/messages.ts @@ -1,3 +1,4 @@ +import type { RuntimeContext } from "../runtime-types.js"; import type { FileUpload } from "chat"; import { createHash } from "node:crypto"; @@ -37,3 +38,16 @@ export function discordMarkdownAttachment(text: string): FileUpload { mimeType: "text/markdown; charset=utf-8", }; } + +export async function admitDiscordSdkMessage( + endpointId: string, + runtimeContext: RuntimeContext | undefined, + options: { + ensureRuntimeIsCurrent: (endpointId: string, context: RuntimeContext) => Promise; + messageAdmissionBarrier: () => Promise | void; + }, +): Promise { + if (!runtimeContext || !(await options.ensureRuntimeIsCurrent(endpointId, runtimeContext))) return false; + await options.messageAdmissionBarrier(); + return true; +} diff --git a/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts b/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts index 7e1f907afc6..7a3fc652157 100644 --- a/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts +++ b/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts @@ -6,7 +6,7 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import { chatConversations, chatEndpoints, heartbeatRuns, toolConnections } from "@paperclipai/db"; import type { ChatSdkRuntimeCallbacks } from "../../chat-sdk-runtime.js"; import type { Db } from "@paperclipai/db"; -import type { RuntimeContext, EndpointRow, EndpointRecord } from "../runtime-types.js"; +import type { RuntimeContext, EndpointRow, EndpointRecord, LiveInboundMessage } from "../runtime-types.js"; import type { createChatReceiverOwnership } from "../runtime/ownership.js"; import type { createChatMessageLifecycle } from "../inbound/lifecycle.js"; type PhotonRuntimeOptions = { @@ -42,4 +42,22 @@ export function createPhotonRuntimeCallbacks(db: Db, options: PhotonRuntimeOptio await writePhotonCheckpoint(new PhotonState({ companyId: companyId, endpointId: endpointId }, createChatSdkStatePersistence(tx as unknown as Db)), lineId, sequence); }) }; -} \ No newline at end of file +} + +export const PHOTON_MESSAGE_INGRESS_ONLY = true; + +export function discardPhotonInboundMessages( + liveInboundMessages: Map, + getCurrentRuntime: () => RuntimeContext["endpointRuntime"] | null, +): void { + for (const [deliveryId, live] of liveInboundMessages) { + if (live.runtimeContext?.endpointRuntime === getCurrentRuntime() && live.thread.id.startsWith("imessage-photon:")) liveInboundMessages.delete(deliveryId); + } +} + +export function photonRuntimeStatusMayAdvance( + initial: Pick, + latest: Pick, +): boolean { + return initial.status === "attention" && ["active", "verifying"].includes(latest.status); +} diff --git a/server/src/services/chat-channels/microsoft-teams/messages.ts b/server/src/services/chat-channels/microsoft-teams/messages.ts index b7bb96de2a5..0472ff228b4 100644 --- a/server/src/services/chat-channels/microsoft-teams/messages.ts +++ b/server/src/services/chat-channels/microsoft-teams/messages.ts @@ -115,3 +115,9 @@ export function teamsWebhookLifecycleEventFromPayload( if (!payload || !endpointRuntime.acceptsProviderScope(payload)) return null; return microsoftTeamsLifecycleEventFromPayload(payload, endpointRuntime); } + +export function teamsRootMentionCanEnableSetupDestination( + resource: { type: string } | null | undefined, +): boolean { + return resource?.type === "channel"; +} diff --git a/server/src/services/chat-channels/runtime/sdk-messages.ts b/server/src/services/chat-channels/runtime/sdk-messages.ts new file mode 100644 index 00000000000..2d85556b11e --- /dev/null +++ b/server/src/services/chat-channels/runtime/sdk-messages.ts @@ -0,0 +1,60 @@ +import type { ChatSdkMessageCallbackEvent } from "../../chat-sdk-runtime.js"; +import type { EndpointRecord, EndpointRow, InboundRuntimeContext, RuntimeContext } from "../runtime-types.js"; +import { notFound } from "../../../errors.js"; + +type SdkMessageOptions = { + receiptReactionSupported?: boolean; + slackSlashControl?: boolean; +}; + +type SdkMessageReceiverOptions = { + endpointRecord: (endpointId: string) => Promise | null>; + ingressOnly: (provider: ChatSdkMessageCallbackEvent["provider"]) => boolean; + processMessage: (...args: [ + endpoint: EndpointRow, + thread: ChatSdkMessageCallbackEvent["thread"], + message: ChatSdkMessageCallbackEvent["message"], + trigger: ChatSdkMessageCallbackEvent["trigger"], + ingressOnly: boolean, + recoveredProviderUrl: string | null, + runtimeContext: InboundRuntimeContext | undefined, + providerUpdateId: number | undefined, + admittedDeliveryId: string | null, + receiptReactionSupported: boolean, + preauthorizedUserId?: string | null, + deferDrainUntilFollowup?: boolean, + suppressSetupDestinationActivation?: boolean, + slackSlashControl?: boolean, + ]) => Promise; +}; + +export function createChatSdkMessageReceiver(options: SdkMessageReceiverOptions) { + const { endpointRecord, ingressOnly, processMessage } = options; + async function receiveSdkMessages( + event: ChatSdkMessageCallbackEvent, + runtimeContext?: RuntimeContext, + messageOptions: SdkMessageOptions = {}, + ) { + const record = await endpointRecord(event.endpointId); + if (!record) throw notFound("Chat endpoint not found"); + const messages = [...(event.context?.skipped ?? []), event.message]; + for (const message of messages) + await processMessage( + record.endpoint, + event.thread, + message, + message === event.message ? event.trigger : "subscribed_message", + ingressOnly(event.provider), + null, + runtimeContext, + event.providerUpdateId, + null, + messageOptions.receiptReactionSupported !== false, + undefined, + false, + false, + messageOptions.slackSlashControl === true, + ); + } + return { receiveSdkMessages }; +} diff --git a/server/src/services/chat-channels/slack/messages.ts b/server/src/services/chat-channels/slack/messages.ts index 5536c335055..e492cf1863f 100644 --- a/server/src/services/chat-channels/slack/messages.ts +++ b/server/src/services/chat-channels/slack/messages.ts @@ -1,3 +1,6 @@ +import type { Message, Thread } from "chat"; +import type { EndpointRow } from "../runtime-types.js"; +import type { ChatSdkMessageCallbackEvent } from "../../chat-sdk-runtime.js"; import { createHash } from "node:crypto"; @@ -96,3 +99,12 @@ export function slackOrphanFollowUp(threadId: string, messageId: string): boolea export function stripSlackTitleMentions(text: string): string { return text.replace(/<@[A-Z0-9]+>/gi, ""); } + +export function slackMessageMeetsMentionRequirement( + endpoint: Pick, + thread: Pick, + trigger: ChatSdkMessageCallbackEvent["trigger"], + message: Pick, +): boolean { + return !endpoint.requireAtMention || thread.isDM || trigger === "mention" || message.isMention === true; +} diff --git a/server/src/services/chat-channels/speko/runtime.ts b/server/src/services/chat-channels/speko/runtime.ts index 1fa6113a32e..98fa1f4c629 100644 --- a/server/src/services/chat-channels/speko/runtime.ts +++ b/server/src/services/chat-channels/speko/runtime.ts @@ -43,3 +43,5 @@ export function replaceSpekoChannelRuntime(runtime: ChatRuntime | ChatSdkRuntime if (!("replaceVoiceEndpoint" in runtime)) throw conflict("Voice runtime is unavailable"); return runtime.replaceVoiceEndpoint({ companyId: record.endpoint.companyId, endpointId: record.endpoint.id, userName: record.assignedAgentName, persistence }); } + +export const SPEKO_CONVERSATION_IS_LINEAR = false;