feat(gateway): 未读面 unreadCount 进 message_persisted 帧(契约 v0.76 ②)
- broadcastPersisted 是 Web 与 IM 两条路的**共用**函数 ⇒ 两处口径天然一致,不存在分叉
- 口径 = 绝对未读 + 主人单一已读游标(chat_sessions.read_at)⇒ 多端共享
- 可选键 omitempty:只在 >0 且确实算得出来时出现;0 与「算不出来」都不顶位
(0 只能从「会话有游标」推出,否则那个 0 就是编的)⇒ 缺席 = 老行为
- Docs/api-client.md:v0.76 ① 按实测改标「未实现且无实现路径」(见下),② 已实现,
并更正「GET /sessions 的 unreadCount 自 D4 起已在」、新增 §4.13 与 §7.4 第 23 条
⚠️ ① toolCalls[] 本版未落地(取证结论,非推断):chat_messages.turn_id(编排器随机路标识
turn-<16hex>)与 flow_graphs.turn_id(trace.FlowGraph.ID = <session>-<轮次序号>,进程内递增)
不同源,真实库中 66 条消息行 vs 21 条图行字面交集为空 ⇒ 无法 join;按时间窗硬凑属于猜,不做。
补齐路径:链路图带回合标识(新键 + 迁移 0023 起)+ 落盘列,之后 toolCalls[] 是一条批量 SQL。
我亲自复跑的变异:去掉帧里的 n > 0 守卫 ⇒ TestBroadcastPersistedUnreadCountAbsentWhenUnknown
与 TestBroadcastPersistedCarriesUnreadCount 双双 FAIL(断言「0 不该顶位」)⇒ 测试是真护栏。
This commit is contained in:
1 parent
cb232e7360
commit
fe4c702f87
6 files changed
+369
-8
No files matched your search
@@ -46,3 +46,39 @@ func (s *Store) MarkSessionRead(ctx context.Context, sessionID string, before ti
|
||||
}
|
||||
return readAt, nil
|
||||
}
|
||||
|
||||
// unreadCountQuery 统计**单个**会话里严格新于读游标的行数。它与
|
||||
// listSessionsQuery 的关联子查询是同一条谓词(等值 + 范围扫描,吃
|
||||
// (session_id, created_at DESC) 索引),只是从「按会话列表一次算全部」变
|
||||
// 「按单个会话算一次」—— 列表接口无 N+1 靠的是关联子查询,这里则是
|
||||
// 一次主键点查,供落库那一刻的单会话实时帧用。
|
||||
//
|
||||
// INNER JOIN chat_sessions 是**刻意**的:只有有会话行的会话(Web 主会话与
|
||||
// Web 子会话)才存在「读游标」这个概念。没有会话行的消息流(隔离渠道会话
|
||||
// sess-*、后台思考会话)因此恒得 0 —— 绝不把它们的全部历史当成未读,
|
||||
// 那种「未读」在被已读游标面(GET /sessions 根本不列它们)里没有归宿。
|
||||
const unreadCountQuery = `SELECT COUNT(*) FROM chat_messages c
|
||||
JOIN chat_sessions s ON s.id = c.session_id
|
||||
WHERE c.session_id = ? AND c.created_at > s.read_at`
|
||||
|
||||
// UnreadCount 返回该会话中**比读游标新**的消息数(绝对未读,非「自上次以来」)。
|
||||
// 它服务落库那一刻的实时面:`message_persisted` 帧要在同一口径下把角标换成
|
||||
// 权威数字,使客户端不必为清角标再拉一次 GET /sessions。
|
||||
//
|
||||
// 口径与 listSessionsQuery 逐字一致(同一个 `created_at > read_at`、同一条游标),
|
||||
// 因此两条面在同一时刻给出的数字相同;多端共享同一游标的取舍见 session_read.go
|
||||
// 顶部的包内说明,本函数不改它。
|
||||
//
|
||||
// 空 sessionID 与「会话不存在」都返回 0:没有游标的会话就没有未读这个概念,
|
||||
// 这与它们在 GET /sessions 里根本不出现是一致的。真正的查询失败**如实上抛**
|
||||
// (调用方据此省略该字段,绝不拿 0 冒充「已读完」)。
|
||||
func (s *Store) UnreadCount(ctx context.Context, sessionID string) (int, error) {
|
||||
if sessionID == "" {
|
||||
return 0, nil
|
||||
}
|
||||
var n int
|
||||
if err := s.db.QueryRowContext(ctx, unreadCountQuery, sessionID).Scan(&n); err != nil {
|
||||
return 0, fmt.Errorf("history: unread count %s: %w", sessionID, err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
@@ -346,3 +346,68 @@ func TestMigration0010ReadAtBackfill(t *testing.T) {
|
||||
t.Fatal("read_at should be dropped by the 0010 down migration")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStoreUnreadCount 锁定**单会话**未读数(契约 v0.76 的实时面用它换角标):
|
||||
// 它与 ListSessions 的列表口径同源 —— 同一个 `created_at > read_at` 谓词,
|
||||
// 因此同一时刻两条面给出同一个数字,不会有第二份未读定义。
|
||||
//
|
||||
// 四条边界一并钉住:空 id ⇒ 0;**没有会话行的消息流** ⇒ 0(不是「全部未读」,
|
||||
// 那种会话根本没有读游标、也不出现在已读游标面上);严格 `>`(游标压在 T 上的
|
||||
// 消息算已读);游标推进后归零、其后的新消息重新计入。
|
||||
func TestStoreUnreadCount(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newStore(t)
|
||||
ctx := context.Background()
|
||||
sub, err := st.NewSubSession(ctx, "owner")
|
||||
if err != nil {
|
||||
t.Fatalf("new sub: %v", err)
|
||||
}
|
||||
for _, m := range []Message{
|
||||
{SessionID: MainSessionID, Role: RoleUser, Content: "main-1", CreatedAt: 1000},
|
||||
{SessionID: MainSessionID, Role: RoleAssistant, Content: "main-2", CreatedAt: 2000},
|
||||
{SessionID: sub, Role: RoleUser, Content: "sub-1", CreatedAt: 1500},
|
||||
// 没有 chat_sessions 行的消息流(隔离渠道会话 / 后台思考会话的形状):
|
||||
// 消息落得进去,会话行从不插入。
|
||||
{SessionID: "sess-backthink", Role: RoleAssistant, Content: "no session row", CreatedAt: 9000},
|
||||
} {
|
||||
if err := st.Append(ctx, m); err != nil {
|
||||
t.Fatalf("seed %q: %v", m.Content, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 与列表口径逐条同值 —— 这是「不许有第二份未读定义」的机器化断言。
|
||||
for _, sessionID := range []string{MainSessionID, sub} {
|
||||
got, err := st.UnreadCount(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("UnreadCount(%s): %v", sessionID, err)
|
||||
}
|
||||
if want := sessionOf(t, st, sessionID).UnreadCount; got != want {
|
||||
t.Fatalf("UnreadCount(%s) = %d, but the session list says %d", sessionID, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
if got, err := st.UnreadCount(ctx, ""); err != nil || got != 0 {
|
||||
t.Fatalf("UnreadCount(\"\") = (%d, %v), want (0, nil)", got, err)
|
||||
}
|
||||
if got, err := st.UnreadCount(ctx, "sess-backthink"); err != nil || got != 0 {
|
||||
t.Fatalf("UnreadCount(sess-backthink) = (%d, %v), want (0, nil): 没有会话行就没有游标,不得把它的历史当成未读", got, err)
|
||||
}
|
||||
|
||||
// 严格 `>`:游标压在最新消息的毫秒上就把角标清零。
|
||||
if _, err := st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(2000)); err != nil {
|
||||
t.Fatalf("mark main read: %v", err)
|
||||
}
|
||||
if got, err := st.UnreadCount(ctx, MainSessionID); err != nil || got != 0 {
|
||||
t.Fatalf("UnreadCount(main) after cursor 2000 = (%d, %v), want (0, nil)", got, err)
|
||||
}
|
||||
// 游标之后的新消息重新计入,且**只**计它 —— 子会话的游标没被碰过。
|
||||
if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "main-3", CreatedAt: 3000}); err != nil {
|
||||
t.Fatalf("append main-3: %v", err)
|
||||
}
|
||||
if got, err := st.UnreadCount(ctx, MainSessionID); err != nil || got != 1 {
|
||||
t.Fatalf("UnreadCount(main) after a newer message = (%d, %v), want (1, nil)", got, err)
|
||||
}
|
||||
if got, err := st.UnreadCount(ctx, sub); err != nil || got != 1 {
|
||||
t.Fatalf("UnreadCount(sub) = (%d, %v), want (1, nil): 读主会话不得动子会话", got, err)
|
||||
}
|
||||
}
|
||||
@@ -32,6 +32,11 @@ func (s *Server) BroadcastPersisted(msg history.Message) {
|
||||
// 「这是她主动说的话」的结构化判据 `proactive:true`(切片 P0b,见 proactive.go 的
|
||||
// BroadcastProactivePersisted)。两条入口走**同一个**函数,因此帧类型推导、thinking
|
||||
// 抹除与角色门(design/11 §7.1)逐字一致 —— 差别只有那一个键。
|
||||
//
|
||||
// 帧的 data 另带**可选**的 `unreadCount`(契约 v0.76):该会话此刻的绝对未读数,
|
||||
// 与 GET /sessions 的 `unreadCount` 同一口径。它只在 >0 且真能算出来时出现(见
|
||||
// unreadCountOf),缺席即老行为。Web 路径与 IM 路径共用本函数 ⇒ 两处口径天然一致,
|
||||
// 不存在「Web 帧带、IM 帧不带」的分叉。
|
||||
func (s *Server) broadcastPersisted(msg history.Message, proactive bool) {
|
||||
wire := toHistoryMessage(msg)
|
||||
wire.Thinking = ""
|
||||
@@ -43,6 +48,13 @@ func (s *Server) broadcastPersisted(msg history.Message, proactive bool) {
|
||||
if proactive {
|
||||
data["proactive"] = true
|
||||
}
|
||||
// 未读数只在**确实算出来、且非零**时进帧(契约 v0.76)。缺席 = 老行为:老客户端
|
||||
// 忽略未知键,新客户端落下这帧时角标已经是对的,不必再拉一次 GET /sessions。
|
||||
// 不用 0 顶位:0 只能从「会话有游标」推出来,而算不出来(未接线/查询失败/
|
||||
// 会话没有游标)时那个 0 就是编的(见 unreadCountOf)。
|
||||
if n, ok := s.unreadCountOf(msg.SessionID); ok && n > 0 {
|
||||
data["unreadCount"] = n
|
||||
}
|
||||
s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: data})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
)
|
||||
|
||||
// 本文件钉住契约 v0.76 的**未读实时面**:`message_persisted` 帧的 data 携带该会话
|
||||
// 此刻的**绝对未读数**(与 GET /api/v1/sessions 的 unreadCount 同一口径、同一条
|
||||
// 读游标)。三条硬约束各有断言:值是对的;只在**确实算得出来且非零**时出现
|
||||
// (缺席 = 老行为);算不出来时绝不冒出一个 0 去冒充「已读完」。
|
||||
|
||||
// unreadOf 从帧 data 里读未读数并返回 (值, 是否在场)。它刻意先断言「在场性」:
|
||||
// 本面的关键取舍就是「缺席」与「0」是两回事。
|
||||
func unreadOf(t *testing.T, ev wsEvent) (int, bool) {
|
||||
t.Helper()
|
||||
data, ok := ev.Data.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("frame data = %+v, want a JSON object", ev.Data)
|
||||
}
|
||||
raw, present := data["unreadCount"]
|
||||
if !present {
|
||||
return 0, false
|
||||
}
|
||||
n, ok := raw.(float64)
|
||||
if !ok {
|
||||
t.Fatalf("frame data.unreadCount = %v (%T), want a number", raw, raw)
|
||||
}
|
||||
return int(n), true
|
||||
}
|
||||
|
||||
// TestBroadcastPersistedCarriesUnreadCount 覆盖顺路径与两条边界:帧上的数字是
|
||||
// **绝对**未读数(不是「自上次以来」),游标推进之后同一条广播路径不再带该键
|
||||
// (0 不顶位),游标之后的新消息重新点亮并给出刷新后的绝对值。末段顺带钉住
|
||||
// `channel_persisted` 与 `message_persisted` 走**同一个**载荷构造函数,因此隔离
|
||||
// 渠道会话(没有 chat_sessions 行)同样不带该键 —— 它连读游标都没有。
|
||||
func TestBroadcastPersistedCarriesUnreadCount(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, _, _, st := newHistoryEnv(t)
|
||||
c := newOwnerConn(srv.hub, 32)
|
||||
srv.hub.register(c)
|
||||
ctx := context.Background()
|
||||
|
||||
appendAt := func(at int64, content string) history.Message {
|
||||
t.Helper()
|
||||
full, err := st.AppendFull(ctx, history.Message{
|
||||
SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: content, CreatedAt: at,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("append %q: %v", content, err)
|
||||
}
|
||||
return full
|
||||
}
|
||||
|
||||
// ① 历史一条(1000)+ 刚落库的一条(2000),游标仍是 0 ⇒ 两条都未读。
|
||||
if err := st.Append(ctx, history.Message{
|
||||
SessionID: history.MainSessionID, Role: history.RoleUser, Content: "old", CreatedAt: 1000,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed old: %v", err)
|
||||
}
|
||||
srv.BroadcastPersisted(appendAt(2000, "new"))
|
||||
ev := readFakeFrame(t, c)
|
||||
if ev.Type != "message_persisted" {
|
||||
t.Fatalf("frame type = %q, want message_persisted", ev.Type)
|
||||
}
|
||||
if got, present := unreadOf(t, ev); !present || got != 2 {
|
||||
t.Fatalf("frame data.unreadCount = %d (present=%v), want 2 (绝对未读,不是本次 +1)", got, present)
|
||||
}
|
||||
|
||||
// ② 游标推进到最新消息 ⇒ 该会话已读:同一条广播路径**不再**带该键。
|
||||
if _, err := st.MarkSessionRead(ctx, history.MainSessionID, time.UnixMilli(2000)); err != nil {
|
||||
t.Fatalf("mark session read: %v", err)
|
||||
}
|
||||
srv.BroadcastPersisted(appendAt(2000, "exactly at the cursor"))
|
||||
if got, present := unreadOf(t, readFakeFrame(t, c)); present {
|
||||
t.Fatalf("frame data.unreadCount = %d,want 该键缺席:未读为 0 时不顶位(缺席 = 老行为,客户端回落 GET /sessions 收敛)", got)
|
||||
}
|
||||
|
||||
// ③ 游标之后的新消息 ⇒ 值是刷新后的绝对值。
|
||||
srv.BroadcastPersisted(appendAt(3000, "after the cursor"))
|
||||
if got, present := unreadOf(t, readFakeFrame(t, c)); !present || got != 1 {
|
||||
t.Fatalf("frame data.unreadCount = %d (present=%v), want 1", got, present)
|
||||
}
|
||||
|
||||
// ④ 隔离渠道会话走同一个构造函数 ⇒ 口径不分叉;它没有游标,因此不带该键。
|
||||
if err := st.Append(ctx, history.Message{
|
||||
SessionID: "sess-g", Role: history.RoleUser, Content: "isolated", CreatedAt: 5000,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed isolated: %v", err)
|
||||
}
|
||||
srv.BroadcastPersisted(history.Message{
|
||||
ID: "m-iso", SessionID: "sess-g", Role: history.RoleUser, Content: "isolated", CreatedAt: 5000,
|
||||
})
|
||||
iso := readFakeFrame(t, c)
|
||||
if iso.Type != "channel_persisted" {
|
||||
t.Fatalf("frame type = %q, want channel_persisted", iso.Type)
|
||||
}
|
||||
if got, present := unreadOf(t, iso); present {
|
||||
t.Fatalf("隔离会话的帧带上了 unreadCount = %d,want 该键缺席(没有会话行就没有读游标)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBroadcastPersistedUnreadCountAbsentWhenUnknown 钉住三条**诚实缺席**路径:
|
||||
// 未注入 HistoryStore、注入的存储没实现单会话未读视图(`sessionUnreadCounter`)、
|
||||
// 以及隔离渠道会话本身。三条都不许冒出 `unreadCount:0` —— 把「算不出来」写成
|
||||
// 「没有未读」就是编数据。
|
||||
func TestBroadcastPersistedUnreadCountAbsentWhenUnknown(t *testing.T) {
|
||||
t.Parallel()
|
||||
pending := history.Message{
|
||||
ID: "row-1", SessionID: history.MainSessionID, Role: history.RoleUser, Content: "hi", CreatedAt: 1000,
|
||||
}
|
||||
|
||||
t.Run("history not wired", func(t *testing.T) {
|
||||
srv, _, _ := buildEnv(t, cfgDefault(), Deps{})
|
||||
c := newOwnerConn(srv.hub, 32)
|
||||
srv.hub.register(c)
|
||||
|
||||
srv.BroadcastPersisted(pending)
|
||||
if got, present := unreadOf(t, readFakeFrame(t, c)); present {
|
||||
t.Fatalf("未接线历史存储的帧带上了 unreadCount = %d,want 缺席", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("store without the unread view", func(t *testing.T) {
|
||||
srv, _, _ := buildEnv(t, cfgDefault(), Deps{History: &fakeHistory{}})
|
||||
c := newOwnerConn(srv.hub, 32)
|
||||
srv.hub.register(c)
|
||||
|
||||
srv.BroadcastPersisted(pending)
|
||||
if got, present := unreadOf(t, readFakeFrame(t, c)); present {
|
||||
t.Fatalf("未实现 sessionUnreadCounter 的存储替身带上了 unreadCount = %d,want 缺席", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("isolated session has no cursor", func(t *testing.T) {
|
||||
srv, _, _, st := newHistoryEnv(t)
|
||||
c := newOwnerConn(srv.hub, 32)
|
||||
srv.hub.register(c)
|
||||
if err := st.Append(context.Background(), history.Message{
|
||||
SessionID: "sess-g", Role: history.RoleUser, Content: "iso", CreatedAt: 1000,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed isolated: %v", err)
|
||||
}
|
||||
|
||||
srv.BroadcastPersisted(history.Message{
|
||||
ID: "m-iso", SessionID: "sess-g", Role: history.RoleUser, Content: "iso", CreatedAt: 1000,
|
||||
})
|
||||
ev := readFakeFrame(t, c)
|
||||
if ev.Type != "channel_persisted" {
|
||||
t.Fatalf("frame type = %q, want channel_persisted", ev.Type)
|
||||
}
|
||||
if got, present := unreadOf(t, ev); present {
|
||||
t.Fatalf("隔离会话的帧带上了 unreadCount = %d,want 缺席", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
@@ -8,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
)
|
||||
|
||||
@@ -18,6 +20,11 @@ import (
|
||||
// 推导 unreadCount(history.go)。游标是会话级、单所有者的:
|
||||
// 一个账号每个会话一个游标 —— 在任意一端阅读即为所有端标记该
|
||||
// 会话已读。
|
||||
//
|
||||
// 同一条游标还喂养**落库那一刻**的实时面(契约 v0.76):单会话的绝对未读数
|
||||
// 由 unreadCountOf 读回,随 `message_persisted` 帧的 `data.unreadCount` 下发,
|
||||
// 使客户端不必为换角标再拉一次 GET /sessions。两条面共用同一个数字口径
|
||||
// (`created_at > read_at`),因此不会互相打脸。
|
||||
|
||||
// wsSessionRead 是已读游标推进的 WS 事件类型。
|
||||
const wsSessionRead = "session_read"
|
||||
@@ -92,3 +99,45 @@ func (h *hub) broadcastSessionRead(sessionID string, readAt int64) {
|
||||
Data: map[string]any{"readAt": readAt},
|
||||
})
|
||||
}
|
||||
|
||||
// sessionUnreadCounter 是 HistoryStore 的一个**可选**更窄视图(与 clientMsgIDReader /
|
||||
// turnIDWriter / documentStatusWriter 同一做法):按会话读回由已读游标推导的绝对未读数。
|
||||
// 未实现该视图的存储替身(测试替身、未接线的构建)**不报错**,只是帧上不带该键 ——
|
||||
// 未读面是增益,绝不能反过来卡住消息广播。
|
||||
type sessionUnreadCounter interface {
|
||||
UnreadCount(ctx context.Context, sessionID string) (int, error)
|
||||
}
|
||||
|
||||
const (
|
||||
// unreadCountTimeout 是**单次**未读数查询的时长上界。它跑在落库之后的广播路径上
|
||||
// (写完一条就发一帧),必须有一个界:一次库卡死绝不允许把广播挂住。
|
||||
unreadCountTimeout = 2 * time.Second
|
||||
|
||||
// unreadCountLogSource 是这条冷路径上日志的来源标签(AGENTS §10 禁无来源裸日志)。
|
||||
unreadCountLogSource = "gateway/broadcast"
|
||||
)
|
||||
|
||||
// unreadCountOf 返回该会话当前的**绝对未读数**(比 chat_sessions.read_at 新的消息数),
|
||||
// 第二个返回值表示「这个数字是真的」—— false 时调用方必须**省略**该键,绝不用 0 冒充
|
||||
// 「已读完」(缺席 = 老行为,客户端回落到 GET /sessions 收敛)。
|
||||
//
|
||||
// 三条不产生数字的情形,都不算错误:① 未注入 HistoryStore;② 注入的存储没实现上面那个
|
||||
// 窄视图;③ 查询失败(记一条带来源的 WARN,因为「读不到」不是「没有未读」)。
|
||||
func (s *Server) unreadCountOf(sessionID string) (int, bool) {
|
||||
if s.history == nil || sessionID == "" {
|
||||
return 0, false
|
||||
}
|
||||
counter, ok := s.history.(sessionUnreadCounter)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), unreadCountTimeout)
|
||||
defer cancel()
|
||||
n, err := counter.UnreadCount(ctx, sessionID)
|
||||
if err != nil {
|
||||
logger.WithSource(s.log, unreadCountLogSource).Warn("unread count read failed",
|
||||
"err", err, "session", sessionID)
|
||||
return 0, false
|
||||
}
|
||||
return n, true
|
||||
}
|
||||
Reference in new issue
Block a user