Merge branch 'feat/m25-proactiveweb-route-web-proactive-pus' into dev
This commit is contained in:
AskaEth committed
commit
960d7b46fd
5 files changed
+173
-7
No files matched your search
@@ -9,6 +9,7 @@ import (
|
||||
"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"
|
||||
)
|
||||
|
||||
@@ -43,10 +44,20 @@ type proactivePayload struct {
|
||||
// SendProactive 把 msg 投递给每个存活的 owner/admin 连接，并返回投递与落库
|
||||
// 两件事的结果（切片 P0，`backthink.ProactiveDelivery`）。
|
||||
//
|
||||
// 顺序是刻意的 —— **先落库、再发帧**：主动消息必须成为主会话
|
||||
// 顺序是刻意的 —— **先规整、再落库、再发帧**：主动消息必须成为主会话
|
||||
// （`history.MainSessionID`）里的一条真实消息（assistant 角色），否则刷新/
|
||||
// 重载后什么都补不回来（这正是「气泡恒落」的根）；而帧要带上落库行的 id，
|
||||
// 客户端才能把实时气泡对齐到历史行。
|
||||
// 客户端才能把实时气泡对齐到历史行。规整必须落在落库**之前** —— 落库行是
|
||||
// 权威文本，先落库后规整只会在库里留下脏的、而帧与历史行逐字对不上。
|
||||
//
|
||||
// 文本与其它三处**同源**（Web 会话回合 / IM 出站 / 主动推送 IM 侧）：一律走
|
||||
// `turnreply.NormalizeParagraphs`，本文件绝不另写一份实现（本仓已因「同一条
|
||||
// 规则两份写法」栽过多次）。修前的真缺口正是这条路由**完全绕过**它（M19 复审
|
||||
// 抓到）：同一条主动消息走 IM 是规整过的、走 Web 实时推还是原样，半包 `<p>` 在
|
||||
// 客户端「只要存在完整 `<p>` 块就不再按空行回退切分」的契约下整条塌成一坨。
|
||||
// 规整是幂等的写入侧加法（已规范的文本逐字不变），因此 `<p>` = 一段一气泡、
|
||||
// `<br>` = 气泡内分隔、`<action>`/`<narration>` 随所在段落进同一个 `<p>` 这些
|
||||
// 既有语义一条不动。
|
||||
//
|
||||
// 落库失败**不静默**：记一条带来源的 WARN，并把错误返回给调用方，同时
|
||||
// Persisted=false —— 帧照发（实时气泡不能丢），但记录里绝不说「已推送」。
|
||||
@@ -60,6 +71,9 @@ func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (ba
|
||||
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",
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/turnreply"
|
||||
)
|
||||
|
||||
// 本文件锁定 M19 复审抓到的真缺口（切片 M25）：**Web 活跃客户端**这条主动推送
|
||||
// 路由过去把 msg.Content 原样落库、原样推帧，**完全绕过**
|
||||
// `turnreply.NormalizeParagraphs` —— 于是同一条主动消息走 IM 是规整过的
|
||||
// （channelbridge/proactive.go 出口），走 Web 实时推还是原样：半包 `<p>`
|
||||
// （只有末段被包）在客户端「只要存在完整 <p> 块就不再按空行回退切分」的契约下
|
||||
// 整条塌成一坨、而不是多个气泡。
|
||||
//
|
||||
// 三条断言缺一不可：
|
||||
// - 落库行吃的是**规整后**的文本（顺序：先规整、再落库 —— 否则库里还是脏的）；
|
||||
// - 推帧文本与落库行**逐字相同**（只断言「各自含换行」会漏掉两侧分叉）；
|
||||
// - 规整确实发生了（原文 ≠ 期望值），否则断言会退化成同义反复。
|
||||
|
||||
// proactiveHalfWrapped 是缺陷形状的输入：第一段裸文本用空行分隔，末段已包 `<p>`。
|
||||
// 它同时带**字面换行**（`\n`）与半包段落 —— 正是主人 2026-10-24 报的那两类。
|
||||
const proactiveHalfWrapped = "今晚早点睡。\n\n<p>明天还有早会。</p>"
|
||||
|
||||
// proactiveHalfWrappedFixed 是同一段文本经共享规整器后的规范形态：每段一个 <p>，
|
||||
// 段间空行原样保留（客户端因此拆出两个气泡）。
|
||||
const proactiveHalfWrappedFixed = "<p>今晚早点睡。</p>\n\n<p>明天还有早会。</p>"
|
||||
|
||||
// TestHub_SendProactive_NormalizesParagraphsAndFrameMatchesRow 是 M25 的主断言。
|
||||
func TestHub_SendProactive_NormalizesParagraphsAndFrameMatchesRow(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newHistoryStoreForGateway(t)
|
||||
hb := newTestHub(nil, nil)
|
||||
hb.history = store
|
||||
owner := newOwnerConn(hb, 8)
|
||||
hb.register(owner)
|
||||
|
||||
if _, err := hb.SendProactive(context.Background(), proactiveStoreMsg(proactiveHalfWrapped)); err != nil {
|
||||
t.Fatalf("SendProactive err = %v", err)
|
||||
}
|
||||
|
||||
rows := mainSessionRows(t, store)
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条", len(rows))
|
||||
}
|
||||
row := rows[0]
|
||||
if row.Content == proactiveHalfWrapped {
|
||||
t.Fatalf("落库文本仍与原文逐字相同 ⇒ 规整没生效（本路由绕过了 NormalizeParagraphs）")
|
||||
}
|
||||
if row.Content != proactiveHalfWrappedFixed {
|
||||
t.Fatalf("落库文本 =\n%q\nwant\n%q\n（Web 活跃路由必须与其它三处同源：先规整再落库）",
|
||||
row.Content, proactiveHalfWrappedFixed)
|
||||
}
|
||||
|
||||
frame := readFakeFrame(t, owner)
|
||||
if frame.Type != wsProactiveMessage {
|
||||
t.Fatalf("frame type = %q, want %q", frame.Type, wsProactiveMessage)
|
||||
}
|
||||
data, ok := frame.Data.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("frame data type = %T, want map[string]any", frame.Data)
|
||||
}
|
||||
content, _ := data["content"].(string)
|
||||
if content != row.Content {
|
||||
t.Fatalf("推帧文本与落库行必须逐字相同：\nframe=%q\nrow =%q", content, row.Content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHub_SendProactive_KeepsExistingRenderSemantics 钉住「不许改语义」：规整只做
|
||||
// 段落包裹，`<br>` 仍是气泡内分隔（由客户端按既有契约渲染）、`<action>` 仍随所在
|
||||
// 段落进**同一个** `<p>`（Web 灰斜体、IM 出口剥除，两条都不许被拆散）。
|
||||
func TestHub_SendProactive_KeepsExistingRenderSemantics(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newHistoryStoreForGateway(t)
|
||||
hb := newTestHub(nil, nil)
|
||||
hb.history = store
|
||||
owner := newOwnerConn(hb, 8)
|
||||
hb.register(owner)
|
||||
|
||||
raw := "先点点头。<br>再说一句。\n\n<action>把杯子放下</action>"
|
||||
if _, err := hb.SendProactive(context.Background(), proactiveStoreMsg(raw)); err != nil {
|
||||
t.Fatalf("SendProactive err = %v", err)
|
||||
}
|
||||
rows := mainSessionRows(t, store)
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条", len(rows))
|
||||
}
|
||||
row := rows[0]
|
||||
|
||||
want := turnreply.NormalizeParagraphs(raw)
|
||||
if row.Content != want {
|
||||
t.Fatalf("落库文本 =\n%q\nwant\n%q", row.Content, want)
|
||||
}
|
||||
if !strings.Contains(row.Content, "<br>") {
|
||||
t.Fatalf("文本里的 <br> 被改掉了：%q（它仍是气泡内分隔，不许被吞）", row.Content)
|
||||
}
|
||||
if !strings.Contains(row.Content, "<p><action>把杯子放下</action></p>") {
|
||||
t.Fatalf("action 段必须随所在段落进同一个 <p>：%q", row.Content)
|
||||
}
|
||||
|
||||
frame := readFakeFrame(t, owner)
|
||||
data, ok := frame.Data.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("frame data type = %T, want map[string]any", frame.Data)
|
||||
}
|
||||
if content, _ := data["content"].(string); content != row.Content {
|
||||
t.Fatalf("推帧文本与落库行必须逐字相同：\nframe=%q\nrow =%q", content, row.Content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestProactiveRoutes_NormalizeSameText 是「同源」的可执行证据：同一条主动消息分别
|
||||
// 走 Web 活跃路由与 IM 桥路由，两侧落库文本必须逐字相同 —— 两条路由各写一份规整
|
||||
// 实现就迟早漂移，本仓已因此栽过多次。
|
||||
func TestProactiveRoutes_NormalizeSameText(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newHistoryStoreForGateway(t)
|
||||
hb := newTestHub(nil, nil)
|
||||
hb.history = store
|
||||
hb.register(newOwnerConn(hb, 8))
|
||||
bridge, _ := newProactiveTestBridge(t, store, nil)
|
||||
|
||||
msg := proactiveStoreMsg(proactiveHalfWrapped)
|
||||
if _, err := hb.SendProactive(context.Background(), msg); err != nil {
|
||||
t.Fatalf("active route err = %v", err)
|
||||
}
|
||||
if rows := mainSessionRows(t, store); len(rows) != 1 {
|
||||
t.Fatalf("活跃路由之后主会话有 %d 条，want 1", len(rows))
|
||||
}
|
||||
if err := bridge.SendProactive(context.Background(), msg); err != nil {
|
||||
t.Fatalf("im route err = %v", err)
|
||||
}
|
||||
rows := mainSessionRows(t, store)
|
||||
if len(rows) != 2 {
|
||||
t.Fatalf("两条路由各跑一次后主会话有 %d 条，want 2", len(rows))
|
||||
}
|
||||
// rows 最新优先：第一条是 IM 路由刚写的，第二条是活跃路由写的。
|
||||
if rows[0].Content != rows[1].Content {
|
||||
t.Fatalf("两条路由的落库文本必须逐字相同（同一份规整）：\nim =%q\nweb =%q",
|
||||
rows[0].Content, rows[1].Content)
|
||||
}
|
||||
if rows[0].Content != proactiveHalfWrappedFixed {
|
||||
t.Fatalf("两条路由的文本 =\n%q\nwant\n%q", rows[0].Content, proactiveHalfWrappedFixed)
|
||||
}
|
||||
}
|
||||
@@ -24,7 +24,8 @@ import (
|
||||
// 三条口径：
|
||||
// - **角色恒为 assistant**：这是她单方面开口说的话，不是主人的输入 —— 塞成
|
||||
// user 角色会污染「谁说了什么」，也会在历史里冒充主人发言（红线）。
|
||||
// - **正文用调用方已经净化过的文本**（backthink 的 stripInternalMarkers 出口），
|
||||
// - **正文用调用方已经净化并规整过的文本**（backthink 的 stripInternalMarkers 出口，
|
||||
// 再过 `turnreply.NormalizeParagraphs` —— 见 SendProactive 的「先规整、再落库」），
|
||||
// 系统元数据（决策标记、渠道选择）绝不进正文，也绝不新增任何隐藏字段 ——
|
||||
// 注入面因此与「气泡里看到的那句话」逐字相同。
|
||||
// - **turn_id 留空**：主动推送不属于任何一路回合（她单方面开口，没有对应的
|
||||
|
||||
@@ -83,8 +83,10 @@ func TestHub_SendProactive_PersistsAssistantMessageToMainSession(t *testing.T) {
|
||||
if row.SessionID != history.MainSessionID {
|
||||
t.Fatalf("session = %q, want %q", row.SessionID, history.MainSessionID)
|
||||
}
|
||||
if row.Content != "记得喝水哦" {
|
||||
t.Fatalf("content = %q, want the proactive text verbatim", row.Content)
|
||||
// 正文经共享规整器（turnreply.NormalizeParagraphs）：纯散文被包成一个 `<p>`
|
||||
// ——与 Web 回合 / IM 出站 / IM 侧主动推送同一份口径（M25：本路由曾绕过它）。
|
||||
if row.Content != "<p>记得喝水哦</p>" {
|
||||
t.Fatalf("content = %q, want the normalized proactive text", row.Content)
|
||||
}
|
||||
if row.TurnID != "" {
|
||||
t.Fatalf("turn_id = %q, want empty（主动推送不属于任何一路回合）", row.TurnID)
|
||||
|
||||
@@ -52,8 +52,11 @@ func TestHub_SendProactive_OwnerAndAdmin(t *testing.T) {
|
||||
if !ok {
|
||||
t.Fatalf("frame data type = %T, want map[string]any", ev.Data)
|
||||
}
|
||||
if data["content"] != "记得喝水哦" {
|
||||
t.Fatalf("frame content = %q, want the proactive text", data["content"])
|
||||
// 正文经共享规整器（turnreply.NormalizeParagraphs）：纯散文也会被包成
|
||||
// 一个 `<p>` —— 这条路由与 Web 回合 / IM 出站 / IM 侧主动推送**同源**，
|
||||
// 未包裹的裸文本同样会包上（M25 修的就是本路由曾绕过它）。
|
||||
if data["content"] != "<p>记得喝水哦</p>" {
|
||||
t.Fatalf("frame content = %q, want the normalized proactive text", data["content"])
|
||||
}
|
||||
if data["sender"] != "Cyrene" {
|
||||
t.Fatalf("frame sender = %q, want Cyrene", data["sender"])
|
||||
|
||||
Reference in new issue
Block a user