feat(history,gateway): D4 会话未读/已读游标 — unreadCount + POST /sessions/{id}/read + session_read 广播

- chat_sessions 增 read_at（毫秒，与 chat_messages.created_at 同单位）；迁移 0010 并把既有会话回填为各自最新消息时间（升级当天不误报）
- ListSessions 单查询派生 unreadCount（无 N+1，复用 idx_chat_msgs_session_created 覆盖索引）
- MarkSessionRead 只前进 MAX(read_at, ?)；未知会话 ErrSessionNotFound；游标方法挂在网关消费侧窄接口（沿用 GetMessage/messageLookup 先例，避免污染 runtime 接口与其 fake）
- POST /api/v1/sessions/{id}/read：writeAuth + 400/404/501，成功广播 session_read、失败路径不广播
- Docs/api-client v0.37（§1 unreadCount / 新行 8a / §3.9 session_read / §4.2 用法）
- 含迁移回填测试、store/handler 测试与变异验证；顺带把 migrate_pinned_test 的「回退一步=0009」硬编码改为对迁移数量免疫

internal/services/gateway/contracts.go

@@ -116,6 +116,11 @@ type HistoryStore interface {
 	// history.ErrSessionNotFound.
 	SetPinned(ctx context.Context, sessionID string, pinned bool) error
 	NewSubSession(ctx context.Context, ownerID string) (string, error)
+	// MarkSessionRead advances a session's read cursor (D4) to before and returns
+	// the resulting cursor as a Unix millisecond timestamp; a zero before means
+	// "now". The cursor only moves forward. An unknown id returns
+	// history.ErrSessionNotFound.
+	MarkSessionRead(ctx context.Context, sessionID string, before time.Time) (int64, error)
 	ListSessions(ctx context.Context) ([]history.SessionMeta, error)
 }
 

internal/services/gateway/history.go

@@ -26,6 +26,12 @@ type historySessionSummary struct {
 	// per-session request. It is omitted for a session with no messages. Content is
 	// already truncated by the store.
 	LastMessage *historyLastMessage `json:"lastMessage,omitempty"`
+	// UnreadCount is the number of messages newer than the session's read cursor
+	// (D4), derived in the same list query. It is always returned (0 when nothing
+	// is unread) so a client can render/clear the badge without a second call. The
+	// cursor is session-scoped and single-owner: reading on any one end clears the
+	// badge for every end of the account.
+	UnreadCount int `json:"unreadCount"`
 }
 
 // historyLastMessage is the wire shape of the last-message preview in a session summary.
@@ -79,6 +85,7 @@ func (s *Server) handleHistorySessions(w http.ResponseWriter, r *http.Request) {
 			CreatedAt:   m.CreatedAt,
 			Pinned:      m.Pinned,
 			LastMessage: toHistoryLastMessage(m.LastMessage),
+			UnreadCount: m.UnreadCount,
 		})
 	}
 	writeJSON(w, http.StatusOK, out)

internal/services/gateway/routes.go

@@ -83,6 +83,10 @@ func (s *Server) registerHistoryRoutes(mux *http.ServeMux, writeAuth func(http.H
 	mux.Handle("POST /api/v1/sessions", writeAuth(http.HandlerFunc(s.handleCreateSubSession)))
 	mux.Handle("DELETE /api/v1/sessions/{id}", writeAuth(http.HandlerFunc(s.handleDeleteSession)))
 	mux.Handle("PATCH /api/v1/sessions/{id}", writeAuth(http.HandlerFunc(s.handlePatchSession)))
+	// Read cursor (D4 / Android collaboration B3): advancing it clears the session's
+	// unread badge on every end of the account, so it is an owner/admin history write
+	// like the other session mutations. It broadcasts a `session_read` frame.
+	mux.Handle("POST /api/v1/sessions/{id}/read", writeAuth(http.HandlerFunc(s.handleMarkSessionRead)))
 	// Single-message edit/delete (Android collaboration batch ③a, D3). A message is
 	// addressed by its own opaque id; both are owner/admin-gated history writes.
 	mux.Handle("PATCH /api/v1/messages/{id}", writeAuth(http.HandlerFunc(s.handlePatchMessage)))

internal/services/gateway/session_orchestrator_test.go

@@ -231,6 +231,13 @@ func (f *fakeHistory) ListSessions(context.Context) ([]history.SessionMeta, erro
 	return nil, nil
 }
 
+// MarkSessionRead satisfies the HistoryStore contract for the read-cursor path (D4).
+// The orchestrator never advances a cursor, so the fake only reports the requested
+// cursor back without storing it.
+func (f *fakeHistory) MarkSessionRead(_ context.Context, _ string, before time.Time) (int64, error) {
+	return before.UnixMilli(), nil
+}
+
 func (f *fakeHistory) records() []history.Message {
 	f.mu.Lock()
 	defer f.mu.Unlock()

internal/services/gateway/session_read.go

@@ -0,0 +1,97 @@
+package gateway
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"time"
+
+	apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
+)
+
+// This file owns the D4 read-cursor surface (Android collaboration B3): the
+// POST /api/v1/sessions/{id}/read endpoint that advances a session's read cursor,
+// plus the `session_read` realtime frame it broadcasts so the account's other
+// browsers clear the unread badge without a re-read. The cursor itself lives in
+// chat_sessions.read_at (internal/runtime/history); GET /sessions derives the
+// unreadCount from it (history.go). The cursor is session-scoped and single-owner:
+// one account, one cursor per session — reading on any one end marks the session
+// read for every end.
+
+// wsSessionRead is the WS event type for a read-cursor advance.
+const wsSessionRead = "session_read"
+
+// sessionReadRequest is the POST /api/v1/sessions/{id}/read body. Before is a
+// pointer so an absent field (an empty body) is distinguishable from an explicit
+// value: absent means "now" — the client marks everything it has seen read.
+type sessionReadRequest struct {
+	Before *int64 `json:"before"`
+}
+
+// handleMarkSessionRead advances a session's read cursor and answers
+// {data:{sessionId, readAt}} with the resulting cursor in Unix milliseconds.
+// It is owner/admin-gated like the other history writes (writeAuth).
+//
+// body {before?} is an optional millisecond timestamp bounding what is marked
+// read — messages with created_at <= readAt become read. Absent (or an empty body)
+// means now. A present non-positive or non-numeric before answers 400: the cursor
+// is a real millisecond timestamp, and "unread by a negative cursor" is not a
+// meaningful client request. An unknown session id answers 404; a missing
+// HistoryStore answers 501.
+//
+// On success the new cursor is broadcast as a `session_read` frame so the account's
+// other browsers clear the badge too (reading on one end = read everywhere). A
+// failure path broadcasts nothing.
+func (s *Server) handleMarkSessionRead(w http.ResponseWriter, r *http.Request) {
+	if s.history == nil {
+		writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented})
+		return
+	}
+	id := r.PathValue("id")
+	if id == "" {
+		writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "session id is required"))
+		return
+	}
+	var req sessionReadRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil && !errors.Is(err, io.EOF) {
+		writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body"))
+		return
+	}
+	var before time.Time
+	if req.Before != nil {
+		if *req.Before <= 0 {
+			writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "before must be a positive millisecond timestamp"))
+			return
+		}
+		before = time.UnixMilli(*req.Before)
+	}
+	readAt, err := s.history.MarkSessionRead(r.Context(), id, before)
+	if err != nil {
+		if errors.Is(err, history.ErrSessionNotFound) {
+			writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "session %q not found", id))
+			return
+		}
+		s.log.Error("history mark session read", "err", err, "session", id)
+		writeError(w, apperr.Wrap(apperr.ErrInternal, "session_failed", "failed to mark session read"))
+		return
+	}
+	s.hub.broadcastSessionRead(id, readAt)
+	writeJSON(w, http.StatusOK, map[string]any{"sessionId": id, "readAt": readAt})
+}
+
+// broadcastSessionRead fans a read-cursor advance out to every client as a
+// `session_read` frame carrying the session's new cursor, so the same account's
+// other ends clear the session's unread badge without a re-read. Like the other
+// history frames it is an all-connections broadcast (the hub has no per-session
+// subscription), and the client filters it by sessionId; the cursor's single source
+// of truth stays chat_sessions.read_at, so a client that missed the frame converges
+// on its next GET /sessions.
+func (h *hub) broadcastSessionRead(sessionID string, readAt int64) {
+	h.broadcast(wsEvent{
+		Type:      wsSessionRead,
+		SessionID: sessionID,
+		Data:      map[string]any{"readAt": readAt},
+	})
+}

internal/services/gateway/session_read_test.go

@@ -0,0 +1,233 @@
+package gateway
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
+)
+
+// These tests cover the D4 read-cursor contract (Android collaboration B3): the
+// unreadCount carried by GET /api/v1/sessions, the advance endpoint
+// POST /api/v1/sessions/{id}/read, and the `session_read` frame that clears the badge
+// on the account's other browsers.
+
+// unreadBySession decodes GET /api/v1/sessions into a map of session id → unreadCount via
+// the wire key, so the test asserts the JSON contract rather than the in-package struct.
+func unreadBySession(t *testing.T, ts *httptest.Server, tok string) map[string]int {
+	t.Helper()
+	resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("list sessions status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var sessions []struct {
+		ID          string `json:"id"`
+		UnreadCount int    `json:"unreadCount"`
+	}
+	decodeDataSlice(t, body, &sessions)
+	out := make(map[string]int, len(sessions))
+	for _, s := range sessions {
+		out[s.ID] = s.UnreadCount
+	}
+	return out
+}
+
+// TestSessionsUnreadCountAndMarkRead covers the happy path end to end: a session with
+// unread messages reports the count, POST /read moves the cursor and answers
+// {sessionId, readAt}, the list drops to the messages sent after the cursor, and a
+// body-less POST (mark everything seen) clears the badge.
+func TestSessionsUnreadCountAndMarkRead(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt, st := newHistoryEnv(t)
+	ctx := context.Background()
+	sub, err := st.NewSubSession(ctx, "owner")
+	if err != nil {
+		t.Fatalf("new sub: %v", err)
+	}
+	for _, m := range []history.Message{
+		{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "a", CreatedAt: 1000},
+		{SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "b", CreatedAt: 2000},
+		{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "c", CreatedAt: 3000},
+		{SessionID: sub, Role: history.RoleUser, Content: "sub", CreatedAt: 1500},
+	} {
+		if err := st.Append(ctx, m); err != nil {
+			t.Fatalf("seed %q: %v", m.Content, err)
+		}
+	}
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+	base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/read"
+
+	// Nothing read yet: the main session is fully unread, the sub session independently so.
+	if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 3 || got[sub] != 1 {
+		t.Fatalf("initial unread = %+v, want main 3 / sub 1", got)
+	}
+
+	// Advance the main cursor to 2000: messages at or before it are read.
+	resp, body := doReq(t, http.MethodPost, base, map[string]any{"before": 2000}, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("mark read status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	env := decodeBody(t, body)
+	if len(env) != 1 {
+		t.Fatalf("envelope = %v, want only a data field", env)
+	}
+	var data struct {
+		SessionID string `json:"sessionId"`
+		ReadAt    int64  `json:"readAt"`
+	}
+	if err := json.Unmarshal(env["data"], &data); err != nil {
+		t.Fatalf("decode read data: %v (body=%s)", err, body)
+	}
+	if data.SessionID != history.MainSessionID || data.ReadAt != 2000 {
+		t.Fatalf("read data = %+v, want sessionId %s readAt 2000", data, history.MainSessionID)
+	}
+
+	// Only the later message stays unread, and the sub session is untouched.
+	if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 1 || got[sub] != 1 {
+		t.Fatalf("unread after cursor 2000 = %+v, want main 1 / sub 1", got)
+	}
+
+	// No body = "mark everything I have seen read" (resolved to now server-side).
+	resp, body = doReq(t, http.MethodPost, base, nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("body-less mark status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	if err := json.Unmarshal(decodeBody(t, body)["data"], &data); err != nil {
+		t.Fatalf("decode body-less read data: %v (body=%s)", err, body)
+	}
+	if data.ReadAt < 3000 {
+		t.Fatalf("body-less readAt = %d, want >= the newest message 3000", data.ReadAt)
+	}
+	if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 0 || got[sub] != 1 {
+		t.Fatalf("unread after mark-now = %+v, want main 0 / sub 1", got)
+	}
+	markedNow := data.ReadAt
+
+	// The cursor only moves forward: re-posting a stale cursor must not re-open the badge.
+	resp, body = doReq(t, http.MethodPost, base, map[string]any{"before": 1000}, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("stale mark status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	if err := json.Unmarshal(decodeBody(t, body)["data"], &data); err != nil {
+		t.Fatalf("decode stale read data: %v", err)
+	}
+	if data.ReadAt != markedNow {
+		t.Fatalf("stale readAt = %d, want the cursor to stay at %d", data.ReadAt, markedNow)
+	}
+	if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 0 {
+		t.Fatalf("unread after stale mark = %d, want 0", got[history.MainSessionID])
+	}
+}
+
+// TestMarkSessionReadValidation covers the failure contract: a present but unusable
+// `before` answers 400, an unknown session 404, a viewer 403 and a missing HistoryStore
+// 501 — none of which may be mistaken for a successful read.
+func TestMarkSessionReadValidation(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt, _ := newHistoryEnv(t)
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+	base := ts.URL + "/api/v1/sessions/web/read"
+
+	for _, tc := range []struct {
+		name string
+		body any
+	}{
+		{name: "zero", body: map[string]any{"before": 0}},
+		{name: "negative", body: map[string]any{"before": -1}},
+		{name: "non-numeric", body: map[string]any{"before": "abc"}},
+		{name: "not an object", body: "not-an-object"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			resp, body := doReq(t, http.MethodPost, base, tc.body, tok)
+			if resp.StatusCode != http.StatusBadRequest {
+				t.Fatalf("status = %d, want 400 (body=%s)", resp.StatusCode, body)
+			}
+			wantErrorShape(t, body)
+		})
+	}
+
+	// An empty object means "now", not "no-op".
+	resp, body := doReq(t, http.MethodPost, base, map[string]any{}, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("empty object status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+
+	// Unknown session → 404; the session id is not auto-created.
+	resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/web:missing/read", map[string]any{"before": 1000}, tok)
+	if resp.StatusCode != http.StatusNotFound {
+		t.Fatalf("unknown session status = %d, want 404 (body=%s)", resp.StatusCode, body)
+	}
+
+	// A viewer cannot write the cursor (owner/admin history write, like PATCH /sessions).
+	viewer := signAny(t, jwt, "acct", auth.RoleViewer)
+	resp, body = doReq(t, http.MethodPost, base, map[string]any{"before": 1000}, viewer)
+	if resp.StatusCode != http.StatusForbidden {
+		t.Fatalf("viewer status = %d, want 403 (body=%s)", resp.StatusCode, body)
+	}
+
+	// And no token at all is 401, not a silent success.
+	resp, body = doReq(t, http.MethodPost, base, map[string]any{"before": 1000}, "")
+	if resp.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("anonymous status = %d, want 401 (body=%s)", resp.StatusCode, body)
+	}
+}
+
+// TestMarkSessionReadNotWired pins the honest degradation: with no HistoryStore injected
+// the endpoint answers 501 not_implemented rather than pretending the cursor moved.
+func TestMarkSessionReadNotWired(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+
+	resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/web/read", map[string]any{"before": 1000}, tok)
+	if resp.StatusCode != http.StatusNotImplemented {
+		t.Fatalf("status = %d, want 501 (body=%s)", resp.StatusCode, body)
+	}
+	wantErrorShape(t, body)
+}
+
+// TestMarkSessionReadBroadcasts asserts the cross-end sync frame: a successful advance
+// pushes exactly one `session_read` frame scoped to that session and carrying the new
+// cursor, while a failed one (404) pushes nothing.
+func TestMarkSessionReadBroadcasts(t *testing.T) {
+	t.Parallel()
+	srv, ts, jwt, _ := newHistoryEnv(t)
+	c := newFakeConn(srv.hub, 32)
+	srv.hub.register(c)
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+
+	resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/read", map[string]any{"before": 2000}, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	ev := readFakeFrame(t, c)
+	if ev.Type != "session_read" {
+		t.Fatalf("frame type = %q, want session_read (frame=%+v)", ev.Type, ev)
+	}
+	if ev.SessionID != history.MainSessionID {
+		t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, history.MainSessionID)
+	}
+	data, ok := ev.Data.(map[string]any)
+	if !ok {
+		t.Fatalf("frame data = %+v, want object", ev.Data)
+	}
+	if data["readAt"] != float64(2000) {
+		t.Fatalf("frame data.readAt = %v, want 2000", data["readAt"])
+	}
+	if err := connHasNoFrame(c); err != nil {
+		t.Fatal(err)
+	}
+
+	// A failed advance must not tell other ends the badge was cleared.
+	resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/web:missing/read", map[string]any{"before": 2000}, tok)
+	if resp.StatusCode != http.StatusNotFound {
+		t.Fatalf("unknown status = %d, want 404 (body=%s)", resp.StatusCode, body)
+	}
+	if err := connHasNoFrame(c); err != nil {
+		t.Fatal(err)
+	}
+}