数据截至 (上游 commit 53ea1e8ba6fd)
一条消息的一生(主线)
本章讲什么: 把「你在 Telegram 里 @ 了一句话」到「助理回你一句话」这条链路,按主机侧的真实代码顺序走一遍。读完你会知道每一步在哪个文件、失败了会怎样。
1. 先看全景:七步链路
怎么读这张图:竖线是时间,左侧是主机进程,右侧是容器;⇢ 表示「写文件后对方轮询发现」,不是同步调用。
[平台] ①收到消息
│
▼
[适配器] ②onInbound → 盖上 instance 戳
│
▼
[router] ③查 messaging_group + 接线数(一次 SQL)
│ ④对每个接线的 agent 独立判断:该不该 engage
▼
[session] ⑤resolveSession → 写 inbound.db 一行
│
▼
[runner] ⑥wakeContainer → docker run
│
⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ ⇢ [容器轮询到,干活,写 outbound.db]
⇣
[delivery] ⑦轮询 outbound.db → 适配器发回平台
2. ①②入口:适配器把消息交出来
适配器是什么
渠道适配器是「某个 IM 平台」到 NanoClaw 的翻译层。接口定义在 src/channels/adapter.ts:188(ChannelAdapter),核心只有四件事:setup / teardown / isConnected / deliver。
主干里一个具体适配器都没有。 Telegram、Slack 这些住在一个长期分支 channels 上,由 /add-telegram 这类技能拷进来。主干只提供注册表和桥接(第 6 章细说)。
主机唯一盖戳的地方
src/index.ts:88 的 initChannelAdapters 回调里,主机给每条入站事件盖上 instance:
// src/index.ts:95 附近
instance: adapter.instance ?? adapter.channelType,
为什么重要: 一个平台可能同时跑多个 bot(比如同一个 Slack workspace 里三个 App)。channelType 是语义上的平台名(slack),instance 是路由用的实例名。适配器自己不需要知道它是哪个实例——主机统一盖戳,注释里管这叫 “the one host-side stamping seam”。
3. ③④路由:决定给谁、要不要理
主函数是 src/router.ts:215 的 routeInbound。它的顺序是刻意排过的,每一步都在为下一步省事。
3.1 先给模块一次「截胡」机会
// src/router.ts:220 附近
for (const intercept of messageInterceptors) {
if (await intercept(event)) return;
}
注册进来的拦截器按注册顺序跑,第一个返回 true 的吃掉这条消息,路由到此为止。用途很具体:审批流程里「请回复一个 agent 名字」这种多步对话,需要把用户下一条自由文本抓走,而不是当普通聊天路由。
3.2 一次 SQL 拿到「群 + 接了几个 agent」
// src/router.ts:241
const found = getMessagingGroupWithAgentCount(
event.channelType, event.platformId, event.instance ?? event.channelType);
合并查询是为了给最常见的情况最短的路径:一个你只是「呆在里面」的群聊,每天几百条闲聊,一次 DB 读就返回,不建行、不解析发送者、不打日志。
没有记录行时,只有在「被 @ 或私聊」(isMention)的情况下才自动建 messaging_groups 行;纯闲聊直接静默丢弃。
3.3 没有接线怎么办:升级给主人
agentCount === 0 且被 @ 了,路由会记一条 dropped_messages 审计行(理由 no_agent_wired),然后 fire-and-forget 调 channelRequestGate——权限模块注册的钩子,负责给主人发一张审批卡「有人在某某群里叫你的 bot,要不要接?」(src/router.ts:299-336)。
这里有个诚实的设计声明: 用户这条消息照样是丢掉的,审批通过后由钩子自己重放 routeInbound。
3.4 扇出:每个 agent 独立判断
一个聊天窗口可以接多个 agent。路由对每个接线独立跑一遍判断(src/router.ts:374-442),三道关:
| 关卡 | 问什么 | 谁实现 |
|---|---|---|
| engage | 这条消息够不够触发这个 agent | evaluateEngage(src/router.ts:477) |
| access gate | 这个发送者能不能访问这个 agent 组 | 权限模块注册的钩子;不装模块 = 全放行 |
| sender scope | 这条接线本身是不是只收「已知发送者」 | 权限模块的 senderScopeGate |
engage 的三种模式(src/db/schema.ts:50 的 engage_mode 列):
| 模式 | 什么时候触发 | 备注 |
|---|---|---|
pattern | 正则匹配消息文本;. 表示「永远触发」 | 正则写错时故意 fail-open(返回 true),让管理员看到 agent 在响应从而去修 |
mention | 平台确认的 @ 信号(event.message.isMention) | 不做名字文本匹配——用户 @ 的是 bot 的平台用户名,不是 agent 的显示名 |
mention-sticky | 被 @,或者这个线程已经有活跃会话 | 「会话存在」本身就是订阅状态;第一次 @ 之后跟帖不用再 @ |
3.5 没触发也可能要存:accumulate
没触发的那些 agent,还有第二条路。接线上的 ignored_message_policy 有两个值:
drop—— 直接跳过。accumulate—— 消息照样写进这个 agent 的会话,但打上trigger=0:存为上下文,不唤醒。下次真的有触发消息时,这些积累的上下文会一起被读到。
这里有个安全细节值得单独看(src/router.ts:423-431):如果是「触发了但被权限闸门拒绝」,那就不走 accumulate。理由写在注释里——accumulate 会经 writeSessionMessage 把附件落盘,而闸门拒绝的正是「这个发送者不可信」,悄悄存下他的附件恰恰是闸门要防的事。
4. ⑤落库:会话解析 + 写一行
4.1 会话三种模式
resolveSession(src/session-manager.ts:103)按 session_mode 找或建会话:
| 模式 | 一个会话 = | 典型场景 |
|---|---|---|
shared | 一个聊天群 | 群里所有线程共用一个上下文 |
per-thread | (聊 天群, 线程) | Slack/Discord 一个 thread 一个上下文 |
agent-shared | 一个 agent 组 | GitHub + Slack 折叠成同一段对话 |
线程策略是三方与运算(src/router.ts:385):接线的 threads 覆盖值(NULL = 继承)AND 渠道声明的默认值 AND 适配器的真实能力。任何一环说「不支持线程」,threadId 就被抹成 null。
4.2 命令闸门在写库之前
src/command-gate.ts:23 的 gateCommand 在消息进容器之前分类斜杠命令:
| 类别 | 命令 | 结果 |
|---|---|---|
| filtered | /start /help /login /logout /doctor /config /remote-control | 静默丢弃 |
| admin | /clear /compact /context /cost /files /upload-trace | 查 user_roles;非管理员直接写一条「Permission denied」到 outbound.db |
| 其他 | 任何别的 /xxx | 放行,交给 agent/SDK |
注意 deny 的实现:writeOutboundDirect(src/session-manager.ts:466)直接往 outbound.db 插一行拒绝消息,不唤醒容器。省一次 docker run。
4.3 写消息 = 开库、插一行、关库
// src/session-manager.ts:210 writeSessionMessage 的骨架
const db = openInboundDb(agentGroupId, sessionId);
try { insertMessage(db, { ... }); } finally { db.close(); }
每次调用都开一次关一次。 函数注释上有一行醒目的 ⚠ 警告不许改成长连接——原因在第 2 章。
附件在这一步被从 base64 抽出来落盘(extractAttachmentFiles,src/session-manager.ts:333),content 里换成 inbox/<messageId>/<filename> 路径。这个函数有四层防护(basename 检查、拒绝预置符号链接、realpath 包含性检查、wx 独占创建),第 3 章细说。
4.4 消息 id 要按 agent 命名空间
// src/router.ts:655(messageIdForAgent,签名在 :653)
return `${id}:${agentGroupId}`;
扇出时同一条平台消息会落进多个 agent 的会话库,而 messages_in.id 是主键。加 agent 组后缀就不会撞。
5. ⑥唤醒容器
// src/router.ts:636 附近
const freshSession = getSession(session.id);
if (freshSession) {
const woke = await wakeContainer(freshSession);
if (!woke) stopTypingRefresh(freshSession.id);
}
wakeContainer(src/container-runner.ts:134)有一条明确的契约:永不抛异常。返回 true 表示起来了,false 表示这次拉起失败(比如 OneCLI 网关不可达)。失败时那条 inbound 行保持 pending,60 秒后 sweep 会重试。
它还做了并发去重:
// src/container-runner.ts:100 附近
const existing = wakePromises.get(session.id);
if (existing) return existing; // 合流到同一个 promise
为什么需要:spawnContainer 里有一段异步的准备工作(算挂载、调 OneCLI 网关)。只查 activeContainers.has() 的话,这段窗口期内第二次唤醒会通过检查、起第二个容器,导致同一个会话目录被两个容器写、用户收到两 条重复回复。
容器怎么起来的、挂了什么,是第 3 章的内容。
6. ⑦投递:把回复发回去
6.1 两个轮询,一个互斥集合
src/delivery.ts 开了两条轮询链:
| 轮询 | 周期 | 覆盖 |
|---|---|---|
pollActive | 1 秒(ACTIVE_POLL_MS,src/delivery.ts:32) | container_status='running' 的会话 |
pollSweep | 60 秒(SWEEP_POLL_MS,src/delivery.ts:33) | 所有 status='active' 的会话 |
两条链最终都调同一个入口 deliverSessionMessages(src/delivery.ts:163),而一个运行中的会话同时在两个集合里。所以入口第一件事是查 inflightDeliveries(src/delivery.ts:52):同一会话正在排空时,第二个调用直接跳过,真正的排空由 drainSession(:189)做。
注释里把「为什么跳过而不是排队」讲得很清楚:漏掉的行下一个 tick(约 1 秒)就会被捡起来;而并发投递意味着用户已经看到两遍了,DB 层的 INSERT OR IGNORE 幂等救不回来。
6.2 已投递标记写在 inbound.db
这是双库设计的直接后果:主机不能写 outbound.db,所以「这条发过了」记在自己拥有的 inbound.db 的 delivered 表里(src/mailbox/sqlite/session-db.ts:258 markDelivered)。
投递循环因此是:读 outbound 全部到期行 → 减去 inbound 的 delivered 集合(getDeliveredIds,src/mailbox/sqlite/session-db.ts:250)→ 逐条发。
6.3 出站消息的四种去向
deliverMessage(src/delivery.ts:283)按 kind / channel_type 分流:
messages_out 一行
│
┌──────────────┼──────────────┬──────────────┐
▼ ▼ ▼ ▼
kind=system kind=task_log channel=agent 其他
(系统动作) (任务运行日志) (agent→agent) (发到平台)
│ │ │ │
注册表分发 追加到 路由到目标 权限检查
(ncl 请求、 tasks/<id>.md agent 的会话 → 适配器
装包审批) 永不投递 deliver()
6.4 出站权限:两条通过路径
src/delivery.ts:347-410 的检查逻辑:
- 目标就是会话自己的来源聊天 → 永远放行(注释说「要求给显而易见的情况配 ACL 是个 footgun」)。
- 否则必须在
agent_destinations表里有一行明确授权。
失败时抛异常而不是静默 return——注释点明了这是修过的 bug:静默 return 会让消息被标记成「已投递」,而实际上什么都没发出去。抛异常则落进重试路径,3 次后标记 failed。
6.5 重试与放弃
| 第几次失败 | 日志级别 | 这条消息的下场 |
|---|---|---|
| 第 1 次 | warn | 留着,下个 tick 再试 |
| 第 2 次 | warn | 留着,下个 tick 再试 |
| 第 3 次 | error | markDeliveryFailed,不再重试 |
MAX_DELIVERY_ATTEMPTS = 3(src/delivery.ts:34)。计数在内存 Map 里,进程重启就清零——注释说这是故意的,给失败消息一次新机会。
7. 每一步失败了会怎样(兜底表)
这张表是本章最该带走的东西:这个系统几乎每一步都有一条「什么都不做也能自愈」的路径。
| 环节 | 失败形态 | 兜底 |
|---|---|---|
| 适配器收消息 | 抛异常 | src/index.ts:106 的 .catch 记日志,不影响别的消息 |
| 没有接线 | 消息被丢 | 记 dropped_messages 审计行 + 给主人发审批卡 |
| engage 正则写错 | 正则编译失败 | fail-open,agent 照常响应,管理员能看到并去修 |
| 写 inbound.db 时目录被人删了 | 打不开库 | writeSessionMessage 检测到缺文件就重建目录+双库 |
| 容器拉起失败 | OneCLI 不可达等 | wakeContainer 返回 false,消息保持 pending,60 秒后 sweep 重试 |
| 容器中途崩了 | processing 认领残留 | sweep 重置为 pending + 指数退避,5 次后标记 failed |
| 容器卡死但活着 | 心跳文件超 30 分钟没更新 | sweep 杀容器并重置消息 |
| 投递失败 | 适配器抛异常 | 3 次重试后 markDeliveryFailed |
| 主机自己崩溃循环 | 反复启动 | 熔断器按 0/0/10/30/120/300/900 秒退避(BACKOFF_SCHEDULE_S,src/circuit-breaker.ts:11) |
8. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 入站路由主函数 | src/router.ts | routeInbound |
| 触发判定 | src/router.ts | evaluateEngage |
| 扇出到单个 agent | src/router.ts | deliverToAgent |
| 消息 id 命名空间化 | src/router.ts | messageIdForAgent |
| 会话解析(三种模式) | src/session-manager.ts | resolveSession |
| 写入站消息 | src/session-manager.ts | writeSessionMessage |
| 免唤醒的直写出站 | src/session-manager.ts | writeOutboundDirect |
| 斜杠命令闸门 | src/command-gate.ts | gateCommand |
| 容器唤醒(去重、不抛) | src/container-runner.ts | wakeContainer |
| 投递入口(互斥守卫) | src/delivery.ts | deliverSessionMessages |
| 投递排空 | src/delivery.ts | drainSession / deliverMessage |
| 已投递标记 | src/mailbox/sqlite/session-db.ts | markDelivered / getDeliveredIds |
| 线程策略解析 | src/channels/channel-defaults.ts | resolveThreadPolicy |
| 适配器接口 | src/channels/adapter.ts | ChannelAdapter / InboundEvent |