AskaEth/Cyrene_v2
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(gateway): WS 帧门相关 11 个文件的注释改中文（只改注释，代码零字节改动）
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
Details
CI / Frontend (tsc + vite build) (push) Has been cancelled
Details

Browse Source 
主人 2026-10-03 拍板「注释一律中文」。本次先把本会话动过的文件就地翻；
不做全仓一次性翻译，存量按包分批排期（见 工作小结.md 待办）。

范围：ws_frame_scope.go / ws_frame_scope_test.go / broadcast.go / ws_hub_turn.go /
ws_hub_message_event.go / session_read.go / server.go / ws_hub_stream_scope_test.go /
history_write_broadcast_test.go / mood_test.go / extra_test.go
含中文的整行注释 18 → 382。

校验（判据 = "只改注释、零代码字节改动"）：
- `git diff -U0 | 排除 +++/--- | 排除注释行/空行/行尾注释` → 为空
- 独立实现：两侧剥掉注释后逐行比对代码文本 → 11/11 文件全等，不一致数 0
- 行尾注释 7 行逐条列出，代码列与原文逐字节相同（gofmt 未要求重排）
- gofmt -l 干净；全树 go build/vet 绿；go test ./... = 58 包 ok / 0 FAIL；
  linux/amd64 + darwin/arm64 交叉编译绿
语义未删减：部分折行英文长句合并为更少行，故整行注释数 438 → 399。
This commit is contained in:
AskaEth committed 2026-10-03 22:07:56 +08:00
1 parent 770a83c3fb
commit fe1ff33115
11 files changed
+392 -431
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/services/gateway/broadcast.go
+34 -39
View File
@@ -6,23 +6,20 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)
// This file owns the gateway's outbound realtime pushes: the persisted-message
// frames shared with the channel bridge plus the per-persona state frames the
// frontend renders (mood badge, capacity awaken).
// 本文件归属 gateway 的下行实时推送：与 channel bridge 共用的
// 持久化消息帧，以及前端渲染的角色级状态帧（心情徽标、能力唤醒）。
// BroadcastPersisted broadcasts a just-persisted chat message as a WS event so
// every browser of the same account appends it without a manual refresh
// (design/16 §4.2). It is shared by the orchestrator (Web turns) and the channel
// bridge (IM messages), so both paths flow through one frame shape. The event type
// is split by session: the main web session and its web sub-sessions broadcast
// `message_persisted` for web-chat realtime sync, while any other session id is an
// isolated channel adapter session (sess-xxx) and broadcasts `channel_persisted`
// for the IM adapter log page — the web chat has no handler for the latter, so it
// is naturally ignored there and never pollutes the conversation. Thinking is
// stripped to stay consistent with the WS history replay (design/16 §6). It is a
// no-op-safe call when the hub has broadcast wired. Both frame types carry message
// content, so the design/11 §7.1 frame gate delivers them to owner/admin
// connections only.
// BroadcastPersisted 把一条刚刚持久化的聊天消息作为 WS 事件广播出去，
// 使同一账号的每个浏览器无需手动刷新即可追加该消息
// （design/16 §4.2）。它由 orchestrator（Web 轮次）与 channel bridge
// （IM 消息）共用，因此两条路径走同一种帧结构。事件类型按会话
// 区分：Web 主会话及其 Web 子会话广播 `message_persisted` 用于
// Web 聊天实时同步，而其他任何会话 id 都是隔离的渠道适配器会话
// （sess-xxx），广播 `channel_persisted` 供 IM 适配器日志页使用
// —— Web 聊天没有后者的处理器，因此在那里自然被忽略、永不污染
// 会话。Thinking 会被抹除，以与 WS 历史回放保持一致（design/16 §6）。
// 当 hub 接好广播时它是安全空操作。两种帧类型都携带消息内容，
// 因此 design/11 §7.1 帧门只把它们投递给 owner/admin 连接。
func (s *Server) BroadcastPersisted(msg history.Message) {
wire := toHistoryMessage(msg)
wire.Thinking = ""
@@ -33,18 +30,18 @@ func (s *Server) BroadcastPersisted(msg history.Message) {
s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: map[string]any{"message": wire}})
}
// wsCapacityAwaken is the WS event type for a capacity state transition
// (design/14 §5). Its data field is the frontend AwakenPayload {fromState,toState,tool}.
// wsCapacityAwaken 是能力状态迁移的 WS 事件类型（design/14 §5）。
// 其 data 字段是前端的 AwakenPayload {fromState,toState,tool}。
const wsCapacityAwaken = "capacity_awaken"
// BroadcastCapacityAwaken fans a capacity state transition to the Web UI as a
// `capacity_awaken` frame (design/14 §5). sessionID is the conversation that
// transitioned, so the client targets the correct session rather than treating the
// wake event as global. data matches the frontend AwakenPayload {fromState, toState,
// tool}; tool may be empty (no single tool was lit, the client then keeps the prior
// tool badge). It is NOT one of the design/11 §7.1 「状态机信号」 (its `tool` names
// what woke, and the panel that renders it is an owner surface), so the frame gate
// delivers it to owner/admin connections only.
// BroadcastCapacityAwaken 把一次能力状态迁移作为 `capacity_awaken` 帧
// 扇出到 Web UI（design/14 §5）。sessionID 是发生迁移的会话，因此
// 客户端能定位到正确的会话，而不会把唤醒事件当成全局事件。data 与
// 前端 AwakenPayload {fromState, toState, tool} 对应；tool 可以为空
// （没有点名单个工具，客户端此时保留原有的工具徽标）。它不属于
// design/11 §7.1 的「状态机信号」（其 `tool` 指明的是被什么唤醒，而
// 渲染它的面板属于 owner 界面），因此帧门只把它投递给 owner/admin
// 连接。
func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool string) {
s.hub.broadcast(wsEvent{
Type: wsCapacityAwaken,
@@ -57,21 +54,19 @@ func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool str
})
}
// wsMoodEvent is the WS event type carrying the persona's mood after a turn
// (design/07 §3 下行 `mood`; design/01 §2 情绪状态机). The frontend's existing
// `mood` handler (ws.ts) reads data.mood and shows the 心情 badge.
// wsMoodEvent 是携带角色轮次结束后心情的 WS 事件类型（design/07 §3
// 下行 `mood`；design/01 §2 情绪状态机）。前端既有的 `mood` 处理器
// （ws.ts）读取 data.mood 并显示心情徽标。
const wsMoodEvent = "mood"
// BroadcastMood fans one emotion change to the owner/admin Web clients as a `mood`
// frame (design/07 §3). Persona mood/affection is private state, so the design/11
// §7.1 frame gate keeps it off viewer connections. It is called only when the mood
// actually changed — the
// machinery in the composition root is the gate — so the badge does not flicker
// with an unchanged value on every turn. data matches what the frontend `mood`
// handler reads ({mood}); affection/personaId ride along as the honest extra
// detail (affection is the persona's current 好感度, personaId the persona whose
// mood moved) and are ignored by clients that do not render them. The mood is a
// persona-level value, not a per-session one, so no sessionId is attached.
// BroadcastMood 把一次情绪变化作为 `mood` 帧扇出到 owner/admin 的 Web
// 客户端（design/07 §3）。角色心情/好感度属于私有状态，因此 design/11
// §7.1 帧门让它远离 viewer 连接。它仅在心情确实发生变化时被调用
// —— 组合根中的机制就是那道门 —— 因此徽标不会因为每轮都出现未变化
// 的值而闪烁。data 与前端 `mood` 处理器读取的内容对应（{mood}）；
// affection/personaId 作为诚实的额外细节一同携带（affection 是该角色
// 当前好感度，personaId 是心情发生变化的角色），不渲染它们的客户端
// 会忽略之。心情是角色级而非会话级的值，因此不附加 sessionId。
func (s *Server) BroadcastMood(personaID, mood string, affection int) {
s.hub.broadcast(wsEvent{
Type: wsMoodEvent,
internal/services/gateway/extra_test.go
+98 -100
View File
@@ -36,13 +36,13 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
)
// cfgDefault builds the minimum config used across tests.
// cfgDefault 构造各测试共用的最小配置。
func cfgDefault() *config.Config {
return &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: testSecret}}
}
// buildEnv constructs a Server + httptest server from an explicit cfg/deps,
// defaulting Auth to a real JWTService and Bus to a real in-process bus.
// buildEnv 由显式的 cfg/deps 构造 Server + httptest server，Auth 默认为
// 真实 JWTService，Bus 默认为真实的进程内 bus。
func buildEnv(t *testing.T, cfg *config.Config, deps Deps) (*Server, *httptest.Server, *auth.JWTService) {
t.Helper()
if cfg == nil {
@@ -64,8 +64,8 @@ func buildEnv(t *testing.T, cfg *config.Config, deps Deps) (*Server, *httptest.S
return srv, ts, jwt
}
// buildServerOnly constructs a Server without an httptest server (used for
// start/shutdown lifecycle tests).
// buildServerOnly 构造不带 httptest server 的 Server（用于启动/关闭
// 生命周期测试）。
func buildServerOnly(t *testing.T, cfg *config.Config, deps Deps) *Server {
t.Helper()
srv, err := New(cfg, deps)
@@ -75,13 +75,13 @@ func buildServerOnly(t *testing.T, cfg *config.Config, deps Deps) *Server {
return srv
}
// mustJWT builds a real JWTService for a test secret.
// mustJWT 为测试密钥构造真实 JWTService。
func mustJWT(t *testing.T) *auth.JWTService {
t.Helper()
return auth.New(config.AuthConfig{JWTSecret: testSecret})
}
// signAny issues a token for sub under the given role via a real JWTService.
// signAny 经真实 JWTService 为 sub 签发指定角色的 token。
func signAny(t *testing.T, jwt *auth.JWTService, sub string, role auth.Role) string {
t.Helper()
tok, err := jwt.Sign(sub, role, time.Hour)
@@ -91,7 +91,7 @@ func signAny(t *testing.T, jwt *auth.JWTService, sub string, role auth.Role) str
return tok
}
// doRaw performs a request with a raw (already-encoded) body.
// doRaw 用原始（已编码）请求体发起请求。
func doRaw(t *testing.T, method, url, raw, token string) (*http.Response, []byte) {
t.Helper()
req, err := http.NewRequest(method, url, strings.NewReader(raw))
@@ -114,9 +114,9 @@ func doRaw(t *testing.T, method, url, raw, token string) (*http.Response, []byte
return resp, b
}
// --- Mocks ------------------------------------------------------------------
// --- 测试替身（mock） ---------------------------------------------------------
// stubAuth wraps a real JWTService but lets Sign/Verify fail on demand.
// stubAuth 包装真实 JWTService，但可按需让 Sign/Verify 失败。
type stubAuth struct {
jwt *auth.JWTService
signErr error
@@ -141,7 +141,7 @@ func (s *stubAuth) Authorize(required ...auth.Role) func(http.Handler) http.Hand
return s.jwt.Authorize(required...)
}
// stubCaps mimics CapabilityStore with a fixed list and a forced SetPolicy error.
// stubCaps 以固定列表和强制出错的 SetPolicy 模拟 CapabilityStore。
type stubCaps struct {
list []capreg.Capability
setErr error
@@ -167,7 +167,7 @@ func (s *stubCaps) SetPolicy(name string, p capreg.ToolPolicy) error {
return nil
}
// stubSessions mimics SessionStore.
// stubSessions 模拟 SessionStore。
type stubSessions struct {
list []session.Session
err error
@@ -175,20 +175,20 @@ type stubSessions struct {
func (s *stubSessions) List(ctx context.Context) ([]session.Session, error) { return s.list, s.err }
// stubPersonas mimics PersonaStore, returning domain characters that the gateway
// maps to PersonaSummary.
// stubPersonas 模拟 PersonaStore，返回 gateway 会映射为 PersonaSummary
// 的领域角色对象。
type stubPersonas struct {
list []personareg.Character
}
func (s *stubPersonas) List() []personareg.Character { return s.list }
// Persona satisfies PersonaStore; tests don't rely on visual assets, so it
// always reports no persona found (empty visuals, safe fallback).
// Persona 满足 PersonaStore；测试不依赖视觉素材，因此它总是报告未找到
// 角色（空视觉，安全回退）。
func (s *stubPersonas) Persona(string) (*loader.Persona, bool) { return nil, false }
// stubMemories mimics MemoryStore, returning domain memory items that the gateway
// maps to MemorySummary.
// stubMemories 模拟 MemoryStore，返回 gateway 会映射为 MemorySummary
// 的领域记忆条目。
type stubMemories struct {
list []memory.MemoryItem
err error
@@ -196,8 +196,8 @@ type stubMemories struct {
func (s *stubMemories) List(ctx context.Context) ([]memory.MemoryItem, error) { return s.list, s.err }
// stubRag mimics RagStore, returning domain docs/hits the gateway maps to its
// admin wire shape.
// stubRag 模拟 RagStore，返回 gateway 会映射为其 admin 传输结构的
// 领域文档/命中。
type stubRag struct {
docs []knowledge.Doc
hits []knowledge.Hit
@@ -211,7 +211,7 @@ func (s *stubRag) Search(ctx context.Context, query string) ([]knowledge.Hit, er
func (s *stubRag) IndexDocument(ctx context.Context, doc knowledge.Doc) error { return s.err }
func (s *stubRag) DeleteDocument(ctx context.Context, id string) error { return s.err }
// stubIoT mimics IoTStore.
// stubIoT 模拟 IoTStore。
type stubIoT struct {
list []iot.Device
err error
@@ -219,7 +219,7 @@ type stubIoT struct {
func (s *stubIoT) List(ctx context.Context) ([]iot.Device, error) { return s.list, s.err }
// stubAuditor mimics SubagentAuditor for the subagent_ask route.
// stubAuditor 为 subagent_ask 路由模拟 SubagentAuditor。
type stubAuditor struct {
res *orchestration.AuditResult
err error
@@ -229,7 +229,7 @@ func (s *stubAuditor) AuditContext(ctx context.Context, id string, mode orchestr
return s.res, s.err
}
// stubTaskOutputs mimics TaskOutputStore for the task_output_query route.
// stubTaskOutputs 为 task_output_query 路由模拟 TaskOutputStore。
type stubTaskOutputs struct {
ev task.OutputEvent
err error
@@ -239,7 +239,7 @@ func (s *stubTaskOutputs) TaskOutput(ctx context.Context, id string) (task.Outpu
return s.ev, s.err
}
// stubChannel implements sdk.Channel.
// stubChannel 实现 sdk.Channel。
type stubChannel struct{}
func (stubChannel) Name() string { return "qq" }
@@ -250,8 +250,8 @@ func (stubChannel) HealthCheck(ctx context.Context) error { return nil }
var _ sdk.Channel = stubChannel{}
// stubBotChannel mimics a live kook adapter that resolved its bot identity via
// GET /api/v3/user/me on Connect (surfaced through Config()).
// stubBotChannel 模拟一个在 Connect 时经 GET /api/v3/user/me 解析出
// 机器人身份的在线 kook 适配器（经 Config() 暴露）。
type stubBotChannel struct{}
func (stubBotChannel) Name() string { return "kook:conf0" }
@@ -265,7 +265,7 @@ func (stubBotChannel) Config() map[string]any {
var _ sdk.Channel = stubBotChannel{}
// captureRunner records the sessionID each Run call sees.
// captureRunner 记录每次 Run 调用看到的 sessionID。
type captureRunner struct {
got []string
res *loop.Result
@@ -277,14 +277,14 @@ func (c *captureRunner) Run(ctx context.Context, sessionID string, messages []mo
return c.res, c.err
}
// plainWriter is a bare ResponseWriter implementing neither Flusher nor Hijacker.
// plainWriter 是既未实现 Flusher 也未实现 Hijacker 的裸 ResponseWriter。
type plainWriter struct{}
func (plainWriter) Header() http.Header { return http.Header{} }
func (plainWriter) Write(b []byte) (int, error) { return len(b), nil }
func (plainWriter) WriteHeader(code int) {}
// --- Server construction / validation --------------------------------------
// --- Server 构造 / 校验 -------------------------------------------------------
func TestNewValidation(t *testing.T) {
t.Parallel()
@@ -297,7 +297,7 @@ func TestNewValidation(t *testing.T) {
}
}
// --- Login ------------------------------------------------------------------
// --- 登录 --------------------------------------------------------------------
func TestLoginBadBody(t *testing.T) {
t.Parallel()
@@ -321,7 +321,7 @@ func TestLoginSignError(t *testing.T) {
wantErrorShape(t, body)
}
// --- Chat -------------------------------------------------------------------
// --- 聊天 --------------------------------------------------------------------
func TestChatBadBody(t *testing.T) {
t.Parallel()
@@ -356,8 +356,8 @@ func TestChatRequestPersistsMode(t *testing.T) {
History: hist,
})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
// The REST body carries the client's behavior mode (design/16 §5.1); the
// orchestrator persists both the user turn and the assistant reply under it.
// REST 请求体携带客户端的行为模式（design/16 §5.1）；orchestrator 在
// 它之下同时持久化用户轮次与助手回复。
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat",
map[string]any{"sessionId": "web", "text": "改代码", "mode": "code"}, tok)
if resp.StatusCode != http.StatusOK {
@@ -400,7 +400,7 @@ func TestChatRunnerError(t *testing.T) {
wantErrorShape(t, body)
}
// --- List handlers ---------------------------------------------------------
// --- 列表处理器 --------------------------------------------------------------
func TestListSessionsNonEmpty(t *testing.T) {
t.Parallel()
@@ -523,7 +523,7 @@ func TestAdminMemoriesErrorStillOK(t *testing.T) {
}
}
// --- Admin: IoT ------------------------------------------------------------
// --- Admin：IoT --------------------------------------------------------------
func TestAdminIoTNonEmpty(t *testing.T) {
t.Parallel()
@@ -568,7 +568,7 @@ func TestAdminIoTError(t *testing.T) {
wantErrorShape(t, body)
}
// --- Admin: channels -------------------------------------------------------
// --- Admin：渠道 -------------------------------------------------------------
func TestAdminChannelsFromRegistry(t *testing.T) {
t.Parallel()
@@ -576,7 +576,7 @@ func TestAdminChannelsFromRegistry(t *testing.T) {
if err := env.caps.RegisterChannel(stubChannel{}); err != nil {
t.Fatalf("register channel: %v", err)
}
// A channel-kind capability whose Value is not an sdk.Channel must be skipped.
// Value 不是 sdk.Channel 的 channel 类能力必须被跳过。
if err := env.caps.Register(capreg.Capability{Name: "badchan", Kind: capreg.KindChannel, Value: stubTool{}}); err != nil {
t.Fatalf("register bad channel: %v", err)
}
@@ -597,9 +597,9 @@ func TestAdminChannelsFromRegistry(t *testing.T) {
}
}
// TestAdminChannels_SurfacesBotIdentity verifies a connected kook adapter's
// resolved bot identity (from user/me) reaches the accounts of GET /admin/channels
// so the panel can show who the bot is.
// TestAdminChannels_SurfacesBotIdentity 验证已连接 kook 适配器解析出的
// 机器人身份（来自 user/me）能到达 GET /admin/channels 的 accounts，
// 使面板能展示机器人是谁。
func TestAdminChannels_SurfacesBotIdentity(t *testing.T) {
t.Parallel()
caps := capreg.New()
@@ -637,11 +637,11 @@ func TestAdminChannels_SurfacesBotIdentity(t *testing.T) {
}
}
// --- Admin: policy errors --------------------------------------------------
// --- Admin：策略错误 ---------------------------------------------------------
func TestAdminSetPolicyUnavailable(t *testing.T) {
t.Parallel()
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{}) // no Registry
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{}) // 不接 Registry
_ = srv
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/plugins/policy", map[string]any{"name": "echo"}, tok)
@@ -686,7 +686,7 @@ func TestAdminSetPolicyMissingName(t *testing.T) {
wantErrorShape(t, body)
}
// --- CORS ------------------------------------------------------------------
// --- 跨域（CORS） -------------------------------------------------------------
func TestCORSAllowAll(t *testing.T) {
t.Parallel()
@@ -748,7 +748,7 @@ func TestCORSPreflight(t *testing.T) {
}
}
// --- Recovery + statusRecorder --------------------------------------------
// --- panic 恢复 + statusRecorder ---------------------------------------------
func TestRecoverPanicTo500(t *testing.T) {
t.Parallel()
@@ -778,17 +778,17 @@ func TestStatusRecorderHijackUnsupported(t *testing.T) {
func TestStatusRecorderFlush(t *testing.T) {
t.Parallel()
// Flusher present → forwarded.
// 有 Flusher → 转发。
rr := httptest.NewRecorder()
(&statusRecorder{ResponseWriter: rr}).Flush()
if !rr.Flushed {
t.Fatal("Flush should have been forwarded to the underlying Flusher")
}
// No Flusher → no-op, no panic.
// 无 Flusher → 空操作，不 panic。
(&statusRecorder{ResponseWriter: plainWriter{}}).Flush()
}
// --- SPA -------------------------------------------------------------------
// --- 单页应用（SPA） ---------------------------------------------------------
func TestSPA(t *testing.T) {
t.Parallel()
@@ -799,29 +799,29 @@ func TestSPA(t *testing.T) {
srv, ts, _ := buildEnv(t, cfgDefault(), Deps{Frontend: front})
_ = srv
// Real asset is served.
// 真实静态资源被正常服务。
resp, body := doReq(t, http.MethodGet, ts.URL+"/assets/app.js", nil, "")
if resp.StatusCode != http.StatusOK || string(body) != "var x = 1;" {
t.Fatalf("asset status = %d body = %q, want 200 with js", resp.StatusCode, body)
}
// Unknown non-API path falls back to index.html.
// 未知的非 API 路径回退到 index.html。
resp, body = doReq(t, http.MethodGet, ts.URL+"/chat/somewhere", nil, "")
if resp.StatusCode != http.StatusOK || !strings.Contains(string(body), "root") {
t.Fatalf("fallback status = %d body = %q, want 200 with index", resp.StatusCode, body)
}
// Root serves index.html.
// 根路径提供 index.html。
resp, body = doReq(t, http.MethodGet, ts.URL+"/", nil, "")
if resp.StatusCode != http.StatusOK || !strings.Contains(string(body), "root") {
t.Fatalf("root status = %d body = %q, want 200 with index", resp.StatusCode, body)
}
// API route is passed through untouched.
// API 路由原样透传。
resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/health", nil, "")
if resp.StatusCode != http.StatusOK {
t.Fatalf("api status = %d, want 200 (body=%s)", resp.StatusCode, body)
}
}
// --- WS: validation --------------------------------------------------------
// --- WS：校验 ----------------------------------------------------------------
func TestWSRequiresGET(t *testing.T) {
t.Parallel()
@@ -892,7 +892,7 @@ func TestServerCloseShutsHub(t *testing.T) {
conn.Close()
}
// --- WS: hub unit tests ----------------------------------------------------
// --- WS：hub 单元测试 --------------------------------------------------------
func TestHubStreamChat(t *testing.T) {
t.Parallel()
@@ -919,9 +919,8 @@ func TestHubStreamChatPersistsMode(t *testing.T) {
},
History: hist,
})
// The ws.chat payload carries the client's behavior mode (design/16 §5.1);
// streamChat threads it into the orchestrator, which persists both the user
// turn and the assistant reply under it.
// ws.chat 载荷携带客户端的行为模式（design/16 §5.1）；streamChat 把它
// 串进 orchestrator，后者在它之下持久化用户轮次与助手回复二者。
hb.streamChat(context.Background(), map[string]any{"sessionId": "web", "text": "改代码", "mode": "code"})
deadline := time.Now().Add(2 * time.Second)
@@ -946,11 +945,11 @@ func TestHubBroadcastStream(t *testing.T) {
c := newOwnerConn(hb, 32)
hb.register(c)
// A text_delta StreamEvent on the OnStream→bus→relay path becomes a typed
// message.delta frame carrying the delta text and a monotonically increasing
// seq, leaving the legacy stream_chunk shape out of the way. The loop stamps
// the session/turn identity on the event and the hub must carry it onto the
// frame, so a client can route the delta to the active session (design/16 §4.2).
// OnStream→bus→relay 路径上的 text_delta StreamEvent 变成带类型的
// message.delta 帧，携带增量文本与单调递增的 seq，让旧的
// stream_chunk 形态不再参与。loop 把会话/轮次身份盖在事件上，
// hub 必须把它带到帧上，使客户端能把增量路由到活跃会话
// （design/16 §4.2）。
hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel", Index: 0, SessionID: "web", TurnID: "turn-abc"})
ev := readFakeFrame(t, c)
if ev.Type != "message.delta" {
@@ -967,7 +966,7 @@ func TestHubBroadcastStream(t *testing.T) {
t.Fatalf("frame scope = session %q turn %q, want session web turn turn-abc", ev.SessionID, ev.TurnID)
}
// A tool_call event maps to tool_call.*; a non-StreamEvent payload is dropped.
// tool_call 事件映射为 tool_call.*；非 StreamEvent 的载荷被丢弃。
hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindToolCallEnd, ID: "toolu_1", Index: 0})
ev = readFakeFrame(t, c)
if ev.Type != "tool_call.end" {
@@ -999,11 +998,10 @@ func TestServerBroadcastCapacityAwaken(t *testing.T) {
}
}
// TestServerBroadcastPersistedSplit verifies BroadcastPersisted routes the event
// type by session: the main web session and its web sub-sessions broadcast
// `message_persisted` (web chat realtime sync), while an isolated channel adapter
// session (sess-xxx) broadcasts `channel_persisted` (IM adapter log page) so it is
// naturally ignored by the web chat handler.
// TestServerBroadcastPersistedSplit 验证 BroadcastPersisted 按会话路由事件
// 类型：Web 主会话及其 Web 子会话广播 `message_persisted`（Web 聊天实时
// 同步），而隔离的渠道适配器会话（sess-xxx）广播 `channel_persisted`
// （IM 适配器日志页），因此它被 Web 聊天处理器自然忽略。
func TestServerBroadcastPersistedSplit(t *testing.T) {
t.Parallel()
env := newEnv(t, nil, false)
@@ -1049,7 +1047,7 @@ func TestServerBroadcastPersistedSplit(t *testing.T) {
func TestHubStreamChatIdle(t *testing.T) {
t.Parallel()
hb := newTestHub(nil, nil) // nil runner
hb := newTestHub(nil, nil) // runner 为 nil
c := newOwnerConn(hb, 32)
hb.register(c)
@@ -1080,7 +1078,7 @@ func TestHubRelayChatBranch(t *testing.T) {
if err := hb.relay(context.Background(), bus.Event{Subject: wsChatSubject, Payload: map[string]any{"sessionId": "s1", "text": "hi"}}); err != nil {
t.Fatalf("relay: %v", err)
}
// relay runs streamChat in a goroutine; read with a deadline.
// relay 在 goroutine 中跑 streamChat；读取时带超时。
if ev := readFakeFrame(t, c); ev.Type != "stream_chunk" {
t.Fatalf("frame type = %q, want stream_chunk (frame=%+v)", ev.Type, ev)
}
@@ -1102,12 +1100,12 @@ func TestHubRoute(t *testing.T) {
c := newOwnerConn(hb, 32)
hb.register(c)
// subagent_ask → subagent_event (real audit data, not a hardcoded pending).
// subagent_ask → subagent_event（真实审计数据，不是硬编码的 pending）。
c.route(wsInbound{Type: "subagent_ask", Data: json.RawMessage(`{"id":"sub-1"}`)})
if ev := readFakeFrame(t, c); ev.Type != "subagent_event" {
t.Fatalf("subagent frame type = %q, want subagent_event", ev.Type)
}
// subagent_ask with no auditor → clear error, never a fabricated pending.
// 无 auditor 的 subagent_ask → 明确报错，绝不伪造 pending。
hb.auditor = nil
c.route(wsInbound{Type: "subagent_ask", Data: json.RawMessage(`{"id":"sub-1"}`)})
if ev := readFakeFrame(t, c); ev.Type != "error" {
@@ -1115,12 +1113,12 @@ func TestHubRoute(t *testing.T) {
}
hb.auditor = &stubAuditor{res: &orchestration.AuditResult{}}
// task_output_query → task_output with real output.
// task_output_query → 带真实输出的 task_output。
c.route(wsInbound{Type: "task_output_query", Data: json.RawMessage(`{"taskId":"task-1"}`)})
if ev := readFakeFrame(t, c); ev.Type != "task_output" {
t.Fatalf("task frame type = %q, want task_output", ev.Type)
}
// task_output_query with no task store → clear error, never an empty {}.
// 无 task store 的 task_output_query → 明确报错，绝不返回空的 {}。
hb.taskOutputs = nil
c.route(wsInbound{Type: "task_output_query", Data: json.RawMessage(`{"taskId":"task-1"}`)})
if ev := readFakeFrame(t, c); ev.Type != "error" {
@@ -1128,24 +1126,24 @@ func TestHubRoute(t *testing.T) {
}
hb.taskOutputs = &stubTaskOutputs{ev: task.OutputEvent{ID: "task-1", Status: task.StatusDone, Output: task.Output{Stdout: "out"}}}
// voice_input with no voice service → clear error, never a crash or a drop.
// 无语音服务的 voice_input → 明确报错，绝不崩溃也不丢弃。
c.route(wsInbound{Type: "voice_input", Data: json.RawMessage(`{"audio":"x"}`)})
if ev := readFakeFrame(t, c); ev.Type != "error" {
t.Fatalf("no-voice voice_input frame type = %q, want error", ev.Type)
}
// unknown → error (enqueueJSON).
// 未知类型 → 报错（enqueueJSON）。
c.route(wsInbound{Type: "nope"})
if ev := readFakeFrame(t, c); ev.Type != "error" {
t.Fatalf("unknown frame type = %q, want error", ev.Type)
}
// chat → publishes to bus (publishPublic non-nil path); no frame enqueued.
// chat → 发布到 bus（publishPublic 非 nil 路径）；不入队任何帧。
c.route(wsInbound{Type: "chat", Data: json.RawMessage(`{"text":"hi"}`)})
if err := connHasNoFrame(c); err != nil {
t.Fatalf("chat should not enqueue a frame directly: %v", err)
}
}
// stubVoice is a deterministic VoiceService for voice route tests.
// stubVoice 是供语音路由测试使用的确定性 VoiceService。
type stubVoice struct {
sttText string
sttErr error
@@ -1164,9 +1162,9 @@ func TestHubVoiceInputTurn(t *testing.T) {
c := newOwnerConn(hb, 32)
hb.register(c)
// "aGVsbG8=" decodes to "hello" — a valid base64 audio payload.
// "aGVsbG8=" 解码为 "hello" —— 合法的 base64 音频载荷。
c.route(wsInbound{Type: "voice_input", Data: json.RawMessage(`{"audio":"aGVsbG8=","format":"wav"}`)})
// runVoiceTurn runs in a goroutine; read with a deadline.
// runVoiceTurn 在 goroutine 中运行；读取时带超时。
ev := readFakeFrame(t, c)
if ev.Type != "voice_output" {
t.Fatalf("voice frame type = %q, want voice_output (frame=%+v)", ev.Type, ev)
@@ -1218,7 +1216,7 @@ func TestHubBridgeSubagentProgress(t *testing.T) {
t.Fatalf("type = %q, want subagent_event (frame=%+v)", ev.Type, ev)
}
// An internal ask subject must never surface to clients.
// 内部的 ask 主题绝不可暴露给客户端。
if err := hb.relay(context.Background(), bus.Event{
Subject: "cyrene.agent.a.b.ask",
Payload: comm.Event{Name: orchestration.EventDone, Data: orchestration.Progress{AgentID: "sub-1"}},
@@ -1232,9 +1230,9 @@ func TestHubBridgeSubagentProgress(t *testing.T) {
func TestPublishPublicNilBus(t *testing.T) {
t.Parallel()
hb := newTestHub(nil, nil) // nil bus
hb := newTestHub(nil, nil) // bus 为 nil
c := newOwnerConn(hb, 32)
// Must not panic and must not enqueue anything.
// 不得 panic，也不得入队任何东西。
c.publishPublic(wsChatSubject, map[string]any{"text": "hi"})
}
@@ -1272,7 +1270,7 @@ func TestOriginCheckerSameHost(t *testing.T) {
allowed := map[string]bool{"http://ok.example": true}
checker := originChecker(allowed, false)
// Same host (exact) — production mode serving embedded frontend on the gateway port.
// 同主机（精确匹配）—— 生产模式下在网关端口提供内嵌前端。
req := httptest.NewRequest(http.MethodGet, "/ws/chat", nil)
req.Host = "localhost:10942"
req.Header.Set("Origin", "http://localhost:10942")
@@ -1280,7 +1278,7 @@ func TestOriginCheckerSameHost(t *testing.T) {
t.Fatal("same host origin should be allowed")
}
// Loopback alias: host is 127.0.0.1 but client origin uses localhost.
// 回环别名：host 是 127.0.0.1，但客户端 origin 用的是 localhost。
req2 := httptest.NewRequest(http.MethodGet, "/ws/chat", nil)
req2.Host = "127.0.0.1:10942"
req2.Header.Set("Origin", "http://localhost:10942")
@@ -1288,7 +1286,7 @@ func TestOriginCheckerSameHost(t *testing.T) {
t.Fatal("localhost<->127.0.0.1 loopback alias should be allowed")
}
// Different host must still be rejected (not a same-origin, not whitelisted).
// 不同主机仍必须被拒（既非同源，也不在白名单）。
req3 := httptest.NewRequest(http.MethodGet, "/ws/chat", nil)
req3.Host = "evil.example:10942"
req3.Header.Set("Origin", "http://localhost:10942")
@@ -1296,7 +1294,7 @@ func TestOriginCheckerSameHost(t *testing.T) {
t.Fatal("cross-host origin should be rejected")
}
// Same host, different port: still same-origin (same physical host), so allow.
// 同主机、不同端口：仍是同源（同一物理主机），因此放行。
req4 := httptest.NewRequest(http.MethodGet, "/ws/chat", nil)
req4.Host = "localhost:8080"
req4.Header.Set("Origin", "http://localhost:10942")
@@ -1320,7 +1318,7 @@ func TestWSTokenExtraction(t *testing.T) {
}
}
// --- Start / ListenAndServe lifecycle -------------------------------------
// --- Start / ListenAndServe 生命周期 ------------------------------------------
func TestStartAndShutdown(t *testing.T) {
t.Parallel()
@@ -1344,7 +1342,7 @@ func TestStartAndShutdown(t *testing.T) {
}
}
// --- Helpers ---------------------------------------------------------------
// --- 辅助函数 ----------------------------------------------------------------
func doHandler(t *testing.T, req *http.Request) *http.Response {
t.Helper()
@@ -1396,11 +1394,11 @@ func newFakeConn(hb *hub, buf int) *wsConn {
}
}
// newOwnerConn is a fake connection carrying owner claims: the design/11 §7.1 frame
// gate sends every content frame (message.*, thinking.*, tool_*, graph_update, mood,
// persisted-message frames, …) to owner/admin only, so a hub unit test that reads
// one has to connect as the role it is gated to. A nil-claims conn (newFakeConn) is
// the fail-closed bare connection — always reachable by the state signals only.
// newOwnerConn 是携带 owner claims 的假连接：design/11 §7.1 帧门把每个
// 内容帧（message.*、thinking.*、tool_*、graph_update、mood、持久化消息
// 帧，……）只发给 owner/admin，因此读取其中之一的 hub 单元测试必须以
// 被放行的角色连接。claims 为 nil 的连接（newFakeConn）是 fail-closed
// 的裸连接 —— 仅状态信号总能到达。
func newOwnerConn(hb *hub, buf int) *wsConn {
c := newFakeConn(hb, buf)
c.claims = &auth.Claims{Role: auth.RoleOwner}
@@ -1441,7 +1439,7 @@ func reqWithQuery(k, v string) *http.Request {
return httptest.NewRequest(http.MethodGet, "/ws/chat?"+k+"="+v, nil)
}
// --- RAG knowledge base ----------------------------------------------------
// --- RAG 知识库 --------------------------------------------------------------
func TestAdminRagListsAndSearches(t *testing.T) {
t.Parallel()
@@ -1451,7 +1449,7 @@ func TestAdminRagListsAndSearches(t *testing.T) {
_ = srv
tok := signAny(t, jwt, "acct", auth.RoleOwner)
// List only.
// 仅列表。
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok)
if resp.StatusCode != http.StatusOK {
t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body)
@@ -1470,7 +1468,7 @@ func TestAdminRagListsAndSearches(t *testing.T) {
t.Fatalf("hits = %+v, want omitted without query", list.Hits)
}
// List + search via ?query=.
// 列表 + 经 ?query= 检索。
resp2, body2 := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag?query=rag", nil, tok)
if resp2.StatusCode != http.StatusOK {
t.Fatalf("search status = %d, want 200 (body=%s)", resp2.StatusCode, body2)
@@ -1516,7 +1514,7 @@ func TestAdminRagEmptyWhenUnwired(t *testing.T) {
_ = srv
tok := signAny(t, jwt, "acct", auth.RoleOwner)
// List degrades to empty (never a 500).
// 列表降级为空（绝不 500）。
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok)
if resp.StatusCode != http.StatusOK {
t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body)
@@ -1531,7 +1529,7 @@ func TestAdminRagEmptyWhenUnwired(t *testing.T) {
t.Fatalf("documents = %+v, want empty", list.Documents)
}
// Query reports a clear 501 (backend not implemented).
// 检索明确返回 501（后端未实现）。
resp2, body2 := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/query", map[string]any{"query": "rag"}, tok)
if resp2.StatusCode != http.StatusNotImplemented {
t.Fatalf("query status = %d, want 501 (body=%s)", resp2.StatusCode, body2)
@@ -1542,8 +1540,8 @@ func TestAdminRagEmptyWhenUnwired(t *testing.T) {
}
}
// TestAdminRagDeleteUnsupportedStore501 asserts a store that cannot delete is
// answered with an honest 501 rather than a fabricated success.
// TestAdminRagDeleteUnsupportedStore501 断言无法删除的存储会被诚实地以
// 501 应答，而不是伪造成功。
func TestAdminRagDeleteUnsupportedStore501(t *testing.T) {
t.Parallel()
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{err: knowledge.ErrWriteUnsupported}})
internal/services/gateway/history_write_broadcast_test.go
+16 -17
View File
@@ -10,15 +10,15 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)
// These tests cover the D3 realtime-sync contract: a successful message edit or
// delete must reach the account's other browsers as a WS frame, not only the client
// that issued the REST call (otherwise every other browser has to re-read to converge).
// 这些测试覆盖 D3 实时同步契约：一次成功的消息编辑或删除必须以
// WS 帧到达该账号的其他浏览器，而不仅发起 REST 调用的客户端
// （否则其他每个浏览器都得重读才能收敛）。
// TestPatchMessageBroadcastsEdited covers ①: PATCH /api/v1/messages/{id} answers 200
// and pushes exactly one `message_edited` frame scoped to the edited message's own
// session, carrying the new content (and the mode/createdAt a client renders with).
// The frame's data is the updated message DTO with thinking stripped, matching the
// message_persisted contract; the REST reply keeps thinking (owner/admin-only read).
// TestPatchMessageBroadcastsEdited 覆盖 ①：PATCH /api/v1/messages/{id}
// 返回 200，并恰好推送一个 `message_edited` 帧，其作用域为被编辑消息
// 自身的会话，携带新内容（以及客户端渲染所用的 mode/createdAt）。该帧
// 的 data 是抹除了 thinking 的更新后消息 DTO，与 message_persisted
// 契约一致；REST 回复保留 thinking（仅 owner/admin 可读）。
func TestPatchMessageBroadcastsEdited(t *testing.T) {
t.Parallel()
ctx := context.Background()
@@ -51,7 +51,7 @@ func TestPatchMessageBroadcastsEdited(t *testing.T) {
if resp.StatusCode != http.StatusOK {
t.Fatalf("patch status = %d, want 200 (body=%s)", resp.StatusCode, body)
}
// The REST reply is the full DTO for the issuing client.
// REST 回复是发给发起方客户端的完整 DTO。
var reply historyMessage
if err := json.Unmarshal(decodeBody(t, body)["data"], &reply); err != nil {
t.Fatalf("decode reply: %v (body=%s)", err, body)
@@ -93,10 +93,9 @@ func TestPatchMessageBroadcastsEdited(t *testing.T) {
}
}
// TestDeleteMessageBroadcastsDeleted covers ②: DELETE /api/v1/messages/{id} answers
// 200, really removes the row, and pushes one `message_deleted` frame carrying the id
// plus the owning session (read before the row disappears) so a remote browser can
// drop it without re-reading.
// TestDeleteMessageBroadcastsDeleted 覆盖 ②：DELETE /api/v1/messages/{id}
// 返回 200，真正删除该行，并推送一个 `message_deleted` 帧，携带 id 以及
// 所属会话（在该行消失前读出），使远端浏览器无需重读即可移除它。
func TestDeleteMessageBroadcastsDeleted(t *testing.T) {
t.Parallel()
ctx := context.Background()
@@ -133,7 +132,7 @@ func TestDeleteMessageBroadcastsDeleted(t *testing.T) {
t.Fatal(err)
}
// The broadcast must not claim a delete that did not happen.
// 广播不得宣称一次并未发生的删除。
remaining, err := st.List(ctx, sub, 0, 50, "")
if err != nil {
t.Fatalf("list after delete: %v", err)
@@ -145,9 +144,9 @@ func TestDeleteMessageBroadcastsDeleted(t *testing.T) {
}
}
// TestDeleteMessageUnknownIDBroadcastsNothing pins the failure side: a 404 delete
// (unknown id) must not push a `message_deleted` frame, or a client would drop a
// message that still exists.
// TestDeleteMessageUnknownIDBroadcastsNothing 钉住失败侧：404 的删除
// （未知 id）不得推送 `message_deleted` 帧，否则客户端会丢掉一条仍然
// 存在的消息。
func TestDeleteMessageUnknownIDBroadcastsNothing(t *testing.T) {
t.Parallel()
srv, ts, jwt, _ := newHistoryEnv(t)
internal/services/gateway/mood_test.go
+13 -14
View File
@@ -11,8 +11,8 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// moodRunner is a TurnRunner that settles one turn with a fixed result, so the
// test can assert the mood the loop reported is what gets persisted.
// moodRunner 是一个 TurnRunner，以固定结果结算一轮，使测试能断言
// 被持久化的正是 loop 报告的心情。
type moodRunner struct {
mood string
}
@@ -21,12 +21,11 @@ func (m moodRunner) Run(_ context.Context, _ string, _ []model.Message, _ loop.S
return &loop.Result{Text: "在的呀♪", Steps: 1, StopReason: "stop", Mood: m.mood}, nil
}
// TestOrchestratorPersistsTurnMood locks the mood column end to end on the Web
// path: the mood the loop's turn-boundary hook reported lands on the persisted
// assistant message (design/16 §3.3 `mode/mood/thinking`), so the DTO,
// `message_persisted` and history replay carry real data instead of an always
// empty mood. The user message keeps an empty mood (design/16 §3.3 assigns the
// mood to the assistant reply).
// TestOrchestratorPersistsTurnMood 在 Web 路径上端到端锁住 mood 列：
// loop 轮次边界钩子报告的心情落到持久化的助手消息上
// （design/16 §3.3 `mode/mood/thinking`），因此 DTO、`message_persisted`
// 与历史回放携带真实数据，而不是永远为空的心情。用户消息保持空
// 心情（design/16 §3.3 把心情分配给助手回复）。
func TestOrchestratorPersistsTurnMood(t *testing.T) {
t.Parallel()
hist := &fakeHistory{}
@@ -56,9 +55,9 @@ func TestOrchestratorPersistsTurnMood(t *testing.T) {
}
}
// TestOrchestratorPersistsEmptyMoodWhenUntracked locks the honest degrade: with no
// emotion tracking (no hook wired, so Result.Mood is empty) the stored mood stays
// empty rather than being fabricated.
// TestOrchestratorPersistsEmptyMoodWhenUntracked 锁住诚实的降级：没有
// 情绪跟踪时（未接钩子，因此 Result.Mood 为空）存储的心情保持为空，
// 而不是被编造。
func TestOrchestratorPersistsEmptyMoodWhenUntracked(t *testing.T) {
t.Parallel()
hist := &fakeHistory{}
@@ -79,9 +78,9 @@ func TestOrchestratorPersistsEmptyMoodWhenUntracked(t *testing.T) {
}
}
// TestServerBroadcastMood locks the `mood` frame the frontend's mood handler reads
// (design/07 §3): the type is `mood`, data.mood is the label, and the persona id /
// affection ride along as extra detail.
// TestServerBroadcastMood 锁住前端 mood 处理器读取的 `mood` 帧
// （design/07 §3）：类型是 `mood`，data.mood 是标签，角色 id / 好感度
// 作为额外细节一同携带。
func TestServerBroadcastMood(t *testing.T) {
t.Parallel()
env := newEnv(t, nil, false)
internal/services/gateway/server.go
+47 -54
View File
@@ -1,24 +1,21 @@
// Package gateway implements the Cyrene v2 HTTP gateway: the only socket the
// single binary opens. It serves the versioned REST API (/api/v1), the realtime
// WebSocket hub (/ws/chat) and the RBAC-protected admin API (/api/v1/admin/*)
// all on one port (:8080).
// Package gateway 实现 Cyrene v2 的 HTTP 网关：单二进制唯一打开的套接字。
// 它在同一个端口（:8080）上提供版本化 REST API（/api/v1）、实时
// WebSocket hub（/ws/chat）以及受 RBAC 保护的管理 API（/api/v1/admin/*）。
//
// It is deliberately lean. All backend capability flows through injected
// dependencies (construction-time dependency injection) and the in-process bus,
// never through cross-process IPC, so the package never opens a listener of its
// own beyond the gateway port.
// 它是刻意精简的。所有后端能力都经由注入的依赖（构造期依赖注入）和
// 进程内 bus 流动，绝不跨进程 IPC，因此本包除网关端口外永不自行
// 打开监听器。
//
// Client contract (DevDocs/02-development-constraints.md §九, §十一):
// - REST responses are {data} on success and {error:{code,message}} on error.
// - Authentication is JWT bearer (internal/auth); authorization is the
// owner>admin>viewer RBAC hierarchy (internal/auth).
// - Admin writes are audited (who/what/when/result).
// - Realtime events follow DevDocs/design/07 §3 (WS protocol).
// 客户端契约（DevDocs/02-development-constraints.md §九、§十一）：
// - REST 成功响应为 {data}，错误响应为 {error:{code,message}}。
// - 认证是 JWT bearer（internal/auth）；授权是 owner>admin>viewer 的
// RBAC 层级（internal/auth）。
// - 管理写操作有审计（谁/做了什么/何时/结果）。
// - 实时事件遵循 DevDocs/design/07 §3（WS 协议）。
//
// Layout: contracts.go declares the injected dependency interfaces, deps.go the
// Deps bundle, server.go the Server lifecycle (construct / serve / close),
// routes.go the route table + outer middleware, broadcast.go the outbound
// realtime pushes.
// 文件布局：contracts.go 声明注入的依赖接口，deps.go 声明 Deps 组合，
// server.go 是 Server 生命周期（构造 / 服务 / 关闭），routes.go 是路由表
// 加外层中间件，broadcast.go 是下行实时推送。
package gateway
import (
@@ -40,11 +37,11 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// source is the log provenance tag for every record this package emits.
// source 是本包每条日志记录的来源标签。
const source = "gateway"
// Server is the gateway. It owns a ServeMux aggregating REST + WS + admin and a
// WebSocket hub. It is safe for concurrent use once constructed.
// Server 是网关本体。它持有聚合 REST + WS + admin 的 ServeMux 与一个
// WebSocket hub。构造完成后可安全并发使用。
type Server struct {
cfg *config.Config
auth Authenticator
@@ -68,32 +65,30 @@ type Server struct {
streaming bool
frontend fs.FS
dbPath string
// dataDir is the resolved data directory backing the upload store
// (<dataDir>/uploads). Injected by the composition root; cfg.DataDir is the
// fallback. See Deps.DataDir.
// dataDir 是支撑上传存储的已解析数据目录（<dataDir>/uploads）。
// 由组合根注入；cfg.DataDir 是回退值。见 Deps.DataDir。
dataDir string
// llm is the hot-swappable provider stack (models.json). When nil the
// admin /llm endpoints fall back to the env config. It is optional so tests
// that do not wire it keep working.
// llm 是可热切换的提供方栈（models.json）。为 nil 时 admin 的 /llm
// 端点回退到环境变量配置。它是可选的，以便未接它的测试仍能工作。
llm *model.RuntimeLLM
channels ChannelConfigStore
// plugins is the runtime-installable external tool provider manager. See
// Deps.Plugins.
// plugins 是可在运行时安装的外部工具提供方管理器。见
// Deps.Plugins。
plugins *pluginmgr.Manager
// ownerRegistrar stamps a scan-login bound owner id into the owner whitelist
// (best-effort). See Deps.OwnerRegistrar.
// ownerRegistrar 把扫码登录绑定的 owner id 写入 owner 白名单
// （尽力而为）。见 Deps.OwnerRegistrar。
ownerRegistrar OwnerRegistrar
webhooks map[string]http.Handler
// envResolver resolves the per-turn loop environment (its role) for a session.
// See persona_env.go / design/22 §4.5.
// envResolver 解析会话每轮的 loop 环境（其角色）。见
// persona_env.go / design/22 §4.5。
envResolver turnEnvResolver
log *slog.Logger
hub *hub
}
// New validates cfg and deps and returns a ready Server. Auth is required; the
// WebSocket hub subscribes to deps.Bus immediately so events published before
// ListenAndServe are still relayed to clients.
// New 校验 cfg 与 deps 并返回就绪的 Server。Auth 必填；WebSocket hub
// 会立即订阅 deps.Bus，因此 ListenAndServe 之前发布的事件仍会转发给
// 客户端。
func New(cfg *config.Config, deps Deps) (*Server, error) {
if cfg == nil {
return nil, fmt.Errorf("gateway: nil config")
@@ -131,9 +126,9 @@ func New(cfg *config.Config, deps Deps) (*Server, error) {
dataDir: deps.DataDir,
log: sourcedLog(deps.Log),
}
// One role resolver shared by the Server handlers and the hub, so every
// owner-facing entry point (REST, WS, voice) resolves the session's role the
// same way (design/22 §4.5) instead of each hardcoding loop.OwnerEnv().
// 一个角色解析器由 Server 各处理器与 hub 共用，使每个面向 owner
// 的入口（REST、WS、语音）以同样方式解析会话角色（design/22 §4.5），
// 而不是各自硬编码 loop.OwnerEnv()。
s.envResolver = turnEnvResolver{history: deps.History, personas: deps.Personas, log: s.log}
s.hub = newHub(s)
s.hub.orchestrator = newOrchestrator(s)
@@ -141,16 +136,16 @@ func New(cfg *config.Config, deps Deps) (*Server, error) {
return s, nil
}
// Start launches the gateway in a goroutine, returning a channel that carries the
// ListenAndServe result. The server runs until ctx is canceled.
// Start 在 goroutine 中启动网关，返回携带 ListenAndServe 结果的
// channel。服务器运行至 ctx 被取消。
func (s *Server) Start(ctx context.Context) (<-chan error, error) {
errCh := make(chan error, 1)
go func() { errCh <- s.ListenAndServe(ctx) }()
return errCh, nil
}
// ListenAndServe binds the configured host:port (default :8080) and blocks until
// ctx is canceled, then gracefully drains in-flight requests.
// ListenAndServe 绑定配置的 host:port（默认 :8080）并阻塞至 ctx 被
// 取消，随后优雅地排空在途请求。
func (s *Server) ListenAndServe(ctx context.Context) error {
if ctx == nil {
ctx = context.Background()
@@ -178,17 +173,15 @@ func (s *Server) ListenAndServe(ctx context.Context) error {
return nil
}
// Close frees the hub's bus subscription and closes live connections.
// Close 释放 hub 的 bus 订阅并关闭存活连接。
func (s *Server) Close() { s.hub.close() }
// newOrchestrator builds the session-message orchestrator (design/15) bound to
// this server's hub. The TurnRunner shares the hub's runTurn (so a turn always
// emits the same stream contract), and the StateSink fans per-session state
// transitions out through the hub broadcast for the frontend — those transitions
// are the design/11 §7.1 「状态机信号」, so every authenticated connection may follow
// them. It is wrapped in a
// function so the composition root can later override the injection (an explicit
// wiring point rather than a hard-coded construct).
// newOrchestrator 构建绑定到本 server 的 hub 的会话消息 orchestrator
// （design/15）。TurnRunner 复用 hub 的 runTurn（因此一轮总是发出相同的
// 流契约），StateSink 则把每会话状态迁移经由 hub 广播扇出给前端 ——
// 这些迁移属于 design/11 §7.1 的「状态机信号」，因此每个已认证连接都
// 可以跟随它们。它被包在函数里，以便组合根之后能覆盖注入（一个显式
// 的接线点，而不是硬编码的构造）。
func newOrchestrator(s *Server) *Orchestrator {
return NewOrchestrator(OrchestratorOptions{
Run: func(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) {
@@ -203,8 +196,8 @@ func newOrchestrator(s *Server) *Orchestrator {
})
}
// sourcedLog returns a logger tagged with source "gateway", substituting a
// discard logger for nil so Server can always log safely.
// sourcedLog 返回带 source "gateway" 标签的 logger，nil 时替换为丢弃型
// logger，使 Server 总能安全地打日志。
func sourcedLog(l *slog.Logger) *slog.Logger {
if l == nil {
l = slog.New(slog.NewTextHandler(io.Discard, nil))
internal/services/gateway/session_read.go
+28 -33
View File
@@ -11,39 +11,36 @@ import (
"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.
// 本文件归属 D4 已读游标面（Android 协作 B3）：推进会话已读游标的
// POST /api/v1/sessions/{id}/read 端点，以及它为清除账号其他浏览器
// 未读徽标（无需重读）而广播的 `session_read` 实时帧。游标本身存于
// chat_sessions.read_at（internal/runtime/history）；GET /sessions 由它
// 推导 unreadCount（history.go）。游标是会话级、单所有者的：
// 一个账号每个会话一个游标 —— 在任意一端阅读即为所有端标记该
// 会话已读。
// wsSessionRead is the WS event type for a read-cursor advance.
// wsSessionRead 是已读游标推进的 WS 事件类型。
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.
// sessionReadRequest 是 POST /api/v1/sessions/{id}/read 的请求体。
// Before 是指针，以便把字段缺失（空请求体）与显式取值区分开：
// 缺失表示「现在」—— 客户端把它已看到的一切都标记为已读。
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).
// handleMarkSessionRead 推进会话的已读游标，并以 Unix 毫秒返回结果
// 游标 {data:{sessionId, readAt}}。与其他历史写入一样走 owner/admin
// 门（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.
// 请求体 {before?} 是可选毫秒时间戳，界定标记已读的范围 ——
// created_at <= readAt 的消息变为已读。缺失（或空请求体）表示现在。
// 已给出但非正数或非数值的 before 返回 400：游标是真实的毫秒
// 时间戳，「负数游标造成的未读」不是有意义的客户端请求。未知
// 会话 id 返回 404；HistoryStore 缺失返回 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.
// 成功时新游标作为 `session_read` 帧广播，使该账号其他浏览器也清除
// 徽标（一端阅读 = 处处已读）。失败路径不广播任何东西。
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})
@@ -81,15 +78,13 @@ func (s *Server) handleMarkSessionRead(w http.ResponseWriter, r *http.Request) {
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. It is one of the
// design/11 §7.1 「状态机信号」 (it reports a cursor, never a message body), so the
// frame gate keeps it on the all-connections path. Like the other history frames it
// relies on the hub having 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.
// broadcastSessionRead 把一次已读游标推进作为 `session_read` 帧扇出到
// 每个客户端，携带会话的新游标，使同一账号的其他端无需重读即可
// 清除该会话的未读徽标。它属于 design/11 §7.1 的「状态机信号」
// （只报告游标，绝不报告消息正文），因此帧门让它留在全连接路径上。
// 与其他历史帧一样，它依赖 hub 没有按会话订阅这一点，由客户端按
// sessionId 过滤；游标的唯一事实来源仍是 chat_sessions.read_at，
// 因此错过该帧的客户端会在下次 GET /sessions 时重新收敛。
func (h *hub) broadcastSessionRead(sessionID string, readAt int64) {
h.broadcast(wsEvent{
Type: wsSessionRead,
internal/services/gateway/ws_frame_scope.go
+24 -24
View File
@@ -1,34 +1,34 @@
package gateway
// This file owns the outbound WS frame role gate (design/11 §7.1 「WS 下行帧的
// 角色门」, decided by the owner on 2026-10-03). It is the ONE place that decides
// which connections a downstream frame may reach:
// 本文件是 WS 下行帧角色门的唯一归属地（design/11 §7.1「WS 下行帧的
// 角色门」，主人于 2026-10-03 决策）。它是判定一个下行帧可以触达哪些
// 连接的唯一位置：
//
// - a frame whose type is listed in frameReachesAll is a 「状态机信号」: it
// carries no content, so every authenticated connection receives it;
// - EVERY other frame type — reply text (message.*), thinking, tool calls and
// results, persisted/edited/deleted messages, graph updates, sub-agent and
// task traffic, private persona/device state — goes to owner + admin only.
// - 类型列在 frameReachesAll 中的帧属于「状态机信号」：不含内容，
// 因此每个已认证连接都能收到；
// - 其余所有帧类型 —— 回复正文（message.*）、思考、工具调用与
// 结果、持久化/编辑/删除的消息、图谱更新、子代理与任务
// 流量、角色私有状态/设备状态 —— 只发 owner + admin。
//
// The classification is fail-closed: a new frame type answers false until someone
// deliberately registers it here AND in design/11 §7.1, so forgetting to classify
// a type can never leak content. No call site decides visibility for itself (a
// per-call-site flag is exactly how one missed spot becomes a leak), which is why
// hub.broadcast looks the class up from the frame it was handed.
// 分类是 fail-closed 的：新帧类型在有人把它显式登记到此处、并且在
// design/11 §7.1 登记之前一律返回 false，因此漏分类某个类型
// 永远不会泄露内容。没有任何调用点自行决定可见性（按调用点传
// 标志正是「漏一处即成泄露」的来源），这也是 hub.broadcast
// 从收到的帧反查类别的原因。
//
// This gate decides who is PUSHED to. Data visibility stays with the REST gates:
// a viewer still reads its 12 read-only monitoring endpoints through viewerAuth
// (design/11 §7.1 「语义澄清」).
// 本门只决定向谁 PUSH。数据可见性仍归 REST 门：
// viewer 依然只能通过 viewerAuth 读取它的 12 个只读监控端点
// （design/11 §7.1「语义澄清」）。
// frameReachesAll reports whether a downstream frame type may be delivered to
// every authenticated connection. Only the 「状态机信号」 of the design/11 §7.1
// whitelist qualify: session_busy / pending_ack / pending_drain / session_idle /
// session_interrupted / pending_full / session_read. Each one only reports that a
// session changed busy state, so a read-only monitor can follow the conversation's
// activity without receiving the conversation itself.
// frameReachesAll 报告某个下行帧类型是否可投递给每个已认证
// 连接。只有 design/11 §7.1 白名单中的「状态机信号」符合条件：
// session_busy / pending_ack / pending_drain / session_idle /
// session_interrupted / pending_full / session_read。它们只报告
// 会话忙闲状态发生了变化，因此只读监控端可以跟随对话的
// 活跃度，而不必接收对话本身。
//
// Widening this list is a contract change: register the reason (「不含内容」) in
// design/11 §7.1 and mirror it in ws_frame_scope_test.go first.
// 扩大这个名单属于契约变更：先在 design/11 §7.1 登记理由
// （「不含内容」），并在 ws_frame_scope_test.go 中同步。
func frameReachesAll(typ string) bool {
switch typ {
case EventSessionBusy,
internal/services/gateway/ws_frame_scope_test.go
+45 -52
View File
@@ -14,45 +14,42 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// These tests lock the design/11 §7.1 frame gate: a viewer connection may follow
// the session state machine but must never receive content, while owner and admin
// keep both classes. frameScopeCases is the design/11 §7.1 table in code — every
// frame type named there, published on the subject that really produces it — so a
// classification mistake (a content frame made public, or a state signal made
// private) fails here instead of leaking silently.
// 这些测试锁住 design/11 §7.1 的帧门：viewer 连接可以跟随会话状态机，
// 但绝不可收到内容，而 owner 与 admin 两类都保留。frameScopeCases 是
// design/11 §7.1 表在代码中的体现 —— 表中点名的每种帧类型，都发布在
// 真正产生它的主题上 —— 因此分类错误（内容帧被公开，或状态信号被
// 私有化）会在这里失败，而不是静默泄露。
// frameScopeCase is one row of the design/11 §7.1 table.
// frameScopeCase 是 design/11 §7.1 表的一行。
type frameScopeCase struct {
typ string // downstream frame type
subject string // bus subject that produces it
payload any // payload published on that subject
public bool // 「状态机信号」: no content → every authenticated connection
typ string // 下行帧类型
subject string // 产生它的 bus 主题
payload any // 发布在该主题上的载荷
public bool // 「状态机信号」：不含内容 → 每个已认证连接
}
// scopeSentinel is published last: the bus runs one worker per subscription
// sequentially, so a connection that has seen the sentinel has already been
// offered every frame published before it. That turns "the viewer received nothing
// else" into an exact count instead of a guessed idle window.
// scopeSentinel 最后发布：bus 每个订阅跑一个 worker 且按序处理，因此
// 已经看到哨兵的连接必定已经被投递过它之前发布的每一个帧。这把
// 「viewer 没有收到别的」从猜测的空闲窗口变成精确计数。
const scopeSentinel = EventSessionIdle
// The batch pacing used by publishFrameScopeCases, kept well under the hub's
// wsSendBuffer depth so a burst never trips the slow-client drop.
// publishFrameScopeCases 使用的批次节流，远低于 hub 的 wsSendBuffer
// 深度，因此一次突发永不触发慢客户端丢弃。
const (
scopePaceFrames = 4
scopePacePause = 50 * time.Millisecond
)
// scopeRead is the outcome of one connection's reader goroutine.
// scopeRead 是一个连接读取 goroutine 的结果。
type scopeRead struct {
role auth.Role
types []string
err error
}
// frameScopeCases returns the design/11 §7.1 table. Relayed frames use their real
// ws.<type> subject; stream frames publish the real model.StreamEvent the loop
// emits on ws.stream, so the test drives the production path
// bus → relay → broadcast instead of a hand-built frame.
// frameScopeCases 返回 design/11 §7.1 表。转发帧使用它们真实的 ws.<type>
// 主题；流帧发布 loop 在 ws.stream 上发出的真实 model.StreamEvent，因此
// 测试驱动的是生产路径 bus → relay → broadcast，而不是手工构造的帧。
func frameScopeCases() []frameScopeCase {
relayed := func(typ string, public bool) frameScopeCase {
return frameScopeCase{typ: typ, subject: wsPrefix + typ, payload: map[string]any{"sessionId": "web"}, public: public}
@@ -113,12 +110,11 @@ func frameScopeCases() []frameScopeCase {
}
}
// TestWSFrameScopePerRole drives the whole design/11 §7.1 table through real
// WebSocket connections with real role tokens: owner and admin receive every frame
// of the table, while the viewer receives exactly the whitelisted state signals, in
// publish order. The viewer assertions the design calls out by name (message.delta /
// thinking.delta / message_persisted / graph_update / mood denied, session_idle
// delivered) are re-stated explicitly so the contract is readable in the test.
// TestWSFrameScopePerRole 用真实角色 token 经真实 WebSocket 连接驱动
// 整张 design/11 §7.1 表：owner 与 admin 收到表中的每一帧，而 viewer
// 恰好收到白名单状态信号，且按发布顺序。设计点名的 viewer 断言
// （message.delta / thinking.delta / message_persisted / graph_update /
// mood 被拒，session_idle 被投递）在此显式复述，使契约在测试中可读。
func TestWSFrameScopePerRole(t *testing.T) {
t.Parallel()
cases := frameScopeCases()
@@ -140,9 +136,9 @@ func TestWSFrameScopePerRole(t *testing.T) {
auth.RoleViewer: len(public),
}
// The clients drain on their own goroutines while the batch is published: the
// hub drops frames for a client whose send buffer is full (its slow-client
// policy), so a client that only started reading afterwards would lose the tail.
// 各客户端在批次发布期间于自己的 goroutine 上排空：hub 会丢弃
// 发送缓冲已满的客户端的帧（其慢客户端策略），因此之后才开始
// 读取的客户端会丢掉尾部。
results := make(chan scopeRead, len(conns))
for role, want := range wants {
role, want := role, want
@@ -177,11 +173,10 @@ func TestWSFrameScopePerRole(t *testing.T) {
}
}
// TestHubBroadcastScopeMatrix is the per-frame-type × per-role matrix, driven
// directly through the delivery entry so every row of the table is asserted in
// isolation (a leaked frame would otherwise be hard to attribute). claims are set
// explicitly per role, so a default-deny regression shows up as the viewer losing
// the frame it is entitled to.
// TestHubBroadcastScopeMatrix 是「每种帧类型 × 每种角色」矩阵，直接经
// 投递入口驱动，使表中每一行都被孤立断言（否则泄露的帧难以归因）。
// claims 按角色显式设置，因此默认拒绝的回归会表现为 viewer 丢掉它
// 本应收到的帧。
func TestHubBroadcastScopeMatrix(t *testing.T) {
t.Parallel()
hb := newTestHub(nil, nil)
@@ -217,11 +212,10 @@ func TestHubBroadcastScopeMatrix(t *testing.T) {
}
}
// TestFrameScopeAllowsExactlyTheStateSignals pins the two properties that keep the
// gate from drifting: an unclassified frame type is never public (so a frame added
// by a later milestone cannot leak by omission), and the public set is exactly the
// state signals design/11 §7.1 registers — widening it means editing the design, the
// gate and this table together.
// TestFrameScopeAllowsExactlyTheStateSignals 钉住两条使帧门不漂移的
// 性质：未分类的帧类型永不是公开的（因此后续里程碑新增的帧不会因
// 漏分类而泄露），且公开集合恰好是 design/11 §7.1 登记的状态信号
// —— 扩大它意味着要同时改设计、帧门和这张表。
func TestFrameScopeAllowsExactlyTheStateSignals(t *testing.T) {
t.Parallel()
if frameReachesAll("frame_type_nobody_classified") {
@@ -250,8 +244,7 @@ func TestFrameScopeAllowsExactlyTheStateSignals(t *testing.T) {
assertFrameTypes(t, "design/11 §7.1 whitelist", public, want)
}
// frameTypes returns the table's frame types in table order (all of them, or only
// the public ones).
// frameTypes 按表顺序返回表中的帧类型（全部，或仅公开的那些）。
func frameTypes(cases []frameScopeCase, publicOnly bool) []string {
out := make([]string, 0, len(cases))
for _, tc := range cases {
@@ -263,10 +256,10 @@ func frameTypes(cases []frameScopeCase, publicOnly bool) []string {
return out
}
// publishFrameScopeCases publishes every row on its production subject, followed by
// the sentinel that closes the batch. The batch is paced: a client whose send buffer
// is full has its frames dropped (the hub's documented slow-client policy), so a
// 37-frame burst would exercise the drop policy rather than the gate.
// publishFrameScopeCases 在每行各自的生产主题上发布它，随后发布关闭
// 批次的哨兵。批次是节流的：发送缓冲已满的客户端其帧会被丢弃
// （hub 有文档的慢客户端策略），因此 37 帧的突发放送会去考验丢弃
// 策略而不是帧门。
func publishFrameScopeCases(t *testing.T, env *testEnv, cases []frameScopeCase) {
t.Helper()
for i, tc := range cases {
@@ -282,9 +275,9 @@ func publishFrameScopeCases(t *testing.T, env *testEnv, cases []frameScopeCase)
}
}
// readFrameTypes reads exactly want frames and returns their types in arrival order.
// It reports an error rather than failing the test so it can run on the reader
// goroutines that drain each client while the batch is published.
// readFrameTypes 恰好读取 want 个帧并按到达顺序返回其类型。它返回错误
// 而不是让测试失败，以便能跑在批次发布期间排空各客户端的读取
// goroutine 上。
func readFrameTypes(conn *websocket.Conn, want int) ([]string, error) {
got := make([]string, 0, want)
for len(got) < want {
@@ -302,7 +295,7 @@ func readFrameTypes(conn *websocket.Conn, want int) ([]string, error) {
return got, nil
}
// assertFrameTypes compares the received frame types to the expected sequence.
// assertFrameTypes 比较收到的帧类型与期望序列。
func assertFrameTypes(t *testing.T, what string, got, want []string) {
t.Helper()
if len(got) != len(want) {
@@ -315,7 +308,7 @@ func assertFrameTypes(t *testing.T, what string, got, want []string) {
}
}
// containsFrameType reports whether typ is among the received frame types.
// containsFrameType 报告 typ 是否在收到的帧类型之中。
func containsFrameType(types []string, typ string) bool {
for _, got := range types {
if got == typ {
internal/services/gateway/ws_hub_message_event.go
+16 -17
View File
@@ -1,31 +1,30 @@
package gateway
// wsMessageEdited / wsMessageDeleted are the downstream event types of the D3
// history writes: a message's content was edited, or a message was removed. Both
// reuse the wsEvent envelope and the same fan-out as `message_persisted` — and,
// like it and every other content frame, they are owner/admin only under the
// design/11 §7.1 frame gate. The hub has no per-session subscription, so the frame
// carries SessionID and a client filters it by its active session, exactly as it
// does for message_persisted.
// wsMessageEdited / wsMessageDeleted 是 D3 历史写入的下行事件类型：
// 消息内容被编辑，或消息被删除。二者复用 wsEvent 信封与
// `message_persisted` 相同的扇出 —— 并且与它以及其他所有内容帧
// 一样，在 design/11 §7.1 帧门下只发 owner/admin。hub 没有按会话的
// 订阅，因此帧携带 SessionID，由客户端按其活跃会话过滤，与它对
// message_persisted 的处理完全一致。
const (
wsMessageEdited = "message_edited"
wsMessageDeleted = "message_deleted"
)
// broadcastMessageEdited fans a content edit out to the owner/admin connections as
// a `message_edited` frame whose data is the updated message DTO. The frame is a
// render-only sync: the edited content's single source of truth stays the history
// store, so a client that missed the frame converges again on its next read. The
// caller strips Thinking to match the message_persisted/replay contract.
// broadcastMessageEdited 把一次内容编辑作为 `message_edited` 帧扇出到
// owner/admin 连接，其 data 是更新后的消息 DTO。该帧只用于渲染同步：
// 编辑后内容的唯一事实来源仍是 history store，因此错过该帧的客户端
// 会在下次读取时重新收敛。调用方会抹除 Thinking，以匹配
// message_persisted/回放契约。
func (h *hub) broadcastMessageEdited(msg historyMessage) {
h.broadcast(wsEvent{Type: wsMessageEdited, SessionID: msg.SessionID, Data: msg})
}
// broadcastMessageDeleted fans a removal out to the owner/admin connections as a
// `message_deleted` frame carrying the removed id (plus the owning sessionId for
// client-side session filtering), so every browser of the account drops the row
// without a refresh. sessionID may be empty when the store could not report it;
// clients then fall back to the globally unique id.
// broadcastMessageDeleted 把一次删除作为 `message_deleted` 帧扇出到
// owner/admin 连接，携带被删除的 id（外加所属 sessionId，供客户端
// 按会话过滤），使该账号的每个浏览器无需刷新即可移除该行。
// 当存储无法报告 sessionID 时它可能为空；客户端此时回退到全局
// 唯一的 id。
func (h *hub) broadcastMessageDeleted(sessionID, id string) {
h.broadcast(wsEvent{Type: wsMessageDeleted, SessionID: sessionID, Data: map[string]any{"id": id}})
}
internal/services/gateway/ws_hub_stream_scope_test.go
+17 -19
View File
@@ -13,8 +13,8 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// scopeStreamAdapter is a model.Adapter that emits two text deltas so a test can
// drive the real loop and observe the framed events it produces.
// scopeStreamAdapter 是一个 model.Adapter，发出两个文本增量，使测试
// 能驱动真实 loop 并观察它产出的成帧事件。
type scopeStreamAdapter struct{}
func (scopeStreamAdapter) Name() string { return "scope-test" }
@@ -31,10 +31,10 @@ func (scopeStreamAdapter) ChatStream(_ context.Context, _ model.Request, on func
return model.Response{Content: "Hel lo"}, nil
}
// newScopedStreamHub wires a hub to a real loop whose OnStream closure mirrors
// cmd/cyrene's streamChunk (publish the whole StreamEvent value to ws.stream), so
// a test exercises the production path OnStream → bus → relay → frame. It returns
// the hub and a registered fake connection.
// newScopedStreamHub 把 hub 接到真实 loop，其 OnStream 闭包与
// cmd/cyrene 的 streamChunk 一致（把整个 StreamEvent 值发布到
// ws.stream），因此测试走的是生产路径 OnStream → bus → relay → frame。
// 它返回 hub 与一个已注册的假连接。
func newScopedStreamHub(t *testing.T) (*hub, *wsConn) {
t.Helper()
b := bus.New()
@@ -58,10 +58,9 @@ func newScopedStreamHub(t *testing.T) (*hub, *wsConn) {
return hb, c
}
// readDeltas scans the connection's frames until want message.delta frames have
// arrived. The deltas reach the conn asynchronously through the bus while
// runTurn's own stream_end is sent straight to the conn, so a test cannot assume
// an order.
// readDeltas 扫描连接的帧，直到 want 个 message.delta 帧到达。增量经
// bus 异步到达该连接，而 runTurn 自己的 stream_end 是直接发给连接
// 的，因此测试不能假定顺序。
func readDeltas(t *testing.T, c *wsConn, want int) []wsEvent {
t.Helper()
var deltas []wsEvent
@@ -83,11 +82,11 @@ func readDeltas(t *testing.T, c *wsConn, want int) []wsEvent {
return deltas
}
// TestStreamFramesCarrySessionScope is the end-to-end lock for A1: a turn driven
// through the hub's runTurn must surface message.delta frames that carry the
// session and the turn id, so a client can route them to the active session. It
// runs the real loop instead of calling broadcastStream directly, which also
// proves the bus carries the new fields untouched — no publisher change needed.
// TestStreamFramesCarrySessionScope 是 A1 的端到端锁定：经 hub 的
// runTurn 驱动的一轮必须发出携带会话与轮次 id 的 message.delta 帧，
// 使客户端能把它们路由到活跃会话。它跑真实 loop 而不是直接调用
// broadcastStream，这同时证明 bus 原样携带了新字段 —— 无需改动
// 发布方。
func TestStreamFramesCarrySessionScope(t *testing.T) {
t.Parallel()
hb, c := newScopedStreamHub(t)
@@ -116,10 +115,9 @@ func TestStreamFramesCarrySessionScope(t *testing.T) {
}
}
// TestStreamFramesTurnScopeChangesPerTurn covers the orchestrated resume shape: a
// pending turn drained after the first one completes is a second runTurn call, so
// it must take a fresh turnId. Without that, deltas from the resumed turn would
// be indistinguishable from the previous turn's.
// TestStreamFramesTurnScopeChangesPerTurn 覆盖编排下的续跑形态：在
// 首轮完成之后被排空的待处理轮次是第二次 runTurn 调用，因此它必须
// 取新的 turnId。否则续跑轮次的增量与上一轮的将无法区分。
func TestStreamFramesTurnScopeChangesPerTurn(t *testing.T) {
t.Parallel()
hb, c := newScopedStreamHub(t)
internal/services/gateway/ws_hub_turn.go
+54 -62
View File
@@ -13,21 +13,20 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// relay is the bus handler for the ws.> pattern and the producer bridge
// patterns. The ws.chat subject is a streaming trigger (run the loop and emit
// chunks); a ws.* subject becomes an outbound event of its design type; a
// bridged producer subject (device_update, task output/done, automation result,
// reminder, subagent progress) is normalized into the matching ws.* event type.
// Events with no WS-visible mapping are dropped so internal agent-to-agent
// traffic never leaks to clients.
// relay 是 ws.> 模式与各生产者桥接模式的 bus 处理器。ws.chat 主题是
// 流式触发器（跑 loop 并发 chunk）；ws.* 主题成为其设计类型对应的
// 下行事件；被桥接的生产者主题（device_update、任务 output/done、
// automation result、reminder、subagent progress）被规范化为匹配的
// ws.* 事件类型。没有 WS 可见映射的事件会被丢弃，使内部 agent 间
// 流量永不泄露给客户端。
func (h *hub) relay(ctx context.Context, e bus.Event) error {
if e.Subject == wsChatSubject {
go h.streamChat(ctx, e.Payload)
return nil
}
if e.Subject == wsStreamSubject {
// OnStream→bus→relay path: the loop already published a structured
// model.StreamEvent; map it to the typed downstream event.
// OnStream→bus→relay 路径：loop 已经发布了结构化的
// model.StreamEvent；把它映射为带类型的下行事件。
h.broadcastStream(e.Payload)
return nil
}
@@ -39,12 +38,12 @@ func (h *hub) relay(ctx context.Context, e bus.Event) error {
return nil
}
// streamChat is the ws.chat trigger entry point. It decodes the request, and
// routes the turn through the session orchestrator when one is wired (the
// non-blocking busy/pending/resume path of design/15); otherwise it falls back to
// the legacy direct run so a hub built without an orchestrator keeps the exact
// stream_chunk/stream_end contract. It runs on its own goroutine so a slow model
// call does not stall the bus subscription worker.
// streamChat 是 ws.chat 触发器的入口。它解码请求，并在接了
// orchestrator 时把该轮路由经会话 orchestrator（design/15 的非阻塞
// busy/pending/resume 路径）；否则回退到旧的直接运行方式，使未接
// orchestrator 的 hub 保持完全相同的 stream_chunk/stream_end 契约。
// 它跑在各自的 goroutine 上，因此慢的模型调用不会阻塞 bus 订阅
// worker。
func (h *hub) streamChat(ctx context.Context, payload any) {
var req chatStreamRequest
if b, err := json.Marshal(payload); err == nil {
@@ -57,15 +56,13 @@ func (h *hub) streamChat(ctx context.Context, payload any) {
if req.SessionID == "" {
req.SessionID = "web"
}
// The Web main session is the owner's private conversation (JWT = owner), so
// it binds the full-access env: memory read+write, unchanged behavior. The role
// comes from the session's binding (design/22 §4.5): a session switched to role Y
// answers as Y on the next turn, while an unbound/existing session keeps
// resolving to the global default.
// Web 主会话是 owner 的私有对话（JWT = owner），因此它绑定全权限
// env：记忆读写，行为不变。角色来自会话的绑定（design/22 §4.5）：
// 切到角色 Y 的会话在下一轮以 Y 应答，而未绑定的/既有的会话继续
// 解析为全局默认角色。
msgs := userChatMessages(req.Text, req.Image)
// Thread the client's deviceID and behavior mode into the orchestrator via ctx
// so both the persisted user turn and the assistant reply are tagged
// (design/16 §4.5, §5.1).
// 通过 ctx 把客户端的 deviceID 与行为模式串进 orchestrator，使
// 持久化的用户轮次与助手回复都带上标记（design/16 §4.5、§5.1）。
ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode)
env := h.envResolver.owner(ctx, req.SessionID)
if h.orchestrator != nil {
@@ -75,20 +72,19 @@ func (h *hub) streamChat(ctx context.Context, payload any) {
_, _ = h.runTurn(ctx, req.SessionID, msgs, env)
}
// runTurn executes one agent turn and broadcasts its output stream. It is the
// unit both the orchestrator (via its injected TurnRunner) and the legacy
// fallback path share, so a turn always emits the same stream_chunk/stream_end
// events regardless of who drives it. When the loop is streaming (OnStream wired)
// the structured deltas already fan out as message.*/thinking.*/tool_call.*
// events through the OnStream→bus→relay path, so only the non-streaming path
// emits a single whole-text chunk to hold the stream contract.
// runTurn 执行一个 agent 轮次并广播其输出流。它是 orchestrator（经其
// 注入的 TurnRunner）与旧回退路径共用的单元，因此无论由谁驱动，一轮
// 总是发出相同的 stream_chunk/stream_end 事件。当 loop 处于流式
// （接了 OnStream）时，结构化增量已经通过 OnStream→bus→relay 路径以
// message.*/thinking.*/tool_call.* 事件扇出，因此只有非流式路径才发出
// 一整段文本 chunk 以维持流契约。
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
turnID := newTurnID()
// Thread the turn identity into the run context so the loop stamps it on every
// streamed event it emits (message.*/thinking.*/tool_call.*); the hub then puts
// it on the outbound frame. Every turn takes a fresh id — a session has at most
// one in-flight turn — so a client can correlate one burst of deltas to one
// assistant turn (design/16 §4.2).
// 把轮次身份串进运行上下文，使 loop 把它盖在它发出的每个流式
// 事件上（message.*/thinking.*/tool_call.*）；hub 随后把它放到
// 下行帧上。每一轮都取全新的 id —— 一个会话最多一个在途轮次
// —— 因此客户端能把一串增量关联到同一个助手轮次
// （design/16 §4.2）。
ctx = loop.WithTurnID(ctx, turnID)
if h.runner == nil || len(msgs) == 0 {
h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "idle"}})
@@ -107,9 +103,9 @@ func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Messag
return res, nil
}
// handleInterrupt routes a client "interrupt" frame (design/15 §3) to the
// orchestrator, canceling the in-flight turn and clearing the pending queue for
// the requested session. It is a no-op when no orchestrator is wired.
// handleInterrupt 把客户端的 "interrupt" 帧（design/15 §3）路由到
// orchestrator，取消在途轮次并清空指定会话的待处理队列。未接
// orchestrator 时是空操作。
func (h *hub) handleInterrupt(payload json.RawMessage) {
if h.orchestrator == nil {
return
@@ -124,15 +120,13 @@ func (h *hub) handleInterrupt(payload json.RawMessage) {
h.orchestrator.Interrupt(req.SessionID)
}
// broadcastStream maps a streamed model.StreamEvent to its downstream WS event
// type and fans it out through the hub's frame gate (design/11 §7.1): every
// structured stream frame carries model output, so it reaches owner + admin only.
// It runs on the bus subscription worker;
// seq is assigned atomically so a burst stays ordered even if the loop emits
// events from more than one goroutine. SessionID/TurnID ride on the event itself
// (the loop stamps them at the turn boundary from the run context), so the frame
// carries the scope a client needs to route it to the active session and to one
// assistant turn.
// broadcastStream 把流式的 model.StreamEvent 映射为其下行 WS 事件类型，
// 并经 hub 的帧门扇出（design/11 §7.1）：每一个结构化流帧都携带模型
// 输出，因此只到达 owner + admin。它跑在 bus 订阅 worker 上；seq 原子
// 分配，因此即便 loop 从多个 goroutine 发出事件，一串帧仍保持有序。
// SessionID/TurnID 随事件本身携带（loop 在轮次边界从运行上下文盖上
// 它们），因此帧带有客户端把它路由到活跃会话以及关联到某个助手
// 轮次所需的范围。
func (h *hub) broadcastStream(payload any) {
ev, ok := payload.(model.StreamEvent)
if !ok {
@@ -142,25 +136,23 @@ func (h *hub) broadcastStream(payload any) {
h.broadcast(wsEvent{Type: typ, SessionID: ev.SessionID, TurnID: ev.TurnID, Seq: int(h.streamSeq.Add(1)), Data: data})
}
// chatStreamRequest is the decoded ws.chat payload.
// chatStreamRequest 是解码后的 ws.chat 载荷。
type chatStreamRequest struct {
SessionID string `json:"sessionId"`
Text string `json:"text"`
Image string `json:"image,omitempty"`
// DeviceID is the client-produced identifier of the sending browser/device
// (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted
// alongside the message so the agent knows which end the owner is chatting from.
// DeviceID 是客户端生成的发送浏览器/设备标识（design/16 §4.5），
// 例如 "web:chrome:a1b2c3"；它可选，并随消息一同持久化，使 agent
// 知道主人是从哪一端发消息。
DeviceID string `json:"deviceId,omitempty"`
// Mode is the behavior mode the client sent the turn in (design/01 §3,
// design/16 §5.1); it is optional and persisted per message, falling back to
// "chat" when omitted.
// Mode 是客户端发送该轮时所用的行为模式（design/01 §3、
// design/16 §5.1）；它可选，按消息持久化，缺省时回退为 "chat"。
Mode string `json:"mode,omitempty"`
}
// newTurnID returns a fresh random hex identifier used to correlate a burst of
// streamed WS events to one assistant turn. It burns the crypto/rand source and
// falls back to a nanosecond timestamp if the source is unavailable, so a id is
// always produced (a blank turnId would break client correlation silently).
// newTurnID 返回新的随机十六进制标识，用于把一串流式 WS 事件关联到
// 同一个助手轮次。它消耗 crypto/rand 源，源不可用时回退到纳秒
// 时间戳，因此总能产出 id（空的 turnId 会静默破坏客户端关联）。
func newTurnID() string {
var b [8]byte
if _, err := rand.Read(b[:]); err == nil {
@@ -169,10 +161,10 @@ func newTurnID() string {
return "turn-" + time.Now().UTC().Format("20060102150405.000000000")
}
// publishChatStream publishes a ws.chat streaming trigger (REST stream mode).
// deviceID is the sending browser/device id and mode the behavior mode the client
// sent the turn in, both carried on the payload so the ws.chat consumer can tag
// the persisted turn (design/16 §4.5, §5.1).
// publishChatStream 发布一个 ws.chat 流式触发器（REST 流式模式）。
// deviceID 是发送方的浏览器/设备 id，mode 是客户端发送该轮时所使用的
// 行为模式，二者都随载荷携带，使 ws.chat 消费方能给持久化的轮次打
// 标记（design/16 §4.5、§5.1）。
func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID, mode string) {
if s.bus == nil {
return