diff --git a/internal/services/gateway/broadcast.go b/internal/services/gateway/broadcast.go index 0ab8d1e..023d298 100644 --- a/internal/services/gateway/broadcast.go +++ b/internal/services/gateway/broadcast.go @@ -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, diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index b356378..23e1ec3 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -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}}) diff --git a/internal/services/gateway/history_write_broadcast_test.go b/internal/services/gateway/history_write_broadcast_test.go index 251a42d..3f8e2e2 100644 --- a/internal/services/gateway/history_write_broadcast_test.go +++ b/internal/services/gateway/history_write_broadcast_test.go @@ -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) diff --git a/internal/services/gateway/mood_test.go b/internal/services/gateway/mood_test.go index 4794663..3041535 100644 --- a/internal/services/gateway/mood_test.go +++ b/internal/services/gateway/mood_test.go @@ -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) diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index a70f0d4..4a4081b 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -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 - // (/uploads). Injected by the composition root; cfg.DataDir is the - // fallback. See Deps.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)) diff --git a/internal/services/gateway/session_read.go b/internal/services/gateway/session_read.go index 5f2c318..930cc40 100644 --- a/internal/services/gateway/session_read.go +++ b/internal/services/gateway/session_read.go @@ -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, diff --git a/internal/services/gateway/ws_frame_scope.go b/internal/services/gateway/ws_frame_scope.go index a2888b3..b441e2e 100644 --- a/internal/services/gateway/ws_frame_scope.go +++ b/internal/services/gateway/ws_frame_scope.go @@ -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, diff --git a/internal/services/gateway/ws_frame_scope_test.go b/internal/services/gateway/ws_frame_scope_test.go index 4b8c523..7889f70 100644 --- a/internal/services/gateway/ws_frame_scope_test.go +++ b/internal/services/gateway/ws_frame_scope_test.go @@ -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. 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. +// 主题；流帧发布 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 { diff --git a/internal/services/gateway/ws_hub_message_event.go b/internal/services/gateway/ws_hub_message_event.go index c9bdf04..127df0a 100644 --- a/internal/services/gateway/ws_hub_message_event.go +++ b/internal/services/gateway/ws_hub_message_event.go @@ -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}}) } diff --git a/internal/services/gateway/ws_hub_stream_scope_test.go b/internal/services/gateway/ws_hub_stream_scope_test.go index 02140b8..3ca3c5f 100644 --- a/internal/services/gateway/ws_hub_stream_scope_test.go +++ b/internal/services/gateway/ws_hub_stream_scope_test.go @@ -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) diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index f10ec4d..c0c7eda 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -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