From 355f81d04af1b14e970d88e03c5ef1c3e610f089 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 22:10:33 +0800 Subject: [PATCH] =?UTF-8?q?docs(gateway):=20gateway=20=E7=AC=AC=202=20?= =?UTF-8?q?=E6=89=B9=205=20=E4=B8=AA=E6=96=87=E4=BB=B6=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E6=94=B9=E4=B8=AD=E6=96=87=EF=BC=88=E5=8F=AA=E6=94=B9=E6=B3=A8?= =?UTF-8?q?=E9=87=8A=EF=BC=8C=E4=BB=A3=E7=A0=81=E9=9B=B6=E5=AD=97=E8=8A=82?= =?UTF-8?q?=E6=94=B9=E5=8A=A8=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主人 2026-10-03 拍板「注释一律中文」。本批 = WS 上行门与握手回放那批文件： ws_hub.go / ws_history.go / ws_route.go / ws_role_gate_test.go / voice_signal_test.go 含中文的整行注释 6 → 148；无行尾注释。 校验（判据 = "只改注释、零代码字节改动"）： - git diff -U0 | 排除 +++/--- / 注释行 / 空行 / 行尾注释 → 为空 （含"把纯空行变动也视为违规"的更严变体，同样为空） - 独立实现：两侧剥掉注释后逐行比对代码文本 → 5/5 全等 - gofmt -l 干净（改动前同样干净，不存在"顺手格式化"） - go test ./... 全绿；go build/vet 绿；linux/amd64 + darwin/arm64 交叉编译绿 安全约束强度逐句保留：ws_route.go 的「默认拒（非黑名单）/ 删掉它等于放开一切」、 ws_history.go 的「门在 replayHistory 第一条语句 / 扩角色须先登记 design/11 §7.2 / 回放不含 thinking」、ws_role_gate_test.go 的测试名与断言字面量一字符未改。 --- .../services/gateway/voice_signal_test.go | 12 +- internal/services/gateway/ws_history.go | 37 +++-- internal/services/gateway/ws_hub.go | 59 ++++---- .../services/gateway/ws_role_gate_test.go | 141 ++++++++---------- internal/services/gateway/ws_route.go | 81 +++++----- 5 files changed, 149 insertions(+), 181 deletions(-) diff --git a/internal/services/gateway/voice_signal_test.go b/internal/services/gateway/voice_signal_test.go index 4fda6d7..2c3b4bd 100644 --- a/internal/services/gateway/voice_signal_test.go +++ b/internal/services/gateway/voice_signal_test.go @@ -12,8 +12,8 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice" ) -// signalVoice is a VoiceService that also implements voiceProviderReporter, so a -// test can control both the STT/TTS result and the reported provider rung. +// signalVoice 是同时实现 voiceProviderReporter 的 VoiceService，使测试既能 +// 控制 STT/TTS 结果，也能控制所上报的 provider 档位。 type signalVoice struct { text string audio voice.Audio @@ -65,8 +65,8 @@ func TestHandleTranscribe_SilentSignal(t *testing.T) { func TestHandleTranscribe_SilentFallbackWithoutReporter(t *testing.T) { env := newEnv(t, nil, false) - // audioStubVoice does not implement voiceProviderReporter, so the handler - // must still flag an empty transcription as silent. + // audioStubVoice 没有实现 voiceProviderReporter，因此处理函数仍须把 + // 空转写标为 silent。 env.srv.SetVoice(&audioStubVoice{}) token := env.token(t, auth.RoleOwner) @@ -138,8 +138,8 @@ func TestHubVoiceOutput_SilentSignal(t *testing.T) { t.Parallel() hb := newTestHub(&stubRunner{res: &loop.Result{Text: "你好", Steps: 1, StopReason: "stop"}}, nil) hb.voice = &signalVoice{text: "你好", rung: "silent"} - // voice_input is an owner-facing instruction since design/11 §7.2 B, so the - // connection that drives it has to carry owner/admin claims. + // 自 design/11 §7.2 B 起 voice_input 是面向主人的指令，因此驱动它的连接 + // 必须携带 owner/admin claims。 c := newOwnerConn(hb, 32) hb.register(c) diff --git a/internal/services/gateway/ws_history.go b/internal/services/gateway/ws_history.go index 6843ce3..80d81bb 100644 --- a/internal/services/gateway/ws_history.go +++ b/internal/services/gateway/ws_history.go @@ -4,31 +4,28 @@ import ( "context" ) -// wsReplayLimit is the number of most-recent messages replayed to a connecting -// client. It is intentionally small so a reconnect is cheap; the frontend pulls -// the full active-session history via REST (design/16 §4.1) and the replay is -// only there to give a brand-new connection immediate context. +// wsReplayLimit 是回放给新接入客户端的最近消息条数。刻意取小值，使重连代价 +// 低廉；完整活跃会话历史由前端走 REST 拉取（design/16 §4.1），回放只是让 +// 一条全新连接立刻拿到上下文。 const wsReplayLimit = 50 -// replayHistory sends the most recent messages for sessionID to a single, -// freshly-registered connection as a ws "history" event. +// replayHistory 把 sessionID 最近的若干条消息作为 ws "history" 事件发给 +// 单条刚注册的连接。 // -// The replay carries the main session's message bodies, so it is owner/admin only -// (design/11 §7.2 A, decided by the owner on 2026-10-03). ownerAdminClaims is the -// same predicate the proactive push uses, so 「who may be handed the owner's -// conversation unasked」 is answered in one place instead of per caller. A viewer -// connection completes the handshake with no replay at all: its read-only window -// is the REST monitoring surface (design/11 §7.1 「语义澄清」), not a push. +// 回放携带主会话的消息正文，因此仅限 owner/admin（design/11 §7.2 A，主人于 +// 2026-10-03 决策）。ownerAdminClaims 与主动推送用的是同一判据，所以 +// 「谁可以在未被要求的情况下拿到主人的对话」只在一处回答，而不散落到各调用点。 +// viewer 连接完成握手时完全不做回放：它的只读窗口是 REST 监控面 +// （design/11 §7.1「语义澄清」），不是推送。 // -// The gate lives here rather than at the register/replay call site in -// ws_hub_conn.go on purpose: this is the only function that emits the replay, so -// no future caller can reach it without the check. A role the owner later wants -// to serve the replay to has to be registered in design/11 §7.2 first, exactly -// like an inbound frame (see ws_route.go). +// 门放在这里而不是 ws_hub_conn.go 里的 register/回放调用点，是刻意的： +// 这是唯一发出回放的函数，此后任何调用者都无法绕开该检查。日后主人想让某个 +// 角色也收到回放，必须先把它登记到 design/11 §7.2，与上行帧的做法完全一致 +// （见 ws_route.go）。 // -// It is best-effort: a nil store or a read error skips the replay silently. -// Thinking is NOT included (design/16 §6), so the frame never leaks -// chain-of-thought even to owner/admin. +// 回放是尽力而为：store 为 nil 或读取出错都静默跳过。 +// 回放不含 thinking（design/16 §6），因此该帧即使对 owner/admin +// 也不会泄漏思维链。 func (h *hub) replayHistory(c *wsConn, sessionID string) { if !ownerAdminClaims(c.claims) { return diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index 91dd516..1d17a0c 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -14,8 +14,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" ) -// WebSocket tuning constants. The ping/pong cadence keeps intermediary proxies -// from dropping idle connections. +// WebSocket 调参常量。ping/pong 节奏用于防止中间代理把空闲连接断开。 const ( wsEventsPattern = "ws.>" wsPrefix = "ws." @@ -30,9 +29,8 @@ const ( wsSendBuffer = 32 ) -// wsEvent is the outbound (server→client) frame. TurnID/Seq are optional and -// used by the streaming path so a client can correlate a burst of streamed -// events to one assistant turn and keep them in order. +// wsEvent 是下行（服务端→客户端）帧。TurnID/Seq 可选，由流式路径使用， +// 让客户端能把一串流式事件归到同一个助手回合并保持顺序。 type wsEvent struct { Type string `json:"type"` TurnID string `json:"turnId,omitempty"` @@ -41,15 +39,15 @@ type wsEvent struct { Data any `json:"data,omitempty"` } -// wsInbound is the client→server frame. Data carries the per-type payload. +// wsInbound 是上行（客户端→服务端）帧。Data 承载按类型区分的载荷。 type wsInbound struct { Type string `json:"type"` ReqID string `json:"reqId,omitempty"` Data json.RawMessage `json:"data,omitempty"` } -// hub owns the connection set and relays bus events to clients, each frame -// gated by its type (see broadcast / ws_frame_scope.go). +// hub 持有连接集合并把 bus 事件转发给客户端，每个帧按其类型过门 +// （见 broadcast / ws_frame_scope.go）。 type hub struct { mu sync.RWMutex conns map[*wsConn]struct{} @@ -61,8 +59,8 @@ type hub struct { streaming bool orchestrator *Orchestrator history HistoryStore - // envResolver resolves each turn's role (design/22 §4.5); it mirrors the - // server's resolver so the WS path and the REST path never diverge. + // envResolver 解析每个回合的角色（design/22 §4.5）；它与 server 的解析器 + // 同源，使 WS 路径与 REST 路径不会分叉。 envResolver turnEnvResolver log *slog.Logger upgrader websocket.Upgrader @@ -72,7 +70,7 @@ type hub struct { streamSeq atomic.Uint64 } -// newHub builds a hub bound to the server's dependencies. +// newHub 构建一个绑定到 server 依赖的 hub。 func newHub(s *Server) *hub { origins, allowAll := s.allowedOrigins() return &hub{ @@ -96,10 +94,9 @@ func newHub(s *Server) *hub { } } -// start subscribes the hub to the bus event stream. Publishing to ws.> arrives -// here and is fanned out to clients; the producer bridge patterns (iot/task/ -// automation/reminder/subagent-progress) are also subscribed and normalized into -// ws.* event types. Nothing happens when no bus is wired. +// start 让 hub 订阅 bus 事件流。发往 ws.> 的事件到达此处后广播给客户端； +// 生产者桥接模式（iot/task/automation/reminder/subagent-progress）也一并订阅， +// 并归一为 ws.* 事件类型。未接入 bus 时什么都不做。 func (h *hub) start() { if h.bus == nil { return @@ -114,8 +111,8 @@ func (h *hub) start() { } } -// close cancels the bus subscription, interrupts every orchestrator session (so -// no background turn outlives the hub), and shuts every live connection down. +// close 取消 bus 订阅、中断所有 orchestrator 会话（使后台回合不会比 hub 活得 +// 更久），并关闭每条在线连接。 func (h *hub) close() { if h.cancel != nil { h.cancel() @@ -135,14 +132,14 @@ func (h *hub) close() { } } -// register adds a connection to the set. +// register 把一条连接加入集合。 func (h *hub) register(c *wsConn) { h.mu.Lock() h.conns[c] = struct{}{} h.mu.Unlock() } -// unregister removes a connection from the set. +// unregister 把一条连接从集合中移除。 func (h *hub) unregister(c *wsConn) { h.mu.Lock() if _, ok := h.conns[c]; ok { @@ -151,24 +148,22 @@ func (h *hub) unregister(c *wsConn) { h.mu.Unlock() } -// connCount reports the number of live connections (observability/tests). +// connCount 返回在线连接数（可观测性/测试）。 func (h *hub) connCount() int { h.mu.RLock() defer h.mu.RUnlock() return len(h.conns) } -// broadcast is the single delivery entry for outbound frames, and the design/11 -// §7.1 role gate itself: it looks the frame's type up (frameReachesAll) and fans -// a registered 「状态机信号」 out to every registered connection, while every other -// type — content, tool, persisted-message, graph, private-state, and any type not -// classified yet — reaches owner + admin only. Registration happens only after the -// JWT is verified (ws_hub_conn.go), so the connection set is the authenticated set. +// broadcast 是下行帧的唯一投递入口，也是 design/11 §7.1 角色门本身：它按帧的 +// 类型查表（frameReachesAll），把登记在册的「状态机信号」广播给每条已注册连接， +// 而其余类型 —— 内容、工具、持久化消息、图谱、角色私有状态，以及任何尚未分类 +// 的类型 —— 只送达 owner + admin。连接只在 JWT 校验通过后才注册 +// （ws_hub_conn.go），因此连接集合就是已认证集合。 // -// Gating here rather than at the call sites is deliberate: there is exactly one -// fan-out loop, so a frame cannot be sent to the wrong audience by a call site that -// forgot to pick the restricted entry. It never blocks on any single (possibly -// slow) client: a full send buffer drops the frame. +// 把门放在这里而非调用点是刻意的：全包只有这一个 fan-out 循环，调用点即使忘了 +// 改走受限入口，也不可能把帧投错受众。它也不会被任何单个（可能很慢的）客户端 +// 阻塞：发送缓冲满即丢弃该帧。 func (h *hub) broadcast(ev wsEvent) { b, err := json.Marshal(ev) if err != nil { @@ -186,12 +181,12 @@ func (h *hub) broadcast(ev wsEvent) { } } -// eventType maps a bus subject to its outbound event type, dropping the ws. prefix. +// eventType 把 bus subject 映射为下行事件类型，去掉 ws. 前缀。 func eventType(subject string) string { return strings.TrimPrefix(subject, wsPrefix) } -// rawData converts a possibly-empty raw payload into a value safe to publish. +// rawData 把可能为空的原始载荷转换为可安全发布的值。 func rawData(d json.RawMessage) any { if len(d) == 0 { return nil diff --git a/internal/services/gateway/ws_role_gate_test.go b/internal/services/gateway/ws_role_gate_test.go index 313cfda..df22a70 100644 --- a/internal/services/gateway/ws_role_gate_test.go +++ b/internal/services/gateway/ws_role_gate_test.go @@ -19,40 +19,35 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task" ) -// This file locks design/11 §7.2 — the two remaining places where a `viewer` -// (the read-only monitoring role) used to reach beyond its lane through -// /ws/chat, both found while implementing the §7.1 downlink gate: +// 本文件锁定 design/11 §7.2 —— `viewer`（只读监控角色）过去经由 /ws/chat +// 越界的最后两处，都是实现 §7.1 下行门时发现的： // -// A. the handshake history replay (ws_history.go) carries the main session's -// message bodies, so it is owner/admin only; -// B. every inbound (client→server) JSON frame (ws_route.go) is denied by -// default for a non owner/admin connection, so a viewer can follow the -// session state machine but can never send an instruction. +// A. 握手时的历史回放（ws_history.go）携带主会话的消息正文，因此仅限 +// owner/admin； +// B. 每一个上行（客户端→服务端）JSON 帧（ws_route.go）对非 owner/admin +// 连接默认拒，因此 viewer 可以跟随会话状态机，但永远发不出指令。 // -// Together with the §7.1 downlink gate they pin the combined contract: -// `viewer` = may connect, may read its 12 read-only REST monitoring endpoints, -// may receive the 7 state-machine signals — and gets and sends nothing else over -// WS. The connections and tokens are real (httptest server + signed JWTs), so a -// mistake in the routing tables cannot hide behind a fake. +// 与 §7.1 下行门一起，它们钉住了合并后的契约： +// `viewer` = 可以连接、可以读它那 12 个只读 REST 监控端点、可以接收 +// 7 个状态机信号 —— 除此之外在 WS 上收不到也发不出任何东西。这里的连接与 +// 令牌都是真的（httptest server + 签名 JWT），所以路由表里的错误无法藏在 +// 假实现后面。 const ( - // wsRoleGateReadWait bounds a read that is expected to produce a frame. - // gorilla corrupts a connection after a read timeout, so every connection - // here performs at most one timeout read, always as its last read. + // wsRoleGateReadWait 界定一次预期会产出帧的读取。gorilla 在读超时后会 + // 弄坏连接，因此这里每条连接至多做一次超时读取，且总是作为最后一次读。 wsRoleGateReadWait = 3 * time.Second - // wsRoleGateNoFrameWait is the window in which a frame that must not exist - // would have arrived (a directed reply is queued synchronously by the - // handler, so an idle window this long is conclusive). + // wsRoleGateNoFrameWait 是「本不该存在的帧」若存在就会到达的窗口 + // （定向回复由处理函数同步入队，所以这么长的空转窗口即已足够判定）。 wsRoleGateNoFrameWait = 250 * time.Millisecond - // wsRoleGateSettle lets the side effects of an accepted frame land before they - // are asserted to be absent. + // wsRoleGateSettle 让被接受帧的副作用先落地，再断言它们不存在。 wsRoleGateSettle = 200 * time.Millisecond - // wsRoleGateWriteWait bounds a control-frame write (ping / close). + // wsRoleGateWriteWait 界定一次控制帧写入（ping / close）。 wsRoleGateWriteWait = 2 * time.Second ) -// countingRunner is the Chatter seam with a call counter: design/11 §7.2 B has to -// prove that a viewer's `chat` frame never reaches the model, so "was the runner -// driven" must be observable rather than inferred from a broadcast. +// countingRunner 是带调用计数的 Chatter 接缝：design/11 §7.2 B 必须证明 +// viewer 的 `chat` 帧不会到达模型，所以「runner 是否被驱动过」要可观测， +// 而不能从广播反推。 type countingRunner struct { mu sync.Mutex calls int @@ -71,9 +66,9 @@ func (r *countingRunner) count() int { return r.calls } -// replayHistoryStore serves the handshake replay with a fixed transcript while -// inheriting fakeHistory's Append recording, so one store both makes the replay -// deterministic and reports what an accepted `chat` frame persisted. +// replayHistoryStore 用固定转录内容提供握手回放，同时继承 fakeHistory 的 +// Append 记录，因此一个 store 既让回放确定，又能报告被接受的 `chat` 帧持久化了 +// 什么。 type replayHistoryStore struct { *fakeHistory mu sync.Mutex @@ -90,7 +85,7 @@ func (s *replayHistoryStore) Recent(context.Context, string, int) ([]history.Mes return append([]history.Message(nil), s.msgs...), nil } -// historyFrameWire is the handshake replay frame decoded for assertions. +// historyFrameWire 是解码后用于断言的握手回放帧。 type historyFrameWire struct { Type string `json:"type"` SessionID string `json:"sessionId"` @@ -103,7 +98,7 @@ type historyFrameWire struct { } `json:"data"` } -// errorFrameWire is any {type:"error"} frame decoded for assertions. +// errorFrameWire 是解码后用于断言的任意 {type:"error"} 帧。 type errorFrameWire struct { Type string `json:"type"` Data struct { @@ -112,11 +107,10 @@ type errorFrameWire struct { } `json:"data"` } -// roleGateEnv wires a gateway for these tests: a counting runner (so a leaked -// inbound frame is visible as a model call), a store that replays a transcript and -// records persistence, and the sub-agent/task seams so a leaked subagent_ask or -// task_output_query would produce a real, assertable content frame instead of a -// handler-level error. +// roleGateEnv 为这些测试接线一个 gateway：带计数的 runner（使泄漏的上行帧 +// 表现为一次模型调用）、会回放转录并记录持久化的 store，以及子代理/任务的 +// 接缝，使泄漏的 subagent_ask 或 task_output_query 会产生一个真实且可断言的 +// 内容帧，而不是处理函数级错误。 type roleGateEnv struct { srv *Server url string @@ -150,12 +144,10 @@ func (e *roleGateEnv) token(t *testing.T, role auth.Role) string { return signAny(t, e.jwt, "acct", role) } -// TestWSHandshakeReplayPerRole locks design/11 §7.2 A: the handshake history -// replay holds the main session's message bodies, so only owner/admin connections -// receive it. A leaked replay is always the FIRST frame on a connection — it is -// queued during the upgrade, before the read pump starts and before any reply — -// so reading the viewer's first frame turns "no replay" into an exact assertion -// instead of a guessed idle window. +// TestWSHandshakeReplayPerRole 锁定 design/11 §7.2 A：握手历史回放持有主会话的 +// 消息正文，因此只有 owner/admin 连接会收到它。泄漏的回放永远是连接上的第一个 +// 帧 —— 它在 upgrade 期间就已入队，早于读泵启动、也早于任何回复 —— 所以读 +// viewer 的第一帧就把「没有回放」变成精确断言，而不是靠猜一个空转窗口。 func TestWSHandshakeReplayPerRole(t *testing.T) { t.Parallel() env := newRoleGateEnv(t, []history.Message{ @@ -199,8 +191,8 @@ func TestWSHandshakeReplayPerRole(t *testing.T) { } } - // The viewer is probed so that a frame is guaranteed to come back (the §7.2 B - // refusal); if the replay leaked it would be ahead of that reply. + // 这里主动探一次 viewer，以保证一定有帧回来（§7.2 B 的拒绝）；若回放泄漏， + // 它会排在那条回复之前。 if err := viewer.WriteJSON(wsInbound{Type: "chat", Data: json.RawMessage(`{"sessionId":"web","text":"viewer"}`)}); err != nil { t.Fatalf("viewer write chat: %v", err) } @@ -209,16 +201,15 @@ func TestWSHandshakeReplayPerRole(t *testing.T) { } } -// TestWSInboundDeniedForViewer locks design/11 §7.2 B: a non owner/admin -// connection may not send any in-app instruction. Every rejected frame gets a -// connection-directed {type:"error", data:{code:"forbidden"}} and nothing else — -// no content frame, no broadcast, no model call, no persisted message. +// TestWSInboundDeniedForViewer 锁定 design/11 §7.2 B：非 owner/admin 连接不得 +// 发送任何应用内指令。每个被拒的帧都得到定向到该连接的 +// {type:"error", data:{code:"forbidden"}}，别的什么都没有 —— 没有内容帧、 +// 没有广播、没有模型调用、没有持久化消息。 func TestWSInboundDeniedForViewer(t *testing.T) { t.Parallel() env := newRoleGateEnv(t, nil) - // A second viewer connection observes that the refusals are directed at the - // sender: a broadcast refusal would land here too. + // 第二条 viewer 连接用于观察拒绝是指向发送方的：若拒绝走广播，这里也会收到。 observer := dialWS(t, env.url, env.token(t, auth.RoleViewer)) defer observer.Close() @@ -252,15 +243,15 @@ func TestWSInboundDeniedForViewer(t *testing.T) { if ef.Data.Code != wsForbiddenCode { t.Fatalf("%s error code = %q, want %q (frame=%s)", tc.name, ef.Data.Code, wsForbiddenCode, raw) } - // For subagent_ask / task_output_query a leak would be a directed - // subagent_event / task_output reply carrying a sub-agent's context or - // a task's output, so assert the refusal is the whole answer. + // 对 subagent_ask / task_output_query 而言，泄漏会是一条定向的 + // subagent_event / task_output 回复，携带子代理上下文或任务输出， + // 因此断言拒绝就是全部应答。 assertNoFrame(t, conn, tc.name+" follow-up") }) } - // The `chat` case is the one that used to drive the model and persist the - // message; give its (refused) side effects a moment to not happen. + // `chat` 这一例正是过去会驱动模型并持久化消息的那个；给它（已被拒的） + // 副作用一点时间，确认它们没有发生。 time.Sleep(wsRoleGateSettle) calls, records := env.run.count(), env.hist.records() if calls != 0 || len(records) != 0 { @@ -269,10 +260,9 @@ func TestWSInboundDeniedForViewer(t *testing.T) { assertNoFrame(t, observer, "observer viewer (a refusal must not be broadcast)") } -// TestWSInboundChatDrivesOwnerTurn is the owner half of design/11 §7.2 B: the very -// frame a viewer is refused (`chat`) still drives the model and persists the turn -// for an owner, so the gate is proven to be a role gate rather than a broken chat -// path. +// TestWSInboundChatDrivesOwnerTurn 是 design/11 §7.2 B 的主人侧对照：被拒给 viewer +// 的那个帧（`chat`）对主人依然会驱动模型并持久化回合，从而证明这道门确实是角色 +// 门，而不是 chat 路径坏了。 func TestWSInboundChatDrivesOwnerTurn(t *testing.T) { t.Parallel() env := newRoleGateEnv(t, nil) @@ -295,10 +285,9 @@ func TestWSInboundChatDrivesOwnerTurn(t *testing.T) { } } -// TestWSInboundGateKeepsProtocolFrames proves the inbound gate cannot kill a -// connection (design/11 §7.2 B 「协议级帧不受影响」): after a refusal the viewer -// still completes a ping/pong round trip and a normal close, and the server -// deregisters it. +// TestWSInboundGateKeepsProtocolFrames 证明上行门不会杀死连接 +// （design/11 §7.2 B「协议级帧不受影响」）：被拒一次之后，viewer 仍能完成 +// ping/pong 往返和正常关闭，服务端也会注销它。 func TestWSInboundGateKeepsProtocolFrames(t *testing.T) { t.Parallel() env := newRoleGateEnv(t, nil) @@ -312,8 +301,8 @@ func TestWSInboundGateKeepsProtocolFrames(t *testing.T) { t.Fatalf("probe reply = %q (%s), want the forbidden error frame", typ, raw) } - // A second refusal round trip proves the read pump survived the first: the - // gate returns before any handler, so the connection is untouched. + // 第二次被拒的往返证明读泵在第一轮之后依然存活：门在任何处理函数之前就返回， + // 连接未被触动。 if err := viewer.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web"}`)}); err != nil { t.Fatalf("write interrupt after refusal: %v", err) } @@ -321,10 +310,9 @@ func TestWSInboundGateKeepsProtocolFrames(t *testing.T) { t.Fatalf("second probe reply = %q (%s), want the forbidden error frame", typ, raw) } - // A client ping is answered by the server's default ping handler. The pong is - // consumed by the client read below, so the handler is what reports it. This - // is the connection's single timeout read (gorilla corrupts a connection after - // a read deadline), hence it comes after every assertion on the read side. + // 客户端 ping 由服务端默认的 ping 处理函数应答。pong 会被下面的客户端读取 + // 消费掉，因此由该处理函数来报告它。这是本连接唯一的一次超时读取 + // （gorilla 在读截止时间过后会弄坏连接），所以它排在读侧所有断言之后。 ponged := make(chan struct{}, 1) viewer.SetPongHandler(func(string) error { select { @@ -346,16 +334,15 @@ func TestWSInboundGateKeepsProtocolFrames(t *testing.T) { t.Fatal("the server never answered the ping; the gate broke the protocol path") } - // A normal close still deregisters the connection. + // 正常关闭依然会注销该连接。 if err := viewer.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(wsRoleGateWriteWait)); err != nil { t.Fatalf("write close: %v", err) } waitConnCount(t, env.srv, 0) } -// readWire reads one frame, returning its type plus the raw bytes (so a caller can -// decode the type-specific payload). It fails the test when no frame arrives -// before the deadline. +// readWire 读取一个帧，返回它的类型与原始字节（以便调用方解码按类型区分的 +// 载荷）。若在截止时间前没有帧到达则让测试失败。 func readWire(t *testing.T, conn *websocket.Conn, what string) (string, []byte) { t.Helper() _ = conn.SetReadDeadline(time.Now().Add(wsRoleGateReadWait)) @@ -372,9 +359,8 @@ func readWire(t *testing.T, conn *websocket.Conn, what string) (string, []byte) return env.Type, msg } -// assertNoFrame asserts that nothing further arrives on conn within a short -// window. It must be the connection's last read: gorilla leaves a connection -// corrupt after a read deadline expires. +// assertNoFrame 断言短窗口内 conn 上不会再有任何东西到达。它必须是该连接的最后 +// 一次读取：读截止时间到期后 gorilla 会留下一个已损坏的连接。 func assertNoFrame(t *testing.T, conn *websocket.Conn, what string) { t.Helper() _ = conn.SetReadDeadline(time.Now().Add(wsRoleGateNoFrameWait)) @@ -388,9 +374,8 @@ func assertNoFrame(t *testing.T, conn *websocket.Conn, what string) { } } -// waitUntil polls cond until it holds or the deadline passes, so a test can assert -// on an asynchronous turn (run → persist) without a fixed sleep that could pass -// vacuously. +// waitUntil 轮询 cond 直到它成立或超过截止时间，使测试能对异步回合（run → persist） +// 做断言，而不必用一段可能空跑通过的固定 sleep。 func waitUntil(t *testing.T, what string, cond func() bool) { t.Helper() deadline := time.Now().Add(wsRoleGateReadWait) diff --git a/internal/services/gateway/ws_route.go b/internal/services/gateway/ws_route.go index 6fca4c7..d6680c6 100644 --- a/internal/services/gateway/ws_route.go +++ b/internal/services/gateway/ws_route.go @@ -8,35 +8,30 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" ) -// wsForbiddenCode is the machine-readable code of the frame a non owner/admin -// connection gets for any inbound instruction (design/11 §7.2 B). It rides the -// file's existing error-frame shape — wsEvent{Type: "error", Data: map[string]any} -// — so the client decodes it with the code it already knows, and distinguishes -// 「你没有这个权限」 from a handler-level failure by the key alone. +// wsForbiddenCode 是非 owner/admin 连接发送任何上行指令时收到的帧的机器可读 +// 错误码（design/11 §7.2 B）。它沿用本文件既有的错误帧形状 —— +// wsEvent{Type: "error", Data: map[string]any} —— 客户端用它已经认识的 code +// 解码，仅凭这个键就能把「你没有这个权限」与处理函数级失败区分开。 const wsForbiddenCode = "forbidden" -// route dispatches one decoded inbound (client→server) frame by its type. +// route 按类型分发一个已解码的上行（客户端→服务端）帧。 // -// Role gate (design/11 §7.2 B, decided by the owner on 2026-10-03): a connection -// that is not owner/admin may not send ANY in-app instruction. The check is the -// first statement here and it denies by default — it is deliberately NOT a list of -// forbidden types — so a frame type added by a later milestone is refused until -// someone registers it on purpose. A viewer's /ws/chat connection is therefore a -// read-only monitor: it follows the session state machine but can never drive the -// model, cancel a turn, read a sub-agent's context or a task's output, or feed -// voice input. Each refused frame is answered on that one connection with -// {type:"error", data:{code:"forbidden"}}; nothing is broadcast, no handler runs, -// so no model call is made and nothing is persisted. +// 角色门（design/11 §7.2 B，主人于 2026-10-03 决策）：非 owner/admin 连接 +// 不得发送任何应用内指令。该检查是本函数的第一条语句，且默认拒 —— 它刻意 +// 不是一张「禁止类型」清单 —— 因此后续里程碑新增的帧类型在有人显式登记 +// 之前一律被拒。viewer 的 /ws/chat 连接因此是只读监视器：它跟随会话状态机， +// 但永远无法驱动模型、取消回合、读取子代理上下文或任务输出，也不能输入语音。 +// 每个被拒的帧只在那一根连接上得到 {type:"error", data:{code:"forbidden"}} +// 的应答；不广播、不进入任何处理函数，因此不产生模型调用，也不落库。 // -// Widening this: a viewer-side inbound frame may be allowed only after it is given -// an opening and registered in design/11 §7.2 (「必须先在设计里给它开口并登记白名单」). -// The blanket deny below is what makes forgetting that step safe, so allowing one -// means replacing it with a registry, not deleting it. +// 放宽此门：viewer 侧的上行帧只有先在设计里给它开口、并登记到 design/11 §7.2 +// 之后才允许（「必须先在设计里给它开口并登记白名单」）。下方的一律拒绝正是 +// 让「忘记这一步」仍然安全的保障，所以放开某一个类型意味着把它换成一张 +// 登记表，而不是删掉它。 // -// Protocol-level frames (ping/pong/close) never reach here — gorilla handles them -// inside ReadMessage — so the gate cannot kill a connection. A frame that is not -// JSON at all is answered by readPump before dispatch; it carries no type and can -// trigger nothing, which is why it keeps its parsing-level answer. +// 协议级帧（ping/pong/close）不会到达这里 —— gorilla 在 ReadMessage 内部 +// 处理 —— 因此这道门不会杀死连接。完全不是 JSON 的帧由 readPump 在分发前 +// 应答；它不带类型、触发不了任何东西，所以保留解析层的应答。 func (c *wsConn) route(in wsInbound) { if !ownerAdminClaims(c.claims) { c.h.log.Debug("ws inbound rejected", "type", in.Type, "role", claimRole(c.claims)) @@ -49,14 +44,13 @@ func (c *wsConn) route(in wsInbound) { case "chat": c.publishPublic(wsChatSubject, rawData(in.Data)) case "interrupt": - // design/15 §3: user says "停下/算了" → cancel the in-flight turn and - // clear the session's pending queue, so the next message starts fresh. + // design/15 §3: 用户说「停下/算了」→ 取消在飞回合并清空该会话的 + // 待处理队列，使下一条消息从干净状态开始。 c.h.handleInterrupt(in.Data) case "voice_input": - // Voice routes through the injected STT→chat→TTS pipeline. Without a - // voice service it replies with a clear error rather than dropping the - // frame. The audio reply is delivered as a voice_output event on the - // requesting connection. + // 语音经注入的 STT→chat→TTS 管线处理。没有 voice service 时回一个 + // 明确的错误，而不是把帧丢掉。音频回复作为 voice_output 事件发在 + // 发起请求的那根连接上。 c.handleVoiceInput(in) case "subagent_ask": c.handleSubagentAsk(in) @@ -67,9 +61,9 @@ func (c *wsConn) route(in wsInbound) { } } -// subagentAskRequest is the decoded payload of a subagent_ask frame. Mode selects -// the audit window (""/full/tail/keyword); n bounds a tail query; query is the -// keyword for a keyword query. +// subagentAskRequest 是 subagent_ask 帧的解码载荷。Mode 选择审计窗口 +// （""/full/tail/keyword）；n 界定 tail 查询的条数；query 是 keyword 查询的 +// 关键词。 type subagentAskRequest struct { ID string `json:"id"` Mode string `json:"mode"` @@ -77,15 +71,14 @@ type subagentAskRequest struct { Query string `json:"query"` } -// taskOutputRequest is the decoded payload of a task_output_query frame. +// taskOutputRequest 是 task_output_query 帧的解码载荷。 type taskOutputRequest struct { TaskID string `json:"taskId"` } -// handleSubagentAsk audits a sub-agent's context and replies with a -// ws.subagent_event frame carrying the real context window. When no auditor is -// injected (gateway not wired with a scheduler) it replies with a clear error -// rather than a fabricated "pending" snapshot. +// handleSubagentAsk 审计某个子代理的上下文，并以携带真实上下文窗口的 +// ws.subagent_event 帧回复。未注入 auditor（gateway 没接 scheduler）时回一个 +// 明确的错误，而不是编造一个 "pending" 快照。 func (c *wsConn) handleSubagentAsk(in wsInbound) { if c.h.auditor == nil { c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "subagent audit unavailable"}}) @@ -110,10 +103,9 @@ func (c *wsConn) handleSubagentAsk(in wsInbound) { c.enqueueJSON(wsEvent{Type: "subagent_event", Data: res}) } -// handleTaskOutputQuery fetches the real task output for taskId and replies with -// a ws.task_output frame. When no task store is injected it replies with a clear -// error rather than an empty {}. A task that is still running returns the output -// captured so far. +// handleTaskOutputQuery 取回 taskId 的真实任务输出，并以 ws.task_output 帧回复。 +// 未注入 task store 时回一个明确的错误，而不是空的 {}。仍在运行的任务返回迄今 +// 已捕获的输出。 func (c *wsConn) handleTaskOutputQuery(in wsInbound) { if c.h.taskOutputs == nil { c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "task output unavailable"}}) @@ -145,9 +137,8 @@ func (c *wsConn) handleTaskOutputQuery(in wsInbound) { }}) } -// claimRole labels a connection's role for the inbound-rejection log record. A -// connection with nil claims (the fail-closed bare connection a unit test builds) -// logs as "none" instead of dereferencing the nil pointer. +// claimRole 为上行业务拒绝的日志记录标注连接角色。claims 为 nil 的连接 +// （单元测试构造的 fail-closed 裸连接）记为 "none"，而不是解引用空指针。 func claimRole(claims *auth.Claims) string { if claims == nil { return "none"