docs(gateway): gateway 第 2 批 5 个文件注释改中文（只改注释，代码零字节改动）
主人 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 的测试名与断言字面量一字符未改。
This commit is contained in:
AskaEth committed
1 parent
fe1ff33115
commit
355f81d04a
5 files changed
+149
-181
No files matched your search
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in new issue
Block a user