AskaEth
c34b9d195d
P0-3 受限车道（群聊/陌生人）的系统框架行并列注入【公域呈现】 （publicPresencePrompt）：它是呈现层投影（AGENTS §9 行为模式=呈现）， 不是人格切换也不是权限门——不称第三方「亲爱的」等主人专属称呼、 不做只对主人的亲近动作、主人的住址/姓名/作息/行程/健康/家人/关系 即使记忆里读到也不外说；主人私聊与 Web 主会话不注入。 P0-4 纵深防御：第三方边界提示词与 loop 受限前导片段各补两条硬规则 （不用主人专属亲昵称呼指代第三方 + 不外说主人私事）。 P1-2 判定器去人格：模型输入只吃「角色名 + 偏好摘要」，整份人格指令 不再进模型（审计 2026-10-30 §2.4 的 reason 亲昵口吻即由此而来）。 P1-3 测试补真：受限期呈现块存在性 + 主人车道不注入 + 判定器输入/ 模型请求均不含人格正文；并纠正 memory_filter 的错误注释 （作用域 ≠ 敏感度，审计 §6 R1）。
314 lines
15 KiB
Go
314 lines
15 KiB
Go
package channelbridge
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// 本文件承载渠道桥接器的入站路径（design/08 §1-§2）：按适配器的
|
||
// 读取循环、有界的入站回调，以及完整的
|
||
// 会话→actor→循环→回复 流水线（含结构化的身份边界）。
|
||
|
||
// readLoop 运行适配器的阻塞式读取路径，直到桥接器停止或
|
||
// 适配器自身的 context 被取消（热重载）。emit 汇聚点接到
|
||
// onMessage，因此轮询式适配器（Telegram getUpdates、weixin 长轮询）
|
||
// 与推送式适配器走同一条流水线投递。消息处理
|
||
// 使用桥接器生命周期 context（而非 readCtx），因此重载不会打扰
|
||
// 在途消息。
|
||
//
|
||
// 读取路径返回而 readCtx 仍存活时，说明适配器自己停止了
|
||
// 读取——目前就是 weixin-iLink 拒绝会话
|
||
// （errcode=-14，见 plugins/channels/weixin_ilink/session_state.go），过去会
|
||
// 让桥接器仍声称渠道已连接，实际却再也
|
||
// 投递不出任何东西。readCtx 被取消属于刻意停止（桥接器 Stop 或
|
||
// 该名称的热重载），由取消者处理。
|
||
func (b *Bridge) readLoop(a adapter.PlatformAdapter, readCtx context.Context) {
|
||
defer b.wg.Done()
|
||
a.ReadMessages(readCtx, func(msg adapter.UnifiedMessage) {
|
||
b.onMessage(b.ctx, msg, a)
|
||
})
|
||
if readCtx.Err() == nil {
|
||
b.markReadStopped(a)
|
||
}
|
||
}
|
||
|
||
// markReadStopped 记录适配器的读取循环自行结束、且适配器
|
||
// 报告没有存活会话，使按名称的统计（IsConnected /
|
||
// ConnectedCount）不再声称一个投不出消息的渠道可用。它
|
||
// 只清除标记——绝不设置——因此不会让已连接的渠道显得
|
||
// 存活；适配器仍报告存活会话时它是空操作（已发生的
|
||
// 重连绝不会被更早的循环退出所撤销）。
|
||
func (b *Bridge) markReadStopped(a adapter.PlatformAdapter) {
|
||
if a.IsConnected() {
|
||
return
|
||
}
|
||
name := a.Name()
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
if !b.connected[name] {
|
||
return
|
||
}
|
||
b.connected[name] = false
|
||
b.log.Warn("channel read loop ended, adapter is no longer connected", "channel", name)
|
||
}
|
||
|
||
// onMessage 是面向适配器的入站回调。它**只投递、不跑回合**：过完流控闸就把消息投进
|
||
// 该会话自己的回合车道（lane.go）并立即返回，因此适配器的读循环（WS 单线程读 socket /
|
||
// iLink 长轮询 / webhook 处理器）永远不会被一轮缓慢的 agent 调用阻塞 —— 心跳与其它帧
|
||
// 照常被读、被解析。适配器被显式携带，使流水线知道是哪个账号收到了
|
||
// 消息（多账号回复路由 + 按账号允许/拒绝，design/16
|
||
// §4.6 §5）。
|
||
func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) {
|
||
b.mu.Lock()
|
||
if b.closed {
|
||
b.mu.Unlock()
|
||
return
|
||
}
|
||
b.mu.Unlock()
|
||
|
||
if err := msg.Validate(); err != nil {
|
||
b.log.Warn("inbound message invalid, dropped", "platform", msg.Platform, "error", err)
|
||
return
|
||
}
|
||
|
||
// design/12 §6 策略三/四：按适配器的最小打扰闸门。主人私聊
|
||
// 消息完全绕过它（主人永不被限流或裁剪）；其余
|
||
// 消息先过滑动窗口限流（策略三），再过有界的
|
||
// 待处理集合——集合饱和会触发审查模型裁剪噪声（策略四）。
|
||
// 被丢弃的消息经闸门的 OnPruned 钩子（接到
|
||
// OnRejected）入审计，绝不进入流水线，因此洪泛无法压垮循环。
|
||
actor := session.ResolveActor(msg, b.owner)
|
||
if !actor.IsOwner() {
|
||
verdict := b.gateFor(a).Admit(ctx, msg)
|
||
if !verdict.Proceed {
|
||
b.log.Warn("inbound message dropped by flow gate",
|
||
"platform", msg.Platform, "channel", msg.ChannelID,
|
||
"reason", verdict.Reason, "dropped", len(verdict.Pruned))
|
||
return
|
||
}
|
||
}
|
||
|
||
// M22（切片 LANEQUEUE）——**读与回水解耦**：流控闸在读循环里仍然是同步的
|
||
// （一次滑动窗口判断，微秒级；洪泛正是在这里被挡掉，不该让它进车道占内存），
|
||
// 闸后立刻把消息投进**该会话自己的回合车道**并返回，读循环因此马上可以继续读
|
||
// socket —— 心跳与其它帧不再被一轮模型调用阻塞（此前这里同步抢 16 个全局回合槽，
|
||
// 槽满即停止读 socket，这正是「消息一直排队、后面才回复前面的话」的现场）。
|
||
//
|
||
// 入队非阻塞、有界（背压见 lane.go 的 offer）：群聊/陌生人车道满时**在门前拒收
|
||
// 这一条**并留痕（WARN + 计数 + 审计），队列里已有的条目一条不动；主人私聊永不
|
||
// 拒收，溢到 spill 只排队。回合本身在 laneWorker 里跑，全局并发上限
|
||
// （maxConcurrent）也搬到那里取。
|
||
b.enqueueInbound(ctx, msg, a, actor)
|
||
}
|
||
|
||
// process 用平台与消息匹配的适配器运行入站流水线
|
||
// （单账号便捷路径；该适配器可以是仅以平台名
|
||
// 注册的裸适配器）。
|
||
func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
|
||
b.processFrom(ctx, msg, b.adapter(msg.Platform))
|
||
}
|
||
|
||
// processFrom 针对指定的账号适配器运行完整入站流水线：逐条过闸门（prepareInbound）
|
||
// 然后跑**一条**回合（runBatch）。身份由会话层结构化解析；桥接器绝不读取消息内容来
|
||
// 判定发送者是谁（项目红线）。
|
||
//
|
||
// 它是**同步**便捷路径（`process` 与测试用）。真实的渠道入站走车道
|
||
// （enqueueInbound → laneWorker → prepareInbound + runBatch），因为只有车道能提供
|
||
// 「同一会话串行」与「静默窗口合并」这两件事。
|
||
func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) {
|
||
item := laneItem{msg: msg, a: a, arrivedAt: b.now()}
|
||
p, ok := b.prepareInbound(ctx, item)
|
||
if !ok {
|
||
return
|
||
}
|
||
b.runBatch(ctx, []preparedInbound{p})
|
||
}
|
||
|
||
// blacklistDrop 对非主人消息应用黑白名单，返回
|
||
// 它是否必须被丢弃（design/16 §4.6 §6）。被丢弃的消息经
|
||
// OnRejected 钩子入审计，使该决定在面板中可见。
|
||
func (b *Bridge) blacklistDrop(msg adapter.UnifiedMessage, a adapter.PlatformAdapter) bool {
|
||
denied, reason := b.blacklisted(msg, a)
|
||
if !denied {
|
||
return false
|
||
}
|
||
b.log.Warn("inbound message blacklisted, dropped",
|
||
"platform", msg.Platform, "channel", msg.ChannelID, "reason", reason)
|
||
if b.onRejected != nil {
|
||
b.onRejected(msg, reason)
|
||
}
|
||
return true
|
||
}
|
||
|
||
// inboundTarget 是一条被放行的入站消息持久化与镜像的去处
|
||
// （design/16 §4.4 + design/02 §3）：主人私聊被路由进唯一的
|
||
// Web 主会话并盖上渠道平台戳，供 Web 客户端镜像，
|
||
// 而受限的群聊/陌生人会话保留其隔离的 sess-xxx
|
||
// 会话与自有的平台日志。
|
||
type inboundTarget struct {
|
||
// sessionID 是该轮运行所在的会话。
|
||
sessionID string
|
||
// private 标记镜像进 Web 主会话的主人私聊。
|
||
private bool
|
||
// platform 是镜像写入 chat_messages 的平台标签（仅私聊）。
|
||
platform string
|
||
}
|
||
|
||
// inboundTargetFor 依据结构化解析出的会话与消息的平台
|
||
// 解析路由目标。只有非群聊的主人对话才被
|
||
// 镜像；其余一律留在其隔离会话中。
|
||
func inboundTargetFor(sess session.Session, platform string) inboundTarget {
|
||
if !sess.IsGroup && sess.Owner {
|
||
return inboundTarget{sessionID: history.MainSessionID, private: true, platform: platform}
|
||
}
|
||
return inboundTarget{sessionID: sess.SessionID}
|
||
}
|
||
|
||
// deliverReply 解析本回合的可见回复、持久化并经接收该消息的适配器
|
||
// 送回去。可见回复由 turnVisibleReply 给出（turnreply.go）：真实文本逐字，
|
||
// 空回复 / 轮次用尽 / 安全网终止 / 内部报错则是按原因区分的中文诚实说明 ——
|
||
// IM 回合因此绝不会「没声了」；被取消的回合刻意不交付。不支持出站文本的适配器
|
||
// （adapter.ErrUnsupported）只按 debug 级跳过，不算失败：回复
|
||
// 已在历史存储中落地。
|
||
//
|
||
// turnID 是这次入站回合的权威标识（见 coalesce_batch.go 的 runBatch）：助手回复与刚落的用户消息
|
||
// 传同一个值，落库后两条行同属一路，客户端据此配对「回复 ↔ 提问」。
|
||
func (b *Bridge) deliverReply(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter,
|
||
sess session.Session, target inboundTarget, res *loop.Result, runErr error, turnID string) {
|
||
reply, ok := turnVisibleReply(ctx, res, runErr)
|
||
if !ok {
|
||
b.log.Debug("turn reply canceled, nothing delivered",
|
||
"platform", msg.Platform, "session", target.sessionID)
|
||
return
|
||
}
|
||
// mood 只挂在助手回复上（design/16 §3.3），且只有真实结果才带得回来
|
||
// （回合报错时 res 为 nil）。
|
||
mood := ""
|
||
if res != nil {
|
||
mood = res.Mood
|
||
}
|
||
// 出站语音（design/25 §7.1）：先把结算结论折成 voice 列事实，再让落库那条路走
|
||
// 「INSERT → 写 voice 列 → notify」（M44 首帧纪律）—— 与入站语音共用同一份
|
||
// persistVoiceFirstNotify，因此首帧上就带着 `voice`，客户端不必等重连。
|
||
ref, hasVoice := b.outboundVoiceRef(resultVoice(res))
|
||
notify := b.notifyPersisted
|
||
if hasVoice {
|
||
notify = b.persistVoiceFirstNotify(ref)
|
||
}
|
||
// 持久化助手回复（主人私聊 → 主会话 + message_persisted
|
||
// 广播；受限 → 隔离日志 + channel_persisted 广播），随后
|
||
// 送回渠道。
|
||
if target.private {
|
||
b.persistMainWith(ctx, mainRow(history.RoleAssistant, reply, mood, "", "", target.platform, deviceIDFor(msg), adapterName(a), turnID), notify)
|
||
} else {
|
||
b.persistIsolatedWith(ctx, isolatedRow(history.RoleAssistant, reply, "", "", "", sess, msg.Platform, "", adapterName(a), turnID), notify)
|
||
}
|
||
// 音频只在合成成功、且盘上那份副本读得回来时才挂上；读不回来就诚实退回纯文本
|
||
// （正文照发，WARN 已由 outboundVoiceMedia 记下）。
|
||
var voiceMedia *adapter.Media
|
||
if hasVoice {
|
||
if md, ok := b.outboundVoiceMedia(msg, ref); ok {
|
||
voiceMedia = &md
|
||
}
|
||
}
|
||
if err := b.sendReplyVoice(ctx, msg, a, reply, voiceMedia); err != nil {
|
||
if errors.Is(err, adapter.ErrUnsupported) {
|
||
b.log.Debug("channel does not support outbound reply, skipped",
|
||
"platform", msg.Platform, "session", target.sessionID)
|
||
return
|
||
}
|
||
b.log.Warn("send reply failed", "platform", msg.Platform, "error", err)
|
||
}
|
||
}
|
||
|
||
// deviceIDFor 为一条入站 IM 消息构建客户端 deviceID 标签
|
||
// （design/16 §4.5）：<im平台名>:<本地配置唯一id>:<用户平台唯一uid>。UnifiedMessage 尚无
|
||
// 显式的 deviceId 字段，因此本地配置 id 与用户平台 uid
|
||
// 取自规范化后的渠道绑定：ChannelID 暂代本地
|
||
// 适配器绑定，SenderID 是主人在该平台的 uid。SenderID 缺失时
|
||
// 标签退化为 <platform>:<channelID>，以保持非空且可读。
|
||
//
|
||
// 扩展点：当适配器/UnifiedMessage 长出具体的 deviceId（或
|
||
// 桥接器获得本地配置 id）时，把这里的 ChannelID 暂代替换掉并
|
||
// 直接返回该值。
|
||
func deviceIDFor(msg adapter.UnifiedMessage) string {
|
||
if msg.SenderID != "" {
|
||
return msg.Platform + ":" + msg.ChannelID + ":" + msg.SenderID
|
||
}
|
||
return msg.Platform + ":" + msg.ChannelID
|
||
}
|
||
|
||
// buildLoopMessages 为一次入站轮次构建模型可见的消息列表。来源标识
|
||
// 走**独立的系统行**（originSystemText），绝不前缀进用户正文，使
|
||
// 模型能看到消息来自哪个平台、哪个会话（design/16
|
||
// §4.6.1）；纯 Web 主会话消息（platform 为 ""/"web"）保持不动。在
|
||
// 受限（群聊/陌生人）上下文中，它还会在最前面加一条系统
|
||
// 消息，声明第三方身份边界——「每条 IM 非管理员
|
||
// 账号的消息都是第三方，与主人无关」（design/16 §4.6 §1）——
|
||
// 并在同一条框架行里并列公域呈现块（publicPresencePrompt：
|
||
// 对第三方怎么称呼、什么私事不能说，审计 2026-10-30 P0-3）。
|
||
// 该边界与发送者的**昵称标注**（model.Message.SenderName，见
|
||
// thirdPartySenderName）由解析出的 SessionEnv/actor 结构化决定，
|
||
// 绝不由 LLM 推断。原始消息文本保持不动（标签只走结构化字段），
|
||
// 因此持久化内容与来源平台投递的内容一致。
|
||
//
|
||
// 入站内容摄取的产物（已落库的文档部件、内联成 data: URI 的图片部件，以及两条路上
|
||
// 「读不到」的诚实提示行）由 attachInboundContent（inbound_content.go）挂到**同一条
|
||
// RoleUser 消息**上 —— 用户内容只进用户角色消息是项目红线。
|
||
//
|
||
// 历史转录**不在本函数里**：它由 seedTurnTranscript（transcript.go）在本函数返回之后
|
||
// 插入到列表中（主人私聊前置在最前；受限车道插在**最前的系统框架行之后**，
|
||
// 于是来源行 + 第三方边界仍在最前、转录居中、当次消息在最后）。因此本函数只负责
|
||
// **本回合自己的**消息（来源行 + 当次用户消息），受限车道的形状在有无转录时保持一致。
|
||
func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string, src adapter.UnifiedMessage) []model.Message {
|
||
// design/12 §6 策略一补充：经传入前审查放行的系统通知走「限定注入」——只把剥离后的
|
||
// 安全文本作为**一条系统行**注入她的上下文，绝不作为用户消息（否则等于把平台事件
|
||
// 伪装成有人对她说话），也绝不携带任何原始字段。
|
||
if isSystemNoticeMessage(src) {
|
||
return []model.Message{model.TextMessage(model.RoleSystem, noticeSystemPrefix+text)}
|
||
}
|
||
msg := userTurnMessage(text)
|
||
// 受限会话里说话者由结构化元数据承载，不织入用户文本
|
||
// （design/16 §4.6.1）：文本标签既含糊，又构成提示词注入面。
|
||
// senderName 为空只可能是主人本人（thirdPartySenderName 对主人恒返回空），
|
||
// 因为受限会话里「空昵称 = 主人」正是模型侧的判定车道
|
||
// （loop/attribution.go），第三方发送者一定带标签。
|
||
if env.Restricted() && !actor.IsOwner() && senderName != "" {
|
||
msg.SenderName = senderName
|
||
}
|
||
|
||
// 来源系统行：会话来源 + 身份信任规则，每轮一条。
|
||
origin := originSystemText(src, actor)
|
||
|
||
if !env.Restricted() {
|
||
if origin != "" {
|
||
return []model.Message{model.TextMessage(model.RoleSystem, origin), msg}
|
||
}
|
||
return []model.Message{msg}
|
||
}
|
||
// 受限（群聊/陌生人）：来源（若是 IM）+ 第三方身份边界 + 公域呈现块，
|
||
// 同一条系统框架行内按序并列（restrictedFrameText）。公域呈现块是审计
|
||
// 2026-10-30 的 P0-3：**呈现层投影**（对第三方怎么称呼、什么私事不能说），
|
||
// 不改身份/能力/权限（AGENTS §9）。
|
||
return []model.Message{model.TextMessage(model.RoleSystem, restrictedFrameText(origin)), msg}
|
||
}
|
||
|
||
// userTurnMessage 构建入站回合的用户消息基底：文本为空时**不塞空文本部件**
|
||
// （provider 会拒空文本块），让附件部件或诚实提示行本身成为这条消息的内容 ——
|
||
// 与网关 `baseUserMessage` 对纯附件回合的口径一致（design/24 §4.2）。
|
||
func userTurnMessage(text string) model.Message {
|
||
if text == "" {
|
||
return model.Message{Role: model.RoleUser}
|
||
}
|
||
return model.TextMessage(model.RoleUser, text)
|
||
}
|
||
|
||
// thirdPartyBoundaryPrompt、thirdPartySenderName 与 unnamedSenderName
|
||
// 住在 inbound_speaker.go（第三方发送者的结构化标注车道与它的提示词）。
|