技术3/26/2026·8 min

用 IM 标准范式重设计多 Agent 协作的 Channel 消息系统

AgentsMesh 把 channel 从全局单 channel store 改成 per-channel 缓存 + IM 标准滚动,并把 @mention 接成一条直达 pod PTY 的 prompt 转发链。

问题

AgentsMesh 把 agent 跑成 pod。Channel 是人和 pod 对话的地方:人打字、pod 回复,人可以用 @ 提及把某个 pod 拉进对话。第一版有两个问题。

消息 UI 围绕一个绑定「当前 channel」的全局 store 搭建。切 channel 会丢掉滚动位置,新消息计数会落到错误的 channel 上,WebSocket 事件按 currentChannel 路由而不是按事件本身所属的 channel。store 还无上限地增长——任何打开过的 channel 都常驻内存。

@mention 这条路会把 prompt 转发给 pod,但 prompt 经常提交不了。文本落进了 agent 的终端,停在输入行不动。

两件事在同一段工作里修掉:基于 IM 标准范式的 channel 重设计(PR #198),以及一条最终落到真实回车键的转发链(PR #401)。

Per-channel 缓存,而非全局 store

store 重写是结构性的改动。不再是一个可变的「当前 channel」整块,缓存改为按 channel id 索引:

cache: Record<number, ChannelMessageCache>

每个条目带自己的 messageshasMoreloadingerror。切 channel 是读另一个 key,不会拆掉再重建那个唯一的全局槽位。WebSocket 事件按 channelId 路由,所以 channel B 上来一条消息就更新 B 的缓存,不管 B 是否在屏幕上。

无界增长用 LRU 淘汰兜住,上限是个常量:

const MAX_CACHED_CHANNELS = 20;

写入第 21 个 channel 时,超出上限的最旧条目被丢弃。这个 trade-off 是有意的:内存里保留 20 个 channel 的历史,重新打开一个久未活动的 channel 代价是一次拉取。

后端这边,「还有没有更多历史」不再靠猜。列表 API 取 limit+1 行,回报 has_more

messages, err := s.repo.GetMessages(ctx, channelID, before, limit+1)
hasMore := len(messages) > limit
if hasMore {
    messages = messages[:limit]
}
slices.Reverse(messages)
return messages, hasMore, nil

多取的那 1 行永不返回客户端,它只回答「加载更早」这个入口该不该存在。MCP 的 getMessages 加了每次最多 100 的硬上限,客户端无法请求一个无界的分页。

IM 标准滚动

滚动行为是用户真正能感知的部分。useMessageListScroll 处理三个场景:

  • 吸底。 如果你在底部、来了新消息,跟着滚到底。
  • 加载更早时保持位置。 往前插历史不能跳动视口。
  • Jump to Latest。 如果你已经往上翻、又来了消息,弹一个带未读计数的浮动按钮,而不是把你硬拽回底部。

「在底部」是一个阈值,不是相等判断:

const atBottom = el.scrollHeight - el.scrollTop - el.clientHeight < 50;

这 50px 余量吸收子像素布局,避免按钮在边界处反复闪烁。新消息计数在 scroll handler 里复位、而非在 effect 里——在 effect 里对 setState 做反应正是 React compiler 会报的问题,而 scroll handler 本就握着这个信号。

@mention → pod prompt

转发链是 IM 语义和真实进程交汇的地方。mention 不是字符串匹配。消息正文是结构化内容,extractMentions 遍历 block 树、收集带类型的 mention,并在过程中去重:

case channel.EntityPod:
    if !podsSeen[el.EntityKey] {
        podsSeen[el.EntityKey] = true
        m.Pods = append(m.Pods, el.EntityKey)
    }

结果以 jsonb 持久化在消息上(Mentions MessageMentions,含 PodsUsersChannel)。前端把 pod 的 mention 文本取成 pod key 的前 8 个字符(pod_key.slice(0, 8)),后端镜像这个完全相同的长度:

const podMentionTextLen = 8

发一条消息会按注册顺序跑一串 post-send hook:校验 mention → 发布事件 → 通知 → 转发给 pod。校验器会把在该 org 里解析不到的 pod key 丢掉,转发根本看不到它们。转发 hook 随后构造 prompt:

func buildPodPrompt(content, channelName string, channelID int64, podKeys []string) string {
    rawPrompt := stripPodMentions(content, podKeys)
    rawPrompt = ptyPromptFlattener.Replace(rawPrompt)
    return fmt.Sprintf("Message from channel(#%s, channel_id=%d): %s. If you finish it, please reply to this channel using send_channel_message(channel_id=%d).", channelName, channelID, rawPrompt, channelID)
}

这里发生三件事。@abcd1234 这个 token 被剥掉,agent 不会看到自己的 handle。正文被压平——\r\n\n\r 都变成一个空格。模板告诉 agent 回到哪里答复,于是 pod 的回答经 send_channel_message 工具落回同一个 channel。提及自己的 pod 会被跳过(SenderPod == podKey),agent 不会在自己的消息上打转。

prompt 为什么提交不了

最初的提交先写 prompt 文本,再用第二次调用写 \r,期望终端 UI 把回车读成 Enter。它没有。两次写在微秒级先后打到 PTY,TUI 的 read(2) 循环把它们折叠进一个 chunk,末尾的 \r 被当成一次粘贴的尾巴吃掉——而不是提交键。

MCP 那条路能成、这条不行,差别在时序。MCP 把文本和按键发成两个独立 RPC,网络往返天然给了 TUI 的 read 循环一次间隔。进程内的 gRPC 控制面没有这个间隔。

修法是把这个间隔显式重建出来,并用按键 API 取代裸字节:

const ptySubmitGap = 80 * time.Millisecond

if err := pod.IO.SendInput(cmd.Prompt); err != nil {
    return err
}
if ta, ok := pod.IO.(TerminalAccess); ok {
    time.Sleep(ptySubmitGap)
    return ta.SendKeys([]string{"enter"})
}

SendInput 是「裸字节」,SendKeys(["enter"]) 是「按一个键」。80ms 间隔后,TUI 看到的是一次只载着 Enter 的 read(2),把它当成真实按键。buildPodPrompt 里的换行压平是另一半:正文里嵌的 \n 会以同样方式预提交半个 prompt、或把最后那个 Enter 吞掉。一个测试钉住这个不变量,正文再也长不出换行:

func TestBuildPodPrompt_NeverContainsNewlines(t *testing.T) {
    // "line1\nline2", "line1\r\nline2", "trailing\n" ...
    if strings.ContainsAny(got, "\n\r") {
        t.Fatalf("buildPodPrompt must not emit \\n or \\r (breaks PTY Enter submit)")
    }
}

agent 数与成员数要分开计

一个相邻的正确性 bug 值得点名。一个 channel 有两张支撑表——channel_members 存用户、channel_pods 存 agent——但成员计数只数了用户。加入一个 agent 会显示「Members · 0」和空的右栏。修法加了一个独立的 AgentCount,由单独的 GetChannelPodCount 填充,于是用户数和 agent 数各自从自己的表算,而不是用一个去顶替另一个(PR #406)。

可迁移的结论

几条约束超出这个代码库本身。

把消息缓存按会话 id 索引,而不是按一个可变的「当前」指针,并让实时事件按同一个 id 路由。一旦有两个会话同时活着,全局-当前这个范式就崩。

当一次写和它的提交按键共享同一条进程内通道时,它们会合并。IM 和终端 UI 都按 chunk 读输入;提交必须落在它自己的一次 read 里。把往返间隔有意复现出来,并优先用「按一个键」的 API,而不是吐一个裸控制字节、指望读取方替你切分。

转发结构化的意图,不是裸文本。mention 是带类型的(EntityPod + key)、对 org 校验过、从正文剥掉、压平成单行、再包上明确的回复路径。每一步都拿掉一种「agent 收到却没法处理」的可能。