package gateway import ( "context" "encoding/json" "fmt" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/turnreply" "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" ) // wsProactiveMessage 是自主主动推送的 WS 事件类型 // (design/12 §3):Cyrene 在没有近期用户输入时决定发送的消息。 const wsProactiveMessage = "proactive_message" // proactivePayload 是主动消息消毒后的线上结构。它只 // 携带客户端渲染气泡所需的字段(content、platform、channel、 // 发送者标签);它刻意省略 RawData / 媒体字节 / token,使 // 主动推送绝不可能把凭据或载荷泄露进前端。 type proactivePayload struct { Platform string `json:"platform,omitempty"` ChannelID string `json:"channelId,omitempty"` ChannelType string `json:"channelType,omitempty"` Content string `json:"content"` ContentType string `json:"contentType,omitempty"` Sender string `json:"sender,omitempty"` Timestamp time.Time `json:"timestamp,omitempty"` MessageID string `json:"messageId,omitempty"` // ID 是这条主动消息在**主会话历史**里的权威行 id(切片 P0「气泡恒落」)。 // 客户端据此把实时气泡对齐到落库行:与握手回放 / REST 历史拉取按 id 去重 // (同一条绝不渲染两个气泡),刷新后仍在。落库失败时该键**省略** —— // 客户端因此知道这条只有实时帧、没有历史行,而不是拿到一个悬空的 id。 ID string `json:"id,omitempty"` // CreatedAt 是落库行的时间戳(Unix 毫秒),与 ID 同时出现/同时省略。它是 // 权威的排序时间(与 `timestamp` 是同一次推送的两个视角:前者是历史行的 // 时间,后者是这条统一消息自身的时间)。 CreatedAt int64 `json:"createdAt,omitempty"` } // SendProactive 把 msg 投递给每个存活的 owner/admin 连接,并返回投递与落库 // 两件事的结果(切片 P0,`backthink.ProactiveDelivery`)。 // // 顺序是刻意的 —— **先规整、再落库、再发帧**:主动消息必须成为主会话 // (`history.MainSessionID`)里的一条真实消息(assistant 角色),否则刷新/ // 重载后什么都补不回来(这正是「气泡恒落」的根);而帧要带上落库行的 id, // 客户端才能把实时气泡对齐到历史行。规整必须落在落库**之前** —— 落库行是 // 权威文本,先落库后规整只会在库里留下脏的、而帧与历史行逐字对不上。 // // 文本与其它三处**同源**(Web 会话回合 / IM 出站 / 主动推送 IM 侧):一律走 // `turnreply.NormalizeParagraphs`,本文件绝不另写一份实现(本仓已因「同一条 // 规则两份写法」栽过多次)。修前的真缺口正是这条路由**完全绕过**它(M19 复审 // 抓到):同一条主动消息走 IM 是规整过的、走 Web 实时推还是原样,半包 `

` 在 // 客户端「只要存在完整 `

` 块就不再按空行回退切分」的契约下整条塌成一坨。 // 规整是幂等的写入侧加法(已规范的文本逐字不变),因此 `

` = 一段一气泡、 // `
` = 气泡内分隔、``/`` 随所在段落进同一个 `

` 这些 // 既有语义一条不动。 // // 落库失败**不静默**:记一条带来源的 WARN,并把错误返回给调用方,同时 // Persisted=false —— 帧照发(实时气泡不能丢),但记录里绝不说「已推送」。 // // 计数为零意味着没有 owner/admin 客户端在线;消息此时仍会作为一条真实记录 // 留在主会话里(下次打开就能看到),调用方据此回退到 IM bridge 把它**投出去**。 // 只有 owner/admin 连接会被作为目标(红线:主动推送仅 owner);viewer 或未认证 // 连接绝不被触达。它绝不阻塞:发送缓冲满时对该慢客户端丢弃该帧, // 而不是卡住后台思考推送。 func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (backthink.ProactiveDelivery, error) { if err := ctx.Err(); err != nil { return backthink.ProactiveDelivery{}, err } // 先规整、再落库:msg 是调用方传来的值副本,就地改写只影响本路由; // 落库与推帧随后取的都是**同一个**字符串,两端因此逐字相同。 msg.Content = turnreply.NormalizeParagraphs(msg.Content) persisted, persistErr := h.persistProactiveToMain(ctx, msg) if persistErr != nil { h.log.Warn("persist proactive message to main session failed", "err", persistErr, "platform", msg.Platform) } b, err := json.Marshal(wsEvent{Type: wsProactiveMessage, Data: proactivePayloadFrom(msg, persisted)}) if err != nil { return backthink.ProactiveDelivery{Persisted: persisted.ID != ""}, fmt.Errorf("gateway: marshal proactive: %w", err) } h.mu.RLock() defer h.mu.RUnlock() n := 0 for c := range h.conns { if !ownerAdminClaims(c.claims) { continue } select { case c.send <- b: n++ default: } } return backthink.ProactiveDelivery{Connections: n, Persisted: persisted.ID != ""}, persistErr } // Server.SendProactive 通过委托给 hub 满足后台思考的 ActiveClient // 接缝,因此组合根能把网关自身交给汇聚点设置器。 func (s *Server) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (backthink.ProactiveDelivery, error) { return s.hub.SendProactive(ctx, msg) } // BroadcastProactivePersisted 把一条**刚落进主会话的主动消息**广播为**带结构化判据**的 // `message_persisted` 帧(切片 P0b)。判据就是帧的 `data` 里的 `proactive:true`:它是「她主动说的 // 话」在线上唯一的标记,客户端据此走与 `proactive_message` **同一套**提醒逻辑(非前台才 // 补一发通知,同一条只提醒一次),于是「她该不该提醒我」**与后端走了哪条路由无关** // (主人 2026-10-05 裁定 C)。 // // 它由 IM 桥的主动推送在**落库之后**调用(组合根接线 SetOnProactivePersisted)。为什么 // 需要它:IM 路由过去只发普通 `message_persisted`,而浏览器通知挂在 `proactive_message` // 上 ⇒ 走 IM 的主动消息一条都不弹;偏偏 `auto` 提示在主人有已连接 IM 渠道时优先走 IM // ⇒ 实际最可能发生的路径恰恰不弹。 // // 判据为什么不会误伤普通消息:只有主动推送这条通道会调用本函数,普通消息(Web 回合 / // IM 回复 / IM 镜像)走的仍是 BroadcastPersisted,那条路径**永不**往帧的 `data` 里放该键。 // 帧类型仍是 `message_persisted`,因此角色门(design/11 §7.1)与普通持久化帧逐字相同。 func (s *Server) BroadcastProactivePersisted(msg history.Message) { s.broadcastPersisted(msg, true) } // proactivePayloadFrom 把 UnifiedMessage 投影为消毒后的线上载荷。 // persisted 是刚落库的主会话消息(未接线历史存储或落库失败时为零值), // 它的 id/时间戳随之进入载荷,使客户端能把实时气泡对齐到历史行。 func proactivePayloadFrom(msg adapter.UnifiedMessage, persisted history.Message) proactivePayload { return proactivePayload{ Platform: msg.Platform, ChannelID: msg.ChannelID, ChannelType: string(msg.ChannelType), Content: msg.Content, ContentType: string(msg.ContentType), Sender: msg.Sender.DisplayName, Timestamp: msg.Timestamp, MessageID: msg.MessageID, ID: persisted.ID, CreatedAt: persisted.CreatedAt, } } // ownerAdminClaims 报告 claims 是否授权一次主动推送(owner 或 // admin)。nil claims(未认证 / 裸测试连接)绝不被作为目标。 func ownerAdminClaims(claims *auth.Claims) bool { if claims == nil { return false } return claims.Role == auth.RoleOwner || claims.Role == auth.RoleAdmin }