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 <noreply@paperclip.ing>

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<Promise<void>>();
   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<boolean> {
-    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<ReturnType<typeof createChatMessageLifecycle>["handleMessageUpdated"]>) {
     return messageLifecycle.handleMessageUpdated(...args);

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<boolean>;
+    messageAdmissionBarrier: () => Promise<void> | void;
+  },
+): Promise<boolean> {
+  if (!runtimeContext || !(await options.ensureRuntimeIsCurrent(endpointId, runtimeContext))) return false;
+  await options.messageAdmissionBarrier();
+  return true;
+}

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);
     })
   };
-}
+}
+
+export const PHOTON_MESSAGE_INGRESS_ONLY = true;
+
+export function discardPhotonInboundMessages(
+  liveInboundMessages: Map<string, LiveInboundMessage>,
+  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<EndpointRow, "status">,
+  latest: Pick<EndpointRow, "status">,
+): boolean {
+  return initial.status === "attention" && ["active", "verifying"].includes(latest.status);
+}

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";
+}

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<Pick<EndpointRecord, "endpoint"> | 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<void>;
+};
+
+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 };
+}

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<EndpointRow, "requireAtMention">,
+  thread: Pick<Thread, "isDM">,
+  trigger: ChatSdkMessageCallbackEvent["trigger"],
+  message: Pick<Message, "isMention">,
+): boolean {
+  return !endpoint.requireAtMention || thread.isDM || trigger === "mention" || message.isMention === true;
+}

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;