feat(routing): route replies to originating channel

Implement reply routing based on OriginatingChannel/OriginatingTo fields.
This ensures replies go back to the provider where the message originated
instead of using the session's lastChannel.

Changes:
- Add OriginatingChannel/OriginatingTo fields to MsgContext (templating.ts)
- Add originatingChannel/originatingTo fields to FollowupRun (queue.ts)
- Create route-reply.ts with provider-agnostic router
- Update all providers (Telegram, Slack, Discord, Signal, iMessage)
  to pass originating channel info
- Update reply.ts to pass originating channel to followupRun
- Update followup-runner.ts to use route-reply for originating channels

This addresses the issue where messages from one provider (e.g., Slack)
would receive replies on a different provider (e.g., Telegram) because
the queue used the last active dispatcher instead of the originating one.
This commit is contained in:
Josh Lehman
2026-01-06 10:58:45 -08:00
committed by Peter Steinberger
parent 514fcfe77e
commit 9d50ebad7d
10 changed files with 238 additions and 12 deletions

View File

@@ -13,6 +13,7 @@ import { SILENT_REPLY_TOKEN } from "../tokens.js";
import type { GetReplyOptions, ReplyPayload } from "../types.js";
import type { FollowupRun } from "./queue.js";
import { extractReplyToTag } from "./reply-tags.js";
import { isRoutableChannel, routeReply } from "./route-reply.js";
import { incrementCompactionCount } from "./session-updates.js";
import type { TypingController } from "./typing.js";
@@ -37,11 +38,28 @@ export function createFollowupRunner(params: {
agentCfgContextTokens,
} = params;
const sendFollowupPayloads = async (payloads: ReplyPayload[]) => {
if (!opts?.onBlockReply) {
/**
* Sends followup payloads, routing to the originating channel if set.
*
* When originatingChannel/originatingTo are set on the queued run,
* replies are routed directly to that provider instead of using the
* session's current dispatcher. This ensures replies go back to
* where the message originated.
*/
const sendFollowupPayloads = async (
payloads: ReplyPayload[],
queued: FollowupRun,
) => {
// Check if we should route to originating channel.
const { originatingChannel, originatingTo } = queued;
const shouldRouteToOriginating =
isRoutableChannel(originatingChannel) && originatingTo;
if (!shouldRouteToOriginating && !opts?.onBlockReply) {
logVerbose("followup queue: no onBlockReply handler; dropping payloads");
return;
}
for (const payload of payloads) {
if (!payload?.text && !payload?.mediaUrl && !payload?.mediaUrls?.length) {
continue;
@@ -54,7 +72,23 @@ export function createFollowupRunner(params: {
continue;
}
await typing.startTypingOnText(payload.text);
await opts.onBlockReply(payload);
// Route to originating channel if set, otherwise fall back to dispatcher.
if (shouldRouteToOriginating) {
const result = await routeReply({
payload,
channel: originatingChannel,
to: originatingTo,
cfg: queued.run.config,
});
if (!result.ok) {
logVerbose(
`followup queue: route-reply failed: ${result.error ?? "unknown error"}`,
);
}
} else if (opts?.onBlockReply) {
await opts.onBlockReply(payload);
}
}
};
@@ -210,7 +244,7 @@ export function createFollowupRunner(params: {
}
}
await sendFollowupPayloads(replyTaggedPayloads);
await sendFollowupPayloads(replyTaggedPayloads, queued);
} finally {
typing.markRunComplete();
}

View File

@@ -3,6 +3,7 @@ import { parseDurationMs } from "../../cli/parse-duration.js";
import type { ClawdbotConfig } from "../../config/config.js";
import type { SessionEntry } from "../../config/sessions.js";
import { defaultRuntime } from "../../runtime.js";
import type { OriginatingChannelType } from "../templating.js";
import type { ElevatedLevel, ThinkLevel, VerboseLevel } from "./directives.js";
export type QueueMode =
| "steer"
@@ -22,6 +23,17 @@ export type FollowupRun = {
prompt: string;
summaryLine?: string;
enqueuedAt: number;
/**
* Originating channel for reply routing.
* When set, replies should be routed back to this provider
* instead of using the session's lastChannel.
*/
originatingChannel?: OriginatingChannelType;
/**
* Originating destination for reply routing.
* The chat/channel/user ID where the reply should be sent.
*/
originatingTo?: string;
run: {
agentId: string;
agentDir: string;

View File

@@ -0,0 +1,135 @@
/**
* Provider-agnostic reply router.
*
* Routes replies to the originating channel based on OriginatingChannel/OriginatingTo
* instead of using the session's lastChannel. This ensures replies go back to the
* provider where the message originated, even when the main session is shared
* across multiple providers.
*/
import type { ClawdbotConfig } from "../../config/config.js";
import { sendMessageDiscord } from "../../discord/send.js";
import { sendMessageIMessage } from "../../imessage/send.js";
import { sendMessageSignal } from "../../signal/send.js";
import { sendMessageSlack } from "../../slack/send.js";
import { sendMessageTelegram } from "../../telegram/send.js";
import type { OriginatingChannelType } from "../templating.js";
import type { ReplyPayload } from "../types.js";
export type RouteReplyParams = {
/** The reply payload to send. */
payload: ReplyPayload;
/** The originating channel type (telegram, slack, etc). */
channel: OriginatingChannelType;
/** The destination chat/channel/user ID. */
to: string;
/** Config for provider-specific settings. */
cfg: ClawdbotConfig;
};
export type RouteReplyResult = {
/** Whether the reply was sent successfully. */
ok: boolean;
/** Optional message ID from the provider. */
messageId?: string;
/** Error message if the send failed. */
error?: string;
};
/**
* Routes a reply payload to the specified channel.
*
* This function provides a unified interface for sending messages to any
* supported provider. It's used by the followup queue to route replies
* back to the originating channel when OriginatingChannel/OriginatingTo
* are set.
*/
export async function routeReply(
params: RouteReplyParams,
): Promise<RouteReplyResult> {
const { payload, channel, to } = params;
const text = payload.text ?? "";
const mediaUrl = payload.mediaUrl ?? payload.mediaUrls?.[0];
// Skip empty replies.
if (!text.trim() && !mediaUrl) {
return { ok: true };
}
try {
switch (channel) {
case "telegram": {
const result = await sendMessageTelegram(to, text, { mediaUrl });
return { ok: true, messageId: result.messageId };
}
case "slack": {
const result = await sendMessageSlack(to, text, { mediaUrl });
return { ok: true, messageId: result.messageId };
}
case "discord": {
const result = await sendMessageDiscord(to, text, { mediaUrl });
return { ok: true, messageId: result.messageId };
}
case "signal": {
const result = await sendMessageSignal(to, text, { mediaUrl });
return { ok: true, messageId: result.messageId };
}
case "imessage": {
const result = await sendMessageIMessage(to, text, { mediaUrl });
return { ok: true, messageId: result.messageId };
}
case "whatsapp": {
// WhatsApp doesn't have a standalone send function in this codebase.
// Falls through to unknown channel handling.
return {
ok: false,
error: `WhatsApp routing not yet implemented`,
};
}
case "webchat": {
// Webchat is typically handled differently (real-time WebSocket).
// Falls through to unknown channel handling.
return {
ok: false,
error: `Webchat routing not supported for queued replies`,
};
}
default: {
// Exhaustive check for unknown channel types.
const _exhaustive: never = channel;
return {
ok: false,
error: `Unknown channel: ${String(_exhaustive)}`,
};
}
}
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return {
ok: false,
error: `Failed to route reply to ${channel}: ${message}`,
};
}
}
/**
* Checks if a channel type is routable via routeReply.
*
* Some channels (webchat, whatsapp) require special handling and
* cannot be routed through this generic interface.
*/
export function isRoutableChannel(
channel: OriginatingChannelType | undefined,
): channel is "telegram" | "slack" | "discord" | "signal" | "imessage" {
if (!channel) return false;
return ["telegram", "slack", "discord", "signal", "imessage"].includes(
channel,
);
}