import type { ClawdbotConfig } from "../../config/config.js"; import { logVerbose } from "../../globals.js"; import { isDiagnosticsEnabled } from "../../infra/diagnostic-events.js"; import { logMessageProcessed, logMessageQueued, logSessionStateChange, } from "../../logging/diagnostic.js"; import { getGlobalHookRunner } from "../../plugins/hook-runner-global.js"; import { getReplyFromConfig } from "../reply.js"; import type { FinalizedMsgContext } from "../templating.js"; import type { GetReplyOptions, ReplyPayload } from "../types.js"; import { formatAbortReplyText, tryFastAbortFromMessage } from "./abort.js"; import { shouldSkipDuplicateInbound } from "./inbound-dedupe.js"; import type { ReplyDispatcher, ReplyDispatchKind } from "./reply-dispatcher.js"; import { isRoutableChannel, routeReply } from "./route-reply.js"; import { maybeApplyTtsToPayload } from "../../tts/tts.js"; export type DispatchFromConfigResult = { queuedFinal: boolean; counts: Record; }; export async function dispatchReplyFromConfig(params: { ctx: FinalizedMsgContext; cfg: ClawdbotConfig; dispatcher: ReplyDispatcher; replyOptions?: Omit; replyResolver?: typeof getReplyFromConfig; }): Promise { const { ctx, cfg, dispatcher } = params; const diagnosticsEnabled = isDiagnosticsEnabled(cfg); const channel = String(ctx.Surface ?? ctx.Provider ?? "unknown").toLowerCase(); const chatId = ctx.To ?? ctx.From; const messageId = ctx.MessageSid ?? ctx.MessageSidFirst ?? ctx.MessageSidLast; const sessionKey = ctx.SessionKey; const startTime = diagnosticsEnabled ? Date.now() : 0; const canTrackSession = diagnosticsEnabled && Boolean(sessionKey); const recordProcessed = ( outcome: "completed" | "skipped" | "error", opts?: { reason?: string; error?: string; }, ) => { if (!diagnosticsEnabled) return; logMessageProcessed({ channel, chatId, messageId, sessionKey, durationMs: Date.now() - startTime, outcome, reason: opts?.reason, error: opts?.error, }); }; const markProcessing = () => { if (!canTrackSession || !sessionKey) return; logMessageQueued({ sessionKey, channel, source: "dispatch" }); logSessionStateChange({ sessionKey, state: "processing", reason: "message_start", }); }; const markIdle = (reason: string) => { if (!canTrackSession || !sessionKey) return; logSessionStateChange({ sessionKey, state: "idle", reason, }); }; if (shouldSkipDuplicateInbound(ctx)) { recordProcessed("skipped", { reason: "duplicate" }); return { queuedFinal: false, counts: dispatcher.getQueuedCounts() }; } const hookRunner = getGlobalHookRunner(); if (hookRunner?.hasHooks("message_received")) { const timestamp = typeof ctx.Timestamp === "number" && Number.isFinite(ctx.Timestamp) ? ctx.Timestamp : undefined; const messageIdForHook = ctx.MessageSidFull ?? ctx.MessageSid ?? ctx.MessageSidFirst ?? ctx.MessageSidLast; const content = typeof ctx.BodyForCommands === "string" ? ctx.BodyForCommands : typeof ctx.RawBody === "string" ? ctx.RawBody : typeof ctx.Body === "string" ? ctx.Body : ""; const channelId = (ctx.OriginatingChannel ?? ctx.Surface ?? ctx.Provider ?? "").toLowerCase(); const conversationId = ctx.OriginatingTo ?? ctx.To ?? ctx.From ?? undefined; void hookRunner .runMessageReceived( { from: ctx.From ?? "", content, timestamp, metadata: { to: ctx.To, provider: ctx.Provider, surface: ctx.Surface, threadId: ctx.MessageThreadId, originatingChannel: ctx.OriginatingChannel, originatingTo: ctx.OriginatingTo, messageId: messageIdForHook, senderId: ctx.SenderId, senderName: ctx.SenderName, senderUsername: ctx.SenderUsername, senderE164: ctx.SenderE164, }, }, { channelId, accountId: ctx.AccountId, conversationId, }, ) .catch((err) => { logVerbose(`dispatch-from-config: message_received hook failed: ${String(err)}`); }); } // Check if we should route replies to originating channel instead of dispatcher. // Only route when the originating channel is DIFFERENT from the current surface. // This handles cross-provider routing (e.g., message from Telegram being processed // by a shared session that's currently on Slack) while preserving normal dispatcher // flow when the provider handles its own messages. // // Debug: `pnpm test src/auto-reply/reply/dispatch-from-config.test.ts` const originatingChannel = ctx.OriginatingChannel; const originatingTo = ctx.OriginatingTo; const currentSurface = (ctx.Surface ?? ctx.Provider)?.toLowerCase(); const shouldRouteToOriginating = isRoutableChannel(originatingChannel) && originatingTo && originatingChannel !== currentSurface; const ttsChannel = shouldRouteToOriginating ? originatingChannel : currentSurface; /** * Helper to send a payload via route-reply (async). * Only used when actually routing to a different provider. * Note: Only called when shouldRouteToOriginating is true, so * originatingChannel and originatingTo are guaranteed to be defined. */ const sendPayloadAsync = async ( payload: ReplyPayload, abortSignal?: AbortSignal, ): Promise => { // TypeScript doesn't narrow these from the shouldRouteToOriginating check, // but they're guaranteed non-null when this function is called. if (!originatingChannel || !originatingTo) return; if (abortSignal?.aborted) return; const result = await routeReply({ payload, channel: originatingChannel, to: originatingTo, sessionKey: ctx.SessionKey, accountId: ctx.AccountId, threadId: ctx.MessageThreadId, cfg, abortSignal, }); if (!result.ok) { logVerbose(`dispatch-from-config: route-reply failed: ${result.error ?? "unknown error"}`); } }; markProcessing(); try { const fastAbort = await tryFastAbortFromMessage({ ctx, cfg }); if (fastAbort.handled) { const payload = { text: formatAbortReplyText(fastAbort.stoppedSubagents), } satisfies ReplyPayload; let queuedFinal = false; let routedFinalCount = 0; if (shouldRouteToOriginating && originatingChannel && originatingTo) { const result = await routeReply({ payload, channel: originatingChannel, to: originatingTo, sessionKey: ctx.SessionKey, accountId: ctx.AccountId, threadId: ctx.MessageThreadId, cfg, }); queuedFinal = result.ok; if (result.ok) routedFinalCount += 1; if (!result.ok) { logVerbose( `dispatch-from-config: route-reply (abort) failed: ${result.error ?? "unknown error"}`, ); } } else { queuedFinal = dispatcher.sendFinalReply(payload); } await dispatcher.waitForIdle(); const counts = dispatcher.getQueuedCounts(); counts.final += routedFinalCount; recordProcessed("completed", { reason: "fast_abort" }); markIdle("message_completed"); return { queuedFinal, counts }; } const replyResult = await (params.replyResolver ?? getReplyFromConfig)( ctx, { ...params.replyOptions, onToolResult: (payload: ReplyPayload) => { const run = async () => { const ttsPayload = await maybeApplyTtsToPayload({ payload, cfg, channel: ttsChannel, kind: "tool", }); if (shouldRouteToOriginating) { await sendPayloadAsync(ttsPayload); } else { dispatcher.sendToolResult(ttsPayload); } }; return run(); }, onBlockReply: (payload: ReplyPayload, context) => { const run = async () => { const ttsPayload = await maybeApplyTtsToPayload({ payload, cfg, channel: ttsChannel, kind: "block", }); if (shouldRouteToOriginating) { await sendPayloadAsync(ttsPayload, context?.abortSignal); } else { dispatcher.sendBlockReply(ttsPayload); } }; return run(); }, }, cfg, ); const replies = replyResult ? (Array.isArray(replyResult) ? replyResult : [replyResult]) : []; let queuedFinal = false; let routedFinalCount = 0; for (const reply of replies) { const ttsReply = await maybeApplyTtsToPayload({ payload: reply, cfg, channel: ttsChannel, kind: "final", }); if (shouldRouteToOriginating && originatingChannel && originatingTo) { // Route final reply to originating channel. const result = await routeReply({ payload: ttsReply, channel: originatingChannel, to: originatingTo, sessionKey: ctx.SessionKey, accountId: ctx.AccountId, threadId: ctx.MessageThreadId, cfg, }); if (!result.ok) { logVerbose( `dispatch-from-config: route-reply (final) failed: ${result.error ?? "unknown error"}`, ); } queuedFinal = result.ok || queuedFinal; if (result.ok) routedFinalCount += 1; } else { queuedFinal = dispatcher.sendFinalReply(ttsReply) || queuedFinal; } } await dispatcher.waitForIdle(); const counts = dispatcher.getQueuedCounts(); counts.final += routedFinalCount; recordProcessed("completed"); markIdle("message_completed"); return { queuedFinal, counts }; } catch (err) { recordProcessed("error", { error: String(err) }); markIdle("message_error"); throw err; } }