From 09a2f0b7b0feb1953b8a81dba9315ecd71d00206 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:50:34 -0500 Subject: [PATCH] refactor(server): isolate provider message admission and runtime policies (#16289) ## Thinking Path > - Paperclip helps people manage AI agents for work. > - Chat channels turn external messages into shared task workflows. > - The main service still contains some provider decisions. > - These decisions belong with the existing channel modules. > - This pull request isolates provider message-admission and runtime policies. > - Shared authorization, persistence, and runtime ownership keep their current order. ## Linked Issues or Issue Description Refs: #16235. Stack part 3/3. Base: `master`. #16276 and #16283 are merged. **What existing behavior does this improve?** Provider ownership and readability in the chat channel service. **Current behavior** The entry point combines shared workflows with provider decisions. **Proposed behavior** Provider modules own their decisions. Shared modules own product state and authorization. **Reason and benefit** Maintainers can find provider behavior in its channel folder. **Breaking changes** None intended. Public service methods and durable data shapes stay compatible. ## What Changed - Move Discord SDK admission and its barrier to the Discord message module. - Move Photon live-message eviction and startup recovery policy to its runtime callbacks module. - Move Slack mention eligibility and Teams setup-channel eligibility to their message modules. - Keep Photon deferred ingress and Speko conversation policy with their provider modules. - Keep ordered SDK message batching in a shared runtime module with a live per-message ingress callback. - Keep receiver leases, cache lifetime, startup failure cleanup, authorization locks, and transaction ownership shared. - Reduce `chat-channels.ts` from 3,570 to 3,548 lines. ## Verification - The same 46 runtime/admission integration cases and all 18 Photon channel cases passed before and after in the owned Linux container. - The existing first-channel Teams orphan-admission case passed before and after moving its eligibility predicate. - The same 94 SDK adapter and lifecycle-normalization tests passed locally. - Server typecheck and module-boundary checks passed. - Token comparison confirms ordered SDK batching and every processor argument, plus Photon invalidation and its live runtime lookup. - The shared factory creates no cache or worker. It reads the ingress policy for each message and awaits processing in order. - The extraction rebased without conflicts onto master after #16276 and #16283 merged. Git range-diff reports the same patch. Final head `7b63b569e2a0dbcc959310f565f77f367fa64aa0` has the same patch by range-diff. The 94 SDK/lifecycle tests, server typecheck, and boundary check passed on the final head. All CI gates pass on that exact head. Greptile reviewed it at 5/5 with no inline findings or unresolved review threads. - One Runner CI test failed its original short timing window. The unchanged test passed earlier-head CI and local reproduction. The single failed-job retry passed. - Cumulative `pnpm -r typecheck` and `pnpm build` passed. All 1,321 channel integration cases passed. The normal UI, CLI, database, adapter, plugin, and serialized API test selections ran. - The first Linux `pnpm test:run` returned 1. Original-source runs reproduced the local failures. Container-only repairs supplied missing tools, process reaping, and sufficient Node heap. The existing mocked-fetch CLI telemetry fixture passed with the image opt-out flag overridden only for that fixture. - One local venue limitation remains: adapter-utils T24 fails unchanged on both original and final source in the Linux container. Its entire 56-case suite passes before and after on the host. All other failed or aborted suites passed in repaired reruns. No test assertion or timeout changed. - These fixtures prove application integration. They do not prove live provider execution. ## Risks - A captured option value or changed callback order could affect deferred ingress or stale-runtime handling. - The ingress callback and runtime getter remain live. Core maps and receiver lease operations keep one owner. - No schema migration or public API change is included. ## Model Used OpenAI GPT-6, running in Codex with reasoning, tool use, and code execution. The exact deployment identifier and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- server/src/services/chat-channels.ts | 74 +++++++------------ .../chat-channels/discord/messages.ts | 14 ++++ .../imessage-photon/runtime-callbacks.ts | 22 +++++- .../chat-channels/microsoft-teams/messages.ts | 6 ++ .../chat-channels/runtime/sdk-messages.ts | 60 +++++++++++++++ .../services/chat-channels/slack/messages.ts | 12 +++ .../services/chat-channels/speko/runtime.ts | 2 + 7 files changed, 140 insertions(+), 50 deletions(-) create mode 100644 server/src/services/chat-channels/runtime/sdk-messages.ts 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;