From 75189b8762cb0a378a438c98a8523ab10682dcca Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:39:08 -0500 Subject: [PATCH] refactor(server): isolate channel runtime admission and maintenance (#16235) ## Thinking Path > - Paperclip helps people manage AI agents for work. > - Chat channels turn provider messages into shared task workflows. > - The main service still implements provider details beside common authorization and persistence. > - These details belong in the existing channel folders. > - This pull request moves Discord runtime command capture and root receipts, Speko runtime and authority, and GitHub maintenance scheduling behind explicit typed functions. > - The service keeps its shared workflow and selects the provider implementation directly. ## Linked Issues or Issue Description Refs: #16192. Stack part 5/5. Depends on #16233. Base: `master` (parent PR merged). **What existing behavior does this improve?** The ownership and readability of provider behavior in the chat channel service. **Current behavior** The service implements Discord runtime command capture and root receipts, Speko runtime and authority, and GitHub maintenance scheduling inline. **Proposed behavior** Channel modules own these details. The core calls their typed functions. Runtime and database effects retain their current order. **Reason and benefit** Maintainers can find channel behavior in its channel folder without following a generic provider framework. **Breaking changes** None intended. Public service entry points and durable data shapes stay compatible. ## What Changed - Extract Discord runtime command capture and root receipts, Speko runtime and authority, and GitHub maintenance scheduling into the existing channel folders. - Keep shared company scope, authorization, and transaction ownership in their current workflow. - Reduce `chat-channels.ts` from 3715 to 3649 lines. ## Verification - The same 167 selected chat-runtime cases and 87 voice ownership cases passed before and after extraction. Full workspace typecheck and build passed. Token parity checks passed for six moved control-flow blocks. - Server typecheck passed. - `node scripts/check-module-boundaries.mjs` passed. - Full workspace typecheck and build passed on the cumulative stack. All current-head CI gates passed. Greptile completed at 5/5 with no findings. - The full local `pnpm test:run` command exited with three database fixture-startup hook timeouts. It reported 19,529 passing tests and 250 skipped tests. No assertion failed. Its later test groups did not run because the driver stops on the first failing group. - The timed-out files were `environment-custom-images-service.test.ts`, `generic-mcp-connection.test.ts`, and `modules/wake-queue/adapters/postgres.test.ts`. All 166 tests in these files passed in isolation on both baseline `edcfcaf3f6` and stack head `faaca26d46`, with existing timeouts and assertions. The tests and database-startup helper are unchanged. - Provider fixtures prove application integration. They do not prove live third-party execution. ## Risks - Incorrect inputs or changed await order could affect provider handling and recovery. - Existing integration tests exercise the assembled service before and after extraction. - No database migration or 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 | 94 +++---------------- .../chat-channels/discord/root-threads.ts | 25 ++++- .../chat-channels/discord/runtime-commands.ts | 21 +++++ .../chat-channels/github/maintenance.ts | 27 ++++++ .../services/chat-channels/speko/runtime.ts | 45 +++++++++ 5 files changed, 131 insertions(+), 81 deletions(-) create mode 100644 server/src/services/chat-channels/discord/runtime-commands.ts create mode 100644 server/src/services/chat-channels/github/maintenance.ts create mode 100644 server/src/services/chat-channels/speko/runtime.ts diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 9732b628c03..6b205ed94a9 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,6 @@ +import { createGitHubMaintenance } from "./chat-channels/github/maintenance.js"; +import { createSpekoChannelService, authorizeSpekoPublicVoiceTask, replaceSpekoChannelRuntime } 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"; import { initializeGitHubWebhookRuntime } from "./chat-channels/github/webhook-runtime.js"; @@ -114,9 +117,7 @@ import { nextChatActionAt } from "./chat-action-work.js"; import type { ScheduledChatAction } from "./chat-work-notifications.js"; import { notifyChatDeliveryWork, notifyChatPublicationWork } from "./chat-work-notifications.js"; -import { voiceSessionService } from "./voice/voice-session-service.js"; -import { createSpekoProvider } from "./voice/speko-provider.js"; -import { currentVoiceCredentialFingerprint, voiceSessionStore } from "./voice/voice-session-store.js"; +import { voiceSessionStore } from "./voice/voice-session-store.js"; import { createChatRuntime, type ChatRuntime } from "./chat-runtime.js"; import type { ChatEndpointRuntime } from "./voice/voice-runtime.js"; @@ -130,11 +131,8 @@ import { canPublishIssueToChatAudience } from "./authorization.js"; import { buildChatCommunicationGuidance } from "./chat-communication-guidance.js"; import { assertChatTaskCreationAccess, resolveChatExecutionDefaults } from "./chat-execution-defaults.js"; -import { githubReviewCheckService } from "./chat-github-checks.js"; import { githubAutomaticAdmission } from "./chat-github-events.js"; import type { GitHubReviewEventContext, GitHubIssueEventContext, GitHubReviewPolicy } from "@paperclipai/shared"; -import { githubChatReviewService } from "./chat-github-reviews.js"; -import { githubResponseCommentService } from "./chat-github-response-comments.js"; import { resumeExternalConversation } from "./slack-conversation-state.js"; @@ -181,7 +179,7 @@ import { import { parseChatWebhookPublicBaseUrl } from "../chat-webhook-public-url.js"; -import { HttpError, conflict, forbidden, notFound, unprocessable } from "../errors.js"; +import { conflict, forbidden, notFound, unprocessable } from "../errors.js"; import { logger } from "../middleware/logger.js"; import { redactSensitiveText } from "../redaction.js"; import type { StorageService } from "../storage/types.js"; @@ -211,9 +209,6 @@ import { } from "./chat-github-attachments.js"; import { getExternalChannelBindingSummary } from "./chat-channel-binding.js"; -import { - readRegisteredDiscordCommandRegistration, -} from "./chat-discord-command-registration-store.js"; import { @@ -538,15 +533,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const getWebhookPublicBaseUrl = () => configuredWebhookPublicBaseUrl ?? getPublicBaseUrl(); const voiceStore = voiceSessionStore(db, { allowLocalBoard: options.allowLocalVoiceBoard === true }); const { findAuthorizedIdentityLink, ensurePrincipal, sponsorAllowsGuest, lockCurrentPrincipalAuthorization, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, ensureResource } = createChatInboundAuthority(db, { voiceStore, runtimeCallbackEndpoint }); - const voice = voiceSessionService(db, { + const voice = createSpekoChannelService(db, { endpointRecord, resolveCredentialRefs: (...args) => resolveCredentialRefs(...args), fetchImpl, allowLocalBoard: options.allowLocalVoiceBoard === true, - credentials: async (endpoint, expectedFingerprint) => { - const record = await endpointRecord(endpoint.id); - if (!record || record.endpoint.companyId !== endpoint.companyId) throw notFound("Voice connection not found"); - if (expectedFingerprint && await currentVoiceCredentialFingerprint(db, endpoint.companyId, record.credentialSecretRefs) !== expectedFingerprint) throw conflict("Voice credentials changed during setup"); - return resolveCredentialRefs(endpoint, record.credentialSecretRefs); - }, - provider: (apiKey) => createSpekoProvider(apiKey, fetchImpl), onQuestionAnswered: (interactionId) => scheduleMessageProcessing(async () => { await questionResponses.deliver(interactionId); await processPendingPublications(); @@ -854,7 +842,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); const { handleReaction, processPendingReactionDeliveries } = createInboundReactions(db, { lockCurrentPrincipalAuthorization, runtimeCallbackRecord, runtimeCallbackEndpoint, ensurePrincipal, durableExternalThreadIdentity, recordCurrentMicrosoftTeamsRoute, lifecycleRuntimeFence, deliveryReady, redactError, nonDirectDestinationAllowed, reactionLinkPreflightBarrier: () => options.reactionLinkPreflightBarrier?.() ?? Promise.resolve(), reactionReplayEndpointLockBarrier: () => options.reactionReplayEndpointLockBarrier?.() ?? Promise.resolve(), reactionReplayConversationLockBarrier: () => options.reactionReplayConversationLockBarrier?.() ?? Promise.resolve() }); const { addReceiptReaction, processReceiptReaction, processPendingReceiptReactions } = createReceiptReactions(db, { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, runtimeFor, fetchImpl, runtime, runtimeGeneration, credentialFingerprint, redactError, MAX_ERROR_TEXT, receiptReactionTransportBarrier: () => options.receiptReactionTransportBarrier?.() ?? Promise.resolve() }); - const { admitDiscordRootMention, discordRootThreadPending, reconcileDiscordRootThread } = createDiscordRootThreads(db, { ensureDiscordGatewayRuntimeIsCurrent, withCredentialMutationLease, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, processMessage, redactError, scheduleConversationDrain, lifecycleRuntimeFence, runtimeFor, DISCORD_ROOT_THREAD_CONFIRMATION_DELAY_MS, discordRootThreadTransportBarrier: () => options.discordRootThreadTransportBarrier?.() ?? Promise.resolve() }); + const { admitDiscordRootMention, discordRootThreadPending, reconcileDiscordRootThread, promoteDiscordRootReceipt } = createDiscordRootThreads(db, { ensureDiscordGatewayRuntimeIsCurrent, withCredentialMutationLease, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, processMessage, redactError, scheduleConversationDrain, lifecycleRuntimeFence, runtimeFor, DISCORD_ROOT_THREAD_CONFIRMATION_DELAY_MS, discordRootThreadTransportBarrier: () => options.discordRootThreadTransportBarrier?.() ?? Promise.resolve() }); const { handleDiscordGatewayEvent } = createDiscordGatewayEvents(db, { ensureDiscordGatewayRuntimeIsCurrent, discordGatewayRuntimeIsCurrent, runtimeCallbackRecord, runtimeCallbackEndpoint, applyProviderLifecycleEffect, runtimeGeneration, discordGatewayEventBarrier: (event) => options.discordGatewayEventBarrier?.(event) ?? Promise.resolve() }); const { handlePhotonFailure, handlePhotonEvent } = createPhotonEvents(db, { runtimeCallbackRecord, runtimeCallbackEndpoint, invalidateRuntime, ensurePrincipal, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, scheduleMessageProcessing, processPendingPublications, questionResponses, redactError, renewDiscordGatewayOwnershipForMessageAdmission, handleSdkMessage }); const { handleTelegramGenerationStopped, reserveTelegramPublicationDraft } = createTelegramPublicationDrafts(db, { runtimeCallbackRecord, runtimeCallbackEndpoint, runtimeFor, runtime, runtimeContexts, credentialFingerprint, authorizeRetainedChatSourcePublication }); @@ -999,26 +987,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } recordChatWebhookStage("runtime_initializing", record.endpoint.id); const credentials = await resolveCredentials(record.endpoint); - const discordCommands = - record.endpoint.provider === "discord" && - record.endpoint.botExternalId && - record.endpoint.providerAccountId - ? await readRegisteredDiscordCommandRegistration(db, { - companyId: record.endpoint.companyId, - endpointId: record.endpoint.id, - applicationId: record.endpoint.botExternalId, - guildId: record.endpoint.providerAccountId, - }) - : null; - context.discordCommandId = - discordCommands && record.endpoint.capabilities.slashCommands - ? discordCommands.receipt.commandId - : undefined; + await prepareDiscordRuntimeCommand(db, record.endpoint, context); instance = record.endpoint.provider === "speko" - ? await (() => { - 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 }); - })() + ? await replaceSpekoChannelRuntime(runtime, record, persistence) : await runtime.replaceEndpoint({ companyId: record.endpoint.companyId, endpointId: record.endpoint.id, @@ -1499,27 +1470,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // The adapter invokes the ordinary message callback only after Discord // has returned the created thread. Promote the pre-admission receipt // atomically so the normal ordered drain can begin immediately. - const candidateId = candidate.id; - const [promoted] = await db.transaction(async (tx) => { - await notifyChatDeliveryWork(tx); - return tx - .update(chatDeliveries) - .set({ - state: "received", - normalizedEvent: sql`${chatDeliveries.normalizedEvent} - 'providerThreadPending'`, - nextAttemptAt: scheduledAt ? new Date(scheduledAt) : null, - redactedError: null, - updatedAt: new Date(), - }) - .where( - and( - eq(chatDeliveries.id, candidateId), - eq(chatDeliveries.state, "retry"), - sql`${chatDeliveries.normalizedEvent}->>'providerThreadPending' = 'true'`, - ), - ) - .returning(); - }); + const promoted = await promoteDiscordRootReceipt(candidate.id, scheduledAt); if (promoted) candidate = promoted; } const now = new Date(); @@ -1810,12 +1761,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // and exact call-owned task with its low-trust containment still intact. let publicVoiceTaskAllowed = false; if (endpoint.provider === "speko" && principalResolution.userId === null && !principalResolution.linkedDenied) { - try { - const authority = await db.transaction(tx => voiceStore.authorizePrincipal(tx, endpoint.companyId, endpoint.id, principalResolution.principal.externalId)); - publicVoiceTaskAllowed = authority.allowed && authority.userId === null; - } catch (error) { - if (!(error instanceof HttpError && [403, 404].includes(error.status))) throw error; - } + publicVoiceTaskAllowed = await authorizeSpekoPublicVoiceTask(db, endpoint, principalResolution.principal.externalId, voiceStore.authorizePrincipal); } const principalAllowed = !principalResolution.linkedDenied && @@ -3424,21 +3370,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { return publicationQueue.processPendingPublications(...args); } - let githubMaintenancePending = false; - function scheduleGitHubMaintenance() { - if (githubMaintenancePending || shuttingDown) return; - githubMaintenancePending = true; - scheduleMessageProcessing(async () => { - try { - // Provider I/O must not block Slack, Teams, or ordinary message receipts. - await githubChatReviewService(db, fetchImpl).processPending(); - await githubResponseCommentService(db, fetchImpl).processPending(); - await githubReviewCheckService(db, fetchImpl).processPending(); - } finally { - githubMaintenancePending = false; - } - }); - } + const { scheduleGitHubMaintenance } = createGitHubMaintenance(db, fetchImpl, { + scheduleMessageProcessing, isShuttingDown: () => shuttingDown, + }); async function processPublicationMaintenance(limit = 25) { scheduleTeamsFileMaintenance(); diff --git a/server/src/services/chat-channels/discord/root-threads.ts b/server/src/services/chat-channels/discord/root-threads.ts index 1772ce375bb..4f7ce603c25 100644 --- a/server/src/services/chat-channels/discord/root-threads.ts +++ b/server/src/services/chat-channels/discord/root-threads.ts @@ -566,5 +566,28 @@ export function createDiscordRootThreads(db: Db, options: DiscordRootThreadsOpti return false; } } - return { admitDiscordRootMention, discordRootThreadPending, reconcileDiscordRootThread }; + async function promoteDiscordRootReceipt(candidateId: string, scheduledAt: number | null): Promise { + const [promoted] = await db.transaction(async (tx) => { + await notifyChatDeliveryWork(tx); + return tx + .update(chatDeliveries) + .set({ + state: "received", + normalizedEvent: sql`${chatDeliveries.normalizedEvent} - 'providerThreadPending'`, + nextAttemptAt: scheduledAt ? new Date(scheduledAt) : null, + redactedError: null, + updatedAt: new Date(), + }) + .where( + and( + eq(chatDeliveries.id, candidateId), + eq(chatDeliveries.state, "retry"), + sql`${chatDeliveries.normalizedEvent}->>'providerThreadPending' = 'true'`, + ), + ) + .returning(); + }); + return promoted; + } + return { admitDiscordRootMention, discordRootThreadPending, reconcileDiscordRootThread, promoteDiscordRootReceipt }; } diff --git a/server/src/services/chat-channels/discord/runtime-commands.ts b/server/src/services/chat-channels/discord/runtime-commands.ts new file mode 100644 index 00000000000..25ebf7a3957 --- /dev/null +++ b/server/src/services/chat-channels/discord/runtime-commands.ts @@ -0,0 +1,21 @@ +import type { Db } from "@paperclipai/db"; +import type { EndpointRow, RuntimeContext } from "../runtime-types.js"; +import { readRegisteredDiscordCommandRegistration } from "../../chat-discord-command-registration-store.js"; + +export async function prepareDiscordRuntimeCommand(db: Db, endpoint: EndpointRow, context: Pick): Promise { + const discordCommands = + endpoint.provider === "discord" && + endpoint.botExternalId && + endpoint.providerAccountId + ? await readRegisteredDiscordCommandRegistration(db, { + companyId: endpoint.companyId, + endpointId: endpoint.id, + applicationId: endpoint.botExternalId, + guildId: endpoint.providerAccountId, + }) + : null; + context.discordCommandId = + discordCommands && endpoint.capabilities.slashCommands + ? discordCommands.receipt.commandId + : undefined; +} diff --git a/server/src/services/chat-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts new file mode 100644 index 00000000000..3e45da8c350 --- /dev/null +++ b/server/src/services/chat-channels/github/maintenance.ts @@ -0,0 +1,27 @@ +import type { Db } from "@paperclipai/db"; +import { githubChatReviewService } from "../../chat-github-reviews.js"; +import { githubResponseCommentService } from "../../chat-github-response-comments.js"; +import { githubReviewCheckService } from "../../chat-github-checks.js"; + +export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fetch, options: { + scheduleMessageProcessing: (task: () => Promise) => void; + isShuttingDown: () => boolean; +}) { + let githubMaintenancePending = false; + function scheduleGitHubMaintenance() { + if (githubMaintenancePending || options.isShuttingDown()) return; + githubMaintenancePending = true; + options.scheduleMessageProcessing(async () => { + try { + // Provider I/O must not block Slack, Teams, or ordinary message receipts. + await githubChatReviewService(db, fetchImpl).processPending(); + await githubResponseCommentService(db, fetchImpl).processPending(); + await githubReviewCheckService(db, fetchImpl).processPending(); + } finally { + githubMaintenancePending = false; + } + }); + } + + return { scheduleGitHubMaintenance }; +} diff --git a/server/src/services/chat-channels/speko/runtime.ts b/server/src/services/chat-channels/speko/runtime.ts new file mode 100644 index 00000000000..1fa6113a32e --- /dev/null +++ b/server/src/services/chat-channels/speko/runtime.ts @@ -0,0 +1,45 @@ +import type { Db } from "@paperclipai/db"; +import { conflict, HttpError, notFound } from "../../../errors.js"; +import { voiceSessionService } from "../../voice/voice-session-service.js"; +import { voiceSessionStore, currentVoiceCredentialFingerprint } from "../../voice/voice-session-store.js"; +import { createSpekoProvider } from "../../voice/speko-provider.js"; +import type { ChatRuntime } from "../../chat-runtime.js"; +import type { ChatSdkRuntime } from "../../chat-sdk-runtime.js"; +import type { ChatSdkStatePersistence } from "../../chat-sdk-state.js"; +import type { createChatCredentials } from "../credentials.js"; +import type { EndpointRow, EndpointRecord } from "../runtime-types.js"; + +type SpekoChannelOptions = Pick[1], "allowLocalBoard" | "onQuestionAnswered"> & { + endpointRecord: (endpointId: string) => Promise; + resolveCredentialRefs: ReturnType["resolveCredentialRefs"]; + fetchImpl: typeof globalThis.fetch; +}; +export function createSpekoChannelService(db: Db, options: SpekoChannelOptions) { + const { endpointRecord, resolveCredentialRefs, fetchImpl } = options; + return voiceSessionService(db, { + allowLocalBoard: options.allowLocalBoard, + credentials: async (endpoint, expectedFingerprint) => { + const record = await endpointRecord(endpoint.id); + if (!record || record.endpoint.companyId !== endpoint.companyId) throw notFound("Voice connection not found"); + if (expectedFingerprint && await currentVoiceCredentialFingerprint(db, endpoint.companyId, record.credentialSecretRefs) !== expectedFingerprint) throw conflict("Voice credentials changed during setup"); + return resolveCredentialRefs(endpoint, record.credentialSecretRefs); + }, + provider: (apiKey) => createSpekoProvider(apiKey, fetchImpl), + onQuestionAnswered: options.onQuestionAnswered, + }); +} + +export async function authorizeSpekoPublicVoiceTask(db: Db, endpoint: EndpointRow, externalId: string, authorizePrincipal: ReturnType["authorizePrincipal"]): Promise { + try { + const authority = await db.transaction(tx => authorizePrincipal(tx, endpoint.companyId, endpoint.id, externalId)); + return authority.allowed && authority.userId === null; + } catch (error) { + if (!(error instanceof HttpError && [403, 404].includes(error.status))) throw error; + } + return false; +} + +export function replaceSpekoChannelRuntime(runtime: ChatRuntime | ChatSdkRuntime, record: EndpointRecord, persistence: ChatSdkStatePersistence) { + 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 }); +}