OpenClaw源码解读之通道与路由系统

export type ChannelPlugin<ResolvedAccount = any, Probe = unknown, Audit = unknown> = {  id: ChannelId;  meta: ChannelMeta;  capabilities: ChannelCapabilities;  defaults?: { queue?: { debounceMs?: number } };  reload?: { configPrefixes: string[]; noopPrefixes?: string[] };  onboarding?: ChannelOnboardingAdapter;  config: ChannelConfigAdapter<ResolvedAccount>;  configSchema?: ChannelConfigSchema;  setup?: ChannelSetupAdapter;  pairing?: ChannelPairingAdapter;  security?: ChannelSecurityAdapter<ResolvedAccount>;  groups?: ChannelGroupAdapter;  mentions?: ChannelMentionAdapter;  outbound?: ChannelOutboundAdapter;  status?: ChannelStatusAdapter<ResolvedAccount, Probe, Audit>;  gateway?: ChannelGatewayAdapter<ResolvedAccount>;  actions?: ChannelMessageActionAdapter;  heartbeat?: ChannelHeartbeatAdapter;  agentTools?: ChannelAgentToolFactory | ChannelAgentTool[];  // ... 还有 streaming, threading, messaging, agentPrompt, directory, resolver 等};

export type ChannelCapabilities = {  chatTypes: Array<ChatType | "thread">;  polls?: boolean;  reactions?: boolean;  edit?: boolean;  unsend?: boolean;  reply?: boolean;  effects?: boolean;  groupManagement?: boolean;  threads?: boolean;  media?: boolean;  nativeCommands?: boolean;  blockStreaming?: boolean;};

export type ChannelConfigAdapter<ResolvedAccount> = {  listAccountIds: (cfg: OpenClawConfig) => string[];  resolveAccount: (cfg: OpenClawConfig, accountId?: string | null) => ResolvedAccount;  defaultAccountId?: (cfg: OpenClawConfig) => string;  isEnabled?: (account: ResolvedAccount, cfg: OpenClawConfig) => boolean;  isConfigured?: (account: ResolvedAccount, cfg: OpenClawConfig) => boolean | Promise<boolean>;  // ...};

// Rules:// - keep this module *light* (no monitors, probes, puppeteer/web login, etc)// - OK: config readers, allowFrom formatting, mention stripping patterns, threading defaults// - shared code should import from here, not from the plugins registryconst DOCKS: Record<ChatChannelId, ChannelDock> = {  telegram: {    id: "telegram",    capabilities: {      chatTypes: ["direct", "group", "channel", "thread"],      nativeCommands: true,      blockStreaming: true,    },    outbound: { textChunkLimit: 4000 },    // ... groups, mentions, threading 等轻量适配器  },  discord: { ... },  whatsapp: { ... },  // ...};

export const CHAT_CHANNEL_ORDER: ChatChannelId[] = [  "telegram", "whatsapp", "discord", "irc", "googlechat",  "slack", "signal", "imessage",];

export function listChannelPlugins(): ChannelPlugin[] { ... }export function getChannelPlugin(id: ChannelId): ChannelPlugin | undefined { ... }

const route = resolveAgentRoute({  cfg: loadConfig(),  channel: "telegram",  accountId: account.accountId,  peer: {    kind: isGroup ? "group" : "direct",    id: peerId,  },  parentPeer,});const baseSessionKey = route.sessionKey;

export function resolveAgentRoute(input: ResolveAgentRouteInput): ResolvedAgentRoute {  const bindings = listBindings(input.cfg).filter((binding) => {    if (!matchesChannel(binding.match, channel)) return false;    return matchesAccountId(binding.match?.accountId, accountId);  });  // 优先级 1：精确 peer 匹配  if (peer) {    const peerMatch = bindings.find((b) => matchesPeer(b.match, peer));    if (peerMatch) return choose(peerMatch.agentId, "binding.peer");  }  // 优先级 2：线程父 peer 继承  if (parentPeer && parentPeer.id) {    const parentPeerMatch = bindings.find((b) => matchesPeer(b.match, parentPeer));    if (parentPeerMatch) return choose(parentPeerMatch.agentId, "binding.peer.parent");  }  // 优先级 3：Guild + 角色匹配（Discord 特有）  if (guildId && memberRoleIds.length > 0) { ... }  // 优先级 4：Guild 匹配  if (guildId) { ... }  // 优先级 5：Team 匹配（Slack 特有）  if (teamId) { ... }  // 优先级 6：Account 匹配（绑定到特定 Bot 账号）  const accountMatch = bindings.find((b) =>    b.match?.accountId?.trim() !== "*" && !b.match?.peer && !b.match?.guildId && !b.match?.teamId  );  if (accountMatch) return choose(accountMatch.agentId, "binding.account");  // 优先级 7：Channel 匹配（通配符，任何账号）  const anyAccountMatch = bindings.find((b) =>    b.match?.accountId?.trim() === "*" && !b.match?.peer && !b.match?.guildId && !b.match?.teamId  );  if (anyAccountMatch) return choose(anyAccountMatch.agentId, "binding.channel");  // 兜底：使用默认 Agent  return choose(resolveDefaultAgentId(input.cfg), "default");}

export function buildAgentPeerSessionKey(params: {  agentId: string;  channel: string;  accountId?: string | null;  peerKind?: ChatType | null;  peerId?: string | null;  dmScope?: "main" | "per-peer" | "per-channel-peer" | "per-account-channel-peer";  identityLinks?: Record<string, string[]>;}): string {  const peerKind = params.peerKind ?? "direct";  if (peerKind === "direct") {    const dmScope = params.dmScope ?? "main";    // ...    if (dmScope === "per-account-channel-peer" && peerId) {      return `agent:${agentId}:${channel}:${accountId}:direct:${peerId}`;    }    if (dmScope === "per-channel-peer" && peerId) {      return `agent:${agentId}:${channel}:direct:${peerId}`;    }    if (dmScope === "per-peer" && peerId) {      return `agent:${agentId}:direct:${peerId}`;    }    return buildAgentMainSessionKey({ agentId, mainKey });  }  // 群聊/频道：总是按 channel + peerKind + peerId 隔离  return `agent:${agentId}:${channel}:${peerKind}:${peerId}`;}

export async function getReplyFromConfig(  ctx: MsgContext,  opts?: GetReplyOptions,): Promise<ReplyPayload | ReplyPayload[] | undefined> {  // 1. 从会话键解析 Agent ID  const agentId = resolveSessionAgentId({ sessionKey: agentSessionKey, config: cfg });  // 2. 合并技能过滤器  const mergedSkillFilter = mergeSkillFilters(opts?.skillFilter, resolveAgentSkillsFilter(cfg, agentId));  // 3. 解析模型（默认模型 or 心跳覆盖模型）  const { defaultProvider, defaultModel, aliasIndex } = resolveDefaultModel({ cfg, agentId });  // 4. 确保 Agent 工作目录存在  const workspace = await ensureAgentWorkspace({ dir: workspaceDirRaw });  // 5. 媒体/链接理解（图片转文字、URL 抓取等）  await applyMediaUnderstanding({ ctx: finalized, cfg, agentDir, activeModel });  await applyLinkUnderstanding({ ctx: finalized, cfg });  // 6. 命令授权检查  resolveCommandAuthorization({ ctx: finalized, cfg, commandAuthorized });  // 7. 初始化会话状态（加载历史、检查是否新会话、处理 reset）  const sessionState = await initSessionState({ ctx: finalized, cfg, commandAuthorized });  // 8. 解析指令（/think, /verbose, /model 等用户指令）  const directiveResult = await resolveReplyDirectives({ ... });  // 9. 处理内联动作（/status, /compact, /session 等内置命令）  const inlineActionResult = await handleInlineActions({ ... });  // 10. 运行 Agent 生成回复  return runPreparedReply({ ... });}

export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDispatcher {  let sendChain: Promise<void> = Promise.resolve();  let pending = 0;  let sentFirstBlock = false;  const enqueue = (kind: ReplyDispatchKind, payload: ReplyPayload) => {    const normalized = normalizeReplyPayloadInternal(payload, { ... });    if (!normalized) return false;    pending += 1;    const shouldDelay = kind === "block" && sentFirstBlock;    if (kind === "block") sentFirstBlock = true;    sendChain = sendChain      .then(async () => {        if (shouldDelay) {          const delayMs = getHumanDelay(options.humanDelay);          if (delayMs > 0) await sleep(delayMs);        }        await options.deliver(normalized, { kind });      })      .catch((err) => options.onError?.(err, { kind }))      .finally(() => {        pending -= 1;        if (pending === 0) options.onIdle?.();      });    return true;  };  return {    sendToolResult: (payload) => enqueue("tool", payload),    sendBlockReply: (payload) => enqueue("block", payload),    sendFinalReply: (payload) => enqueue("final", payload),    waitForIdle: () => sendChain,  };}

export type ChannelOutboundAdapter = {  deliveryMode: "direct" | "gateway" | "hybrid";  chunker?: ((text: string, limit: number) => string[]) | null;  textChunkLimit?: number;  sendPayload?: (ctx: ChannelOutboundPayloadContext) => Promise<OutboundDeliveryResult>;  sendText?: (ctx: ChannelOutboundContext) => Promise<OutboundDeliveryResult>;  sendMedia?: (ctx: ChannelOutboundContext) => Promise<OutboundDeliveryResult>;  sendPoll?: (ctx: ChannelPollContext) => Promise<ChannelPollResult>;};