From bc0047b44952a5d4fbc7567ed18bfc99f6573e6e Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 21:50:29 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway,security):=20WS=20=E4=B8=8B?= =?UTF-8?q?=E8=A1=8C=E5=B8=A7=E8=A7=92=E8=89=B2=E9=97=A8=20=E2=80=94=20?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E5=B8=A7=E4=BB=85=20owner+admin=EF=BC=8Cview?= =?UTF-8?q?er=20=E5=8F=AA=E6=94=B6=E7=8A=B6=E6=80=81=E6=9C=BA=E4=BF=A1?= =?UTF-8?q?=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit design/11 §7.1(主人 2026-10-03 拍板)。此前 hub.broadcast 是全连接广播,而任意角色 (含 viewer)都能连 /ws/chat ⇒ viewer 能收到角色回复正文(message.*)、推理内容 (thinking.*)、消息增量、graph_update、mood 等。这与 design/19 §1/§3「flow 可视化仅 owner/admin 可见」冲突,也与 GET /sessions/{id}/messages 的 owner/admin 档位不一致。 收口:唯一分类点 internal/services/gateway/ws_frame_scope.go 的 frameReachesAll(typ) (fail-closed switch,默认 false);hub.broadcast 投递前查表,白名单外对非 owner/admin 连接直接跳过。**不给 broadcast 加 bool 参数、不让调用点自己判断** —— 那正是"漏一处就是 泄漏"的形状。全包只有一个 fan-out 循环,14 个 .broadcast( 调用点全部落进该入口。 - 白名单只有 7 个状态机信号帧(不含任何内容):session_busy / pending_ack / pending_drain / session_idle / session_interrupted / pending_full / session_read - 其余全部(流式 message/thinking/tool_call、message_persisted/edited/deleted、graph_update、 subagent_event/result、task_output/task_done、automation_result、tool_used、capacity_awaken、 mode_switch、mood、device_update、reminder)仅 owner+admin - 语义澄清:本门管「谁被主动推送」,数据可见性仍由 REST 决定(viewer 的 12 条只读监控端点 不变,含 GET /personas 的 mood);/ws/chat 连接鉴权与 voice_output 定向帧不动 契约:Docs/api-client.md → v0.48(对 viewer 是收紧;owner/admin 零变化;无端点/字段形状变更) 验证:go build/vet 绿;go test ./... = 58 包 ok / 0 FAIL;-race ./internal/services/gateway/... ok; linux/amd64 + darwin/arm64 交叉编译绿。真 httptest WS + 真 JWT 按角色逐帧实测 37 帧: viewer 只收到 7 个状态信号,owner/admin 37/37。 变异验证:① graph_update 加回白名单 → 真 WS 抓到 viewer 收到 graph_update,3 个测试 FAIL; ② session_idle 移出白名单 → 3 个测试 FAIL(viewer 等不到哨兵帧而超时)。均已还原。 遗留(已登记,待主人裁决):① 握手 history 回放对任意角色照发(viewer 连上即得主会话最近 50 条正文);② /ws/chat 上行未按角色授权(viewer 可发 chat = 真驱动模型并落库; subagent_ask / task_output_query 的定向应答绕过本门)。 --- Docs/api-client.md | 12 +- internal/services/gateway/broadcast.go | 14 +- internal/services/gateway/extra_test.go | 35 +- .../gateway/history_write_broadcast_test.go | 6 +- internal/services/gateway/mood_test.go | 2 +- internal/services/gateway/server.go | 4 +- internal/services/gateway/session_read.go | 6 +- internal/services/gateway/ws_frame_scope.go | 45 +++ .../services/gateway/ws_frame_scope_test.go | 326 ++++++++++++++++++ internal/services/gateway/ws_hub.go | 20 +- .../services/gateway/ws_hub_message_event.go | 11 +- .../gateway/ws_hub_stream_scope_test.go | 2 +- internal/services/gateway/ws_hub_turn.go | 4 +- 13 files changed, 453 insertions(+), 34 deletions(-) create mode 100644 internal/services/gateway/ws_frame_scope.go create mode 100644 internal/services/gateway/ws_frame_scope_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 7589d91..b0736a6 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -110,7 +110,7 @@ - Web 端镜像实现在 `frontend/web/src/lib/enums.ts`(`labelFor(kind, value)` 对未知值返回原值),权威仍以本端点为准。 - **反漂移保证(后端)**:后端有测试断言「每个内置适配器的 `Name()` 都在 `platforms` 表内」「`moods` 与 persona 情绪状态机的内置状态一致」「`modes` 与前端 `MODES` 九模式一致」——漏登记即 CI FAIL,客户端拿到的表不会滞后于代码。 -> **契约版本 `v0.47`**(**`graph_update` 子代理泳道根节点新增可选 `props.persona`**:派发子代理时若指定了目标角色,该分支根节点的 `props` 就多一个 `persona` 键,标注它**正在扮演的角色**——用途是让流程图能显示「这个子代理是以谁的身份在替班」,而不只是子代理 id;**纯增量**:不带角色派发时该键不出现,老客户端忽略未知键即可,见 §3.5 与 v0.47 变更记录)。**上一版 `v0.46`**:(**`PATCH /api/v1/sessions/{id}` 的 `personaId` 由「可重新绑定该会话的角色」改为「出现即 `400`」**:**会话角色在创建时确定、之后不可变**,要换角色请新开会话(design/22 §8-1)——**明确拒绝**而非静默忽略/200 空操作;**键出现**(任何取值:`"yuki"` / `""` / `null` / 未知 id / 空白串)一律 `400 bad_request` 且消息点名 `personaId`,该检查**先于**会话存在性检查(未知会话 + `personaId` 也回 `400` 而不是 `404`)。**破坏性**:按 v0.42–v0.45 旧文实现、依赖「PATCH 换角色 / 显式 `""` 解绑」的客户端必须改为**新建会话**选角色;`title`/`pinned` 行为与 `POST /sessions {personaId}`(创建时绑定)**完全不变**,不再有 `501/200 {personaId}` 的 PATCH 路径)。上一版 `v0.45`:`PUT /admin/channels/{platform}/config` 的 `personaId` 由「省略/`""` 都 = 解绑」修订为三态(**省略(或 `null`)= 保留已存绑定**、**显式 `""` = 解绑**、**非空 id = 绑定**,未知角色仍 `400`)。 +> **契约版本 `v0.48`**(**WS 下行帧的角色门(design/11 §7.1)**:hub 的下行现在是**按帧类型分档**的——只有 7 个**状态机信号**(`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full` / `session_read`,**均不含任何内容**)继续发给**所有已认证连接**(含 `viewer`),**其余全部下行帧**(流式内容 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` / `stream_chunk` / `stream_end` / `thinking`;持久化 `message_persisted` / `message_edited` / `message_deleted` / `channel_persisted`;系统内部 `graph_update` / `subagent_event` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`;私有状态 `mood` / `device_update` / `reminder`)**改为仅发 `owner` + `admin`**)。**对 `viewer` 是收紧**:此前它连上 `/ws/chat` 就能收到角色回复正文与推理内容,现在收不到;**`owner` / `admin` 行为不变**;**无端点、无字段形状变更**——`viewer` 的 12 条只读监控 REST 端点(`viewerAuth`,v0.40)与 `/ws/chat` 的连接鉴权都**照旧**;**新帧类型默认只发 owner/admin**(fail-closed)。见 §3 开头的「下行帧的角色门」与 v0.48 变更记录。**上一版 `v0.47`**:(**`graph_update` 子代理泳道根节点新增可选 `props.persona`**:派发子代理时若指定了目标角色,该分支根节点的 `props` 就多一个 `persona` 键,标注它**正在扮演的角色**——用途是让流程图能显示「这个子代理是以谁的身份在替班」,而不只是子代理 id;**纯增量**:不带角色派发时该键不出现,老客户端忽略未知键即可,见 §3.5 与 v0.47 变更记录)。**上一版 `v0.46`**:(**`PATCH /api/v1/sessions/{id}` 的 `personaId` 由「可重新绑定该会话的角色」改为「出现即 `400`」**:**会话角色在创建时确定、之后不可变**,要换角色请新开会话(design/22 §8-1)——**明确拒绝**而非静默忽略/200 空操作;**键出现**(任何取值:`"yuki"` / `""` / `null` / 未知 id / 空白串)一律 `400 bad_request` 且消息点名 `personaId`,该检查**先于**会话存在性检查(未知会话 + `personaId` 也回 `400` 而不是 `404`)。**破坏性**:按 v0.42–v0.45 旧文实现、依赖「PATCH 换角色 / 显式 `""` 解绑」的客户端必须改为**新建会话**选角色;`title`/`pinned` 行为与 `POST /sessions {personaId}`(创建时绑定)**完全不变**,不再有 `501/200 {personaId}` 的 PATCH 路径)。上一版 `v0.45`:`PUT /admin/channels/{platform}/config` 的 `personaId` 由「省略/`""` 都 = 解绑」修订为三态(**省略(或 `null`)= 保留已存绑定**、**显式 `""` = 解绑**、**非空 id = 绑定**,未知角色仍 `400`)。 ## 2. REST 端点(管理,`/api/v1/admin/*`) @@ -244,6 +244,9 @@ - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`;未认证在握手前回 `401` JSON。 - **下行(服务端→客户端)**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`(`turnId`/`seq` 为流式关联字段,见 §3.1;`sessionId` 为**会话持久化(design/16)新增**,能确定会话的事件带上,供客户端按激活会话过滤;**v0.36 起结构化流帧 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 也带 `sessionId` + `turnId`**——多会话并发流式时客户端据此把 delta 路由到激活会话,见 §3.1)——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`(结构化流,design/04 §1.1),以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood`(情绪变化,见 §3.7) / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update` / **`message_persisted`**(会话持久化增量,见 §3.3)/ **`message_edited`**、**`message_deleted`**(消息编辑/删除增量,见 §3.8)/ **`session_read`**(已读游标推进,见 §3.9)。**会话编排状态(design/15,均为广播,`data.sessionId` 区分会话)**:`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`(见 §3.2)。语音回复走 `voice_output`(定向回请求连接,非广播)。 +- **下行帧的角色门(v0.48,design/11 §7.1)**: 下行帧按**类型**分两类,由后端**唯一一处**收口判定(**不在白名单里的帧默认只发 `owner` + `admin`**,fail-closed):① **状态机信号**(白名单,**不含任何内容**)→ **所有已认证连接**(含 `viewer`):`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`(§3.2)、`session_read`(§3.9);② **其余全部** → **仅 `owner` + `admin`**:流式内容 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` / `stream_chunk` / `stream_end` / `thinking`(§3.1)、持久化 `message_persisted`(§3.3)/ `channel_persisted`(§3.4)/ `message_edited` / `message_deleted`(§3.8)、系统内部 `graph_update`(§3.5)/ `subagent_event` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`、私有状态 `mood`(§3.7)/ `device_update` / `reminder`。 + + > **这是「谁被主动推送」的门,不是数据可见性**:`viewer` 的 12 条只读监控端点(`viewerAuth`,v0.40)**照旧**(含 `GET /personas` 的角色级 `mood`);`viewer` **仍可连** `/ws/chat`(连接鉴权不变),只是收不到第 ② 类帧。定向回请求连接的帧(`voice_output` 等)不受影响。 > **`history` 回放事件(design/16,新增)**:WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,platform?,deviceId?,personaId?,createdAt}]}}`——回放主会话最新 50 条(`thinking` 已剥离,不泄给非 owner;默认主会话,次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取)。客户端据此在重连/新开连接后补全上下文。消息体含 `senderName?`(第三方昵称,design/16 §4.6,主人私聊为空)与 `personaId?`(v0.42,说话时的角色,design/22 §7-5,空 = 默认角色)。 @@ -263,6 +266,8 @@ --- ### 3.2 会话编排状态事件(design/15) + +> **角色门(v0.48,design/11 §7.1)**:本节的 6 个状态信号**都在白名单内**——它们只报「会话忙/闲/排队」,**不含任何内容**,所以**所有已认证连接**(含 `viewer`)都收得到;真正的正文/推理内容走 §3.1 的流帧,那些**只发 `owner` + `admin`**。 - **触发**:同一会话内,`chat` 到达时若「她忙」(正在跑 loop/工具循环/子代理/长回复)→ 新消息**不阻塞**、入该会话 pending;这一层的状态经以下事件推给前端,供「正在做 xx… 有新消息稍后回」轻状态展示。 - **`session_busy`**:会话进入忙碌态(`{sessionId, mode}`,`mode` ∈ `idle|thinking|executing|subagent`)。 - **`pending_ack`**:忙时新消息已入队(`{sessionId, mode, queueDepth}`)——「收到,稍后回」。 @@ -285,6 +290,8 @@ - **客户端处理**:Web 聊天**无** `channel_persisted` handler,天然忽略、不进会话;**IM 适配器日志页**订阅该事件,按 `adapterId === 当前适配器`(或总览 `adapterId != ''`)过滤后**仅追加**(按 `id` 去重、不整页重载),首次拉取仍由 `GET /api/v1/admin/channels/messages` + 轮询兜底。 ### 3.5 实时全链路流程事件(`graph_update`,design/19) + +> **角色门(v0.48,design/11 §7.1)**:`graph_update` 属「系统内部」帧 → **仅 `owner` + `admin`**(`viewer` 收不到,与本节「仅主人 Web 会话发布」及 design/19 §1「flow 可视化仅 owner/admin」一致)。REST 侧可见性不变。 - **触发**:主会话 Web 每次会话轮次中,loop 的每个 trace span **开始/结束时**由 `loop.FlowPublisher` 经 bus `ws.graph_update` 广播增量 **FlowGraph**。**仅主人 Web 会话**(`sessionId == "web"` 或其 `web:` 前缀)发布——IM / 群聊 / 陌生人 / 子代理轮次**不**发布,防敏感信息泄向面板。管理员/主人 Web 端订阅后实时渲染「输入 → 思考(LLM) → 工具 → 输出」主干图。 - **载荷**:`{type:"graph_update", data:{id, nodes:[{id,label,kind?,status?,startedAt?,endedAt?,props?,log?}], edges:[{from,to}], sessionId, startedAt, endedAt}}`。 - **子代理分支节点的 `props.persona`(v0.47,design/22 §8-2)**:`subagent_spawn` 指定了目标角色派发时,该子代理在父会话图里的**分支根节点** `props` 会多一个 `persona` 键(值 = 目标角色 id),标注它**正在扮演的角色**;**不指定角色派发时该键不出现**(与旧版逐字一致),客户端按未知键忽略即可。 @@ -303,6 +310,8 @@ - **推送出口(D1,对客户端无新端点)**:在 WS/IM 投递之外,后端会把主动消息**异步镜像**到「推送出口」——主通道是**主人自建统一推送服务**(`PUSH_EXPORTER_URL` webhook,HTTP POST JSON `{source,title,content,platform,timestamp}`,可带 `Authorization: Bearer `),回退是**个推 Getui**(字段预留、SDK 未接入)。该出口只影响**出站投递**,不改变 WS 事件契约;未配/不可达时只记日志(注明「个推回退待接入」),**不影响**上面的 WS/IM 投递。移动端接入自建推送服务即可收到主人主动消息。 ### 3.7 情绪事件(`mood`,design/01 §2 情绪状态机 + 好感度) + +> **角色门(v0.48,design/11 §7.1)**:`mood` 属「私有状态」帧 → **仅 `owner` + `admin`**,`viewer` 不再收到它;但角色当前情绪**仍可读**——`viewer` 照旧调 `GET /personas` 拿到该角色的 `mood` 字段(REST 只读档不变,见 §1)。 - **触发**:每轮对话结束(loop 回合边界,每轮一次、**非**每步)把该轮用户消息的情绪信号应用到**该角色自己的**情绪状态机(按 persona 隔离),且**状态真的变化时**才广播一次 `mood`;状态不变(含「一直 calm」)**不发**,避免每轮噪声。 - **载荷**:`{type:"mood", data:{mood, personaId, affection}}`——`mood` ∈ `calm|happy|excited|sad|angry`(情绪状态机 state id,与消息体 `mood` 同源同值);`personaId` 为该角色 id(情绪/好感度跟随角色,不跟随行为模式);`affection` 为该角色当前好感度(受角色配置 `affection.min/max` 约束)。角色级数值,**不带** `sessionId`。 - **与消息体 `mood` 的关系**:assistant 消息体的 `mood` 是该轮结束后的**当前情绪**(每轮都写,见 §3.3),而 `mood` 事件只在**变化时**发——前者答「这条回复时她什么心情」,后者答「她的心情刚刚变了」。 @@ -368,6 +377,7 @@ - 语音:上行 `voice_input`(base64 音频或已转写 text),下行 `voice_output`(合成音频 base64 + text)。 ## 5. 变更记录 +- `v0.48`(**WS 下行帧的角色门**:`/ws/chat` 的下行帧改为**按帧类型分档**——**对 `viewer` 是收紧**(此前它连上就能收到角色回复正文与推理内容,现在收不到),**`owner` / `admin` 行为不变**,**无端点、无字段形状变更**;依据 design/11 §7.1,主人 2026-10-03 拍板「按建议收紧」):① **问题**:hub 的下行是**全连接广播**,而 `/ws/chat` 只验 JWT(任意角色可连),于是 `viewer` 连接会收到 `message.*` / `thinking.*` / `tool_call.*`(**含角色回复正文与推理内容**)、`message_persisted` / `message_edited` / `message_deleted`、`graph_update`、`mood`、`subagent_event` / `task_output` / `device_update` / `reminder` 等——与 design/19 §1/§3「flow 可视化仅 owner/admin」直接冲突,也与 `GET /sessions/{id}/messages` 的 owner/admin 档位不一致(**同一批内容从 WS 旁路漏出**)。② **规则(fail-closed,默认拒)**:**只有白名单里的 7 个「状态机信号」**(`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full` / `session_read`,**均不含任何内容**)发给**所有已认证连接**(含 `viewer`);**其余全部**(`stream_chunk` / `stream_end` / `thinking`、`message_persisted` / `message_edited` / `message_deleted` / `channel_persisted`、`graph_update` / `subagent_event` / `subagent.result` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`、`mood` / `device_update` / `reminder`)**只发 `owner` + `admin`**;**未登记的新帧类型默认只发 owner/admin**(要进白名单须先在 design/11 §7.1 登记理由「不含内容」)。③ **实现**:收口为**唯一一处**「按帧类型分类后投递」的入口(`internal/services/gateway/ws_frame_scope.go` + `hub.broadcast`),**不给每个调用点加开关**(那正是「漏一处即泄漏」的写法)——流式 `message.*` / `thinking.*` / `tool_call.*`(`broadcastStream`)、`stream_chunk` / `stream_end`(`runTurn` / `streamChat`)、`message_persisted` / `message_edited` / `message_deleted`、`mood` / `capacity_awaken`、编排状态信号(`StateSink`)、`graph_update` / `subagent_event` / `device_update` / `reminder`(`relay` 归一化)**全部经同一入口**,不存在第二条投递路径。④ **语义澄清(不是数据从此不可见)**:这条门管「**谁该被主动推送**」——`viewer` 的 **12 条只读监控端点**(v0.40 `viewerAuth`,含 `GET /personas` 的角色级 `mood`)**逐条照旧**;`/ws/chat` 的**连接鉴权不变**(`viewer` 仍可连,只是收不到上列帧);**定向回请求连接的帧**(`voice_output` / `error` / `subagent_ask`、`task_output_query` 的应答)不进这道门、行为不变。⑤ **测试与变异验证**:真 WS 连接 + 真角色 token 的**逐帧表驱动**断言(design/11 §7.1 表里**每一个**帧类型 × 三种角色:`owner` / `admin` 两类都收得到,`viewer` 只收白名单)+ 帧类型 × 角色的直接矩阵 + 「白名单精确等于 7 个状态信号」与「未登记类型默认拒」的锁定;变异验证:把 `graph_update` 改回全广播 → 「viewer 收不到 `graph_update`」必须 FAIL(已跑,见提交说明),把 `session_idle` 改走受限路径 → 「viewer 收得到 `session_idle`」必须 FAIL(已跑)。⑥ **已知残留(本次未动,留待主人裁决)**:`/ws/chat` 的**上行**仍未按角色授权(`viewer` 仍可发 `chat` / `interrupt` / `subagent_ask` / `task_output_query` / `voice_input`,其应答是**定向**帧、不经过本门),且握手时的 `history` 回放仍对任意角色照发(`thinking` 已剥离)——二者都在 design/11 §7.1 本次门的作用域之外。 - `v0.47`(**`graph_update` 子代理泳道根节点新增可选 `props.persona`**,对客户端**纯增量**;实现依据 design/22 §8-2「跨角色派发 = 把其他角色以子代理形式派出去」):① **`subagent_spawn` 指定目标角色派发时**,该子代理在父会话图里的**分支根节点** `props` 多一个 `persona` 键(值 = 目标角色 id),供流程图显示"这个子代理是以谁的身份在替班",而不只是子代理 id。② **不指定角色派发时该键不出现**:帧形状与 v0.46 逐字一致,老客户端忽略未知键即可。③ **无新端点、无字段改名、无鉴权变更**;`subagent_event` 帧**未新增字段**(其 `role` 会等于目标角色——既有字段的取值语义变化,非形状变更)。④ **`subagent_spawn` 的 `persona` 参数不在本文档契约内**:那是给**模型**用的工具 schema(工具层以「进程能否按该角色运行」为判据,未知角色整批拒绝且不产生任何子代理),不是给客户端的 HTTP/WS 契约。⑤ **验证**:进程内 e2e 断言「子代理实际收到的 system prompt = 目标角色 instructions」+「其记忆写入落在**目标角色**域」+「未知角色整批拒绝」+「**省略 `persona` 时结果 JSON 键集与旧版完全相同**」;变异验证(把目标角色换回父/默认角色、关掉角色校验)对应断言均 FAIL。 - `v0.46`(**会话角色不可变:`PATCH /sessions/{id}` 拒绝 `personaId`**,对客户端**是破坏性变更**——按 v0.42–v0.45 旧文实现、依赖「PATCH 换角色」的客户端必须改为**新建会话**选角;design/22 §8-1 主人拍板定稿):① **`PATCH /api/v1/sessions/{id}` 的 `personaId` 改为「**出现即 `400`**」**:只要请求体里**有 `personaId` 键**(任何取值:`"yuki"` / `""` / `null` / 未知 id / 空白串)就回 `400 bad_request`,**错误消息点名 `personaId`** 并说明「会话角色在创建时确定、之后不可变,要换角色请新开会话」。**实现取「明确拒绝」而非静默忽略或 200 空操作**:静默忽略会让客户端以为绑定已改。② **顺序保证**:该检查在解析完 body 后**立即**执行、**先于**会话存在性检查 —— **未知会话 + `personaId` 回 `400`(不是 `404`)**;`title` + `personaId` 同传也整体 `400`,**其余字段一个都不落库**。③ **不变的部分**:`title` / `pinned` 的语义、边界(≤80 rune 等)与响应形状**逐字不变**;`POST /api/v1/sessions {personaId}`(**创建时**绑定、未知角色 `400`、非空回显)**完全不变**;`GET /sessions` 与消息 DTO 的只读 `personaId` 不变。④ **移除的路径**:PATCH 上再也没有 `{personaId}` 成功响应、也没有「不支持角色绑定的 HistoryStore → `501`」这条路(网关的 `Server.setSessionPersona` 及其中间层随之删除;`history.Store.SetSessionPersona` 作为存储能力保留)。⑤ **前端同步**:用户端**移除**「会话行 → 切换角色」菜单与面板「挂角色」下拉,**保留**「新建会话时选角色」;管理面板「角色」列改为**只读**展示创建时绑定的角色。⑥ **测试与变异验证**:新增断言覆盖「键出现即 400(六种取值)+ 绑定未变 + 未知会话仍 400 + `title`/`pinned` 不变」,并把「静默忽略」与「改回写绑定」两种反向实现各跑一次(两处断言均必须 FAIL,见提交说明)。 - `v0.45`(**渠道 `personaId` 三态修订:修「省略 = 解绑」会静默清空绑定**,对客户端**无字段/端点形状变更**,但**语义收紧**——按 v0.44 旧文实现的客户端若靠「省略 `personaId`」来解绑,必须改为**显式传 `""`**;design/22 §2 方案 b):① **`PUT /api/v1/admin/channels/{platform}/config` 的 `personaId` 改为三态**:**键省略(或 JSON `null`)= 保留该账号已存绑定**(不动);**显式 `""` = 解绑**(写盘时该账号不带 `personaId` 键);**非空 id = 绑定该角色**;**未知角色 id 仍 `400 bad_request`**(且**只校验本次真的传了值**的情形——省略时继承的已存绑定在当初写入时已校验)。② **修的问题(数据安全)**:管理面板只在主人**动过**角色下拉时才发这个键(前端 `channelPersona.ts` / `channelPersonaRequest`),而 v0.44 后端把「键不存在」也当成解绑 ——**主人打开一个已绑角色的账号、不动下拉、只改 token 并保存 → 该账号的角色绑定被静默清空**(无提示、无审计可见的理由),两端语义相反。现在后端的「省略」与前端一致 = 保留。③ **实现**:请求 DTO 的该字段由 `string` 变 `*string`(指针区分「省略」与「显式空串」),应用逻辑在既有「读当前配置(channels.json)→ `UpsertChannel` → 落盘 → `bridge.Reload(platform)`」路径内按三态取值(省略时从**同平台同 id** 的已存账号继承,故保存一个账号不会动到同平台其它账号的绑定);**无新端点、无新字段、无响应形状变更**。④ **`GET /api/v1/admin/channels` 每账号的 `personaId` 回显不变**(v0.44 新增,非敏感、不掩码),面板据此回填角色下拉。⑤ **测试**:省略 → 绑定保留、显式 `""` → 清除、非空 id → 绑定、未知 id → `400`(含「省略时改其它字段后读回确认」与「同平台多账号互不影响」),并把「省略 = 解绑」作为变异反向验证(该断言必须失败)。 diff --git a/internal/services/gateway/broadcast.go b/internal/services/gateway/broadcast.go index 05f648f..0ab8d1e 100644 --- a/internal/services/gateway/broadcast.go +++ b/internal/services/gateway/broadcast.go @@ -20,7 +20,9 @@ import ( // 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. +// 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. func (s *Server) BroadcastPersisted(msg history.Message) { wire := toHistoryMessage(msg) wire.Thinking = "" @@ -40,7 +42,9 @@ const wsCapacityAwaken = "capacity_awaken" // 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). +// 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. func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool string) { s.hub.broadcast(wsEvent{ Type: wsCapacityAwaken, @@ -58,8 +62,10 @@ func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool str // `mood` handler (ws.ts) reads data.mood and shows the 心情 badge. const wsMoodEvent = "mood" -// BroadcastMood fans one emotion change to every Web client as a `mood` frame -// (design/07 §3). It is called only when the mood actually changed — the +// 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 diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 54eef56..b356378 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -897,7 +897,7 @@ func TestServerCloseShutsHub(t *testing.T) { func TestHubStreamChat(t *testing.T) { t.Parallel() hb := newTestHub(&stubRunner{res: &loop.Result{Text: "hi", Steps: 2, StopReason: "stop"}}, nil) - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) hb.streamChat(context.Background(), map[string]any{"sessionId": "s1", "text": "hello"}) @@ -943,7 +943,7 @@ func TestHubStreamChatPersistsMode(t *testing.T) { func TestHubBroadcastStream(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) // A text_delta StreamEvent on the OnStream→bus→relay path becomes a typed @@ -982,7 +982,7 @@ func TestHubBroadcastStream(t *testing.T) { func TestServerBroadcastCapacityAwaken(t *testing.T) { t.Parallel() env := newEnv(t, nil, false) - c := newFakeConn(env.srv.hub, 32) + c := newOwnerConn(env.srv.hub, 32) env.srv.hub.register(c) env.srv.BroadcastCapacityAwaken("web", "quiet", "awake", "web_search") @@ -1007,7 +1007,7 @@ func TestServerBroadcastCapacityAwaken(t *testing.T) { func TestServerBroadcastPersistedSplit(t *testing.T) { t.Parallel() env := newEnv(t, nil, false) - c := newFakeConn(env.srv.hub, 32) + c := newOwnerConn(env.srv.hub, 32) env.srv.hub.register(c) cases := []struct { @@ -1050,7 +1050,7 @@ func TestServerBroadcastPersistedSplit(t *testing.T) { func TestHubStreamChatIdle(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) // nil runner - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) hb.streamChat(context.Background(), map[string]any{"sessionId": "s1", "text": ""}) @@ -1062,7 +1062,7 @@ func TestHubStreamChatIdle(t *testing.T) { func TestHubStreamChatError(t *testing.T) { t.Parallel() hb := newTestHub(&stubRunner{err: errors.New("boom")}, nil) - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) hb.streamChat(context.Background(), map[string]any{"sessionId": "s1", "text": "hi"}) @@ -1074,7 +1074,7 @@ func TestHubStreamChatError(t *testing.T) { func TestHubRelayChatBranch(t *testing.T) { t.Parallel() hb := newTestHub(&stubRunner{res: &loop.Result{Text: "hi", Steps: 1, StopReason: "stop"}}, nil) - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) if err := hb.relay(context.Background(), bus.Event{Subject: wsChatSubject, Payload: map[string]any{"sessionId": "s1", "text": "hi"}}); err != nil { @@ -1099,7 +1099,7 @@ func TestHubRoute(t *testing.T) { hb.taskOutputs = &stubTaskOutputs{ev: task.OutputEvent{ ID: "task-1", Status: task.StatusDone, Output: task.Output{Stdout: "out", Stderr: "err"}, }} - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) // subagent_ask → subagent_event (real audit data, not a hardcoded pending). @@ -1161,7 +1161,7 @@ func TestHubVoiceInputTurn(t *testing.T) { t.Parallel() hb := newTestHub(&stubRunner{res: &loop.Result{Text: "你好", Steps: 1, StopReason: "stop"}}, nil) hb.voice = &stubVoice{sttText: "你好", tts: voice.Audio{Format: "wav", Data: []byte("audio-bytes")}} - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) // "aGVsbG8=" decodes to "hello" — a valid base64 audio payload. @@ -1187,7 +1187,7 @@ func TestHubBridgeDeviceUpdate(t *testing.T) { t.Parallel() b := bus.New() hb := newTestHub(nil, b) - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) hb.start() t.Cleanup(hb.close) @@ -1203,7 +1203,7 @@ func TestHubBridgeDeviceUpdate(t *testing.T) { func TestHubBridgeSubagentProgress(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) hb.register(c) if err := hb.relay(context.Background(), bus.Event{ @@ -1233,7 +1233,7 @@ func TestHubBridgeSubagentProgress(t *testing.T) { func TestPublishPublicNilBus(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) // nil bus - c := newFakeConn(hb, 32) + c := newOwnerConn(hb, 32) // Must not panic and must not enqueue anything. c.publishPublic(wsChatSubject, map[string]any{"text": "hi"}) } @@ -1396,6 +1396,17 @@ 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. +func newOwnerConn(hb *hub, buf int) *wsConn { + c := newFakeConn(hb, buf) + c.claims = &auth.Claims{Role: auth.RoleOwner} + return c +} + func readFakeFrame(t *testing.T, c *wsConn) wsEvent { t.Helper() select { diff --git a/internal/services/gateway/history_write_broadcast_test.go b/internal/services/gateway/history_write_broadcast_test.go index bed4e46..251a42d 100644 --- a/internal/services/gateway/history_write_broadcast_test.go +++ b/internal/services/gateway/history_write_broadcast_test.go @@ -23,7 +23,7 @@ func TestPatchMessageBroadcastsEdited(t *testing.T) { t.Parallel() ctx := context.Background() srv, ts, jwt, st := newHistoryEnv(t) - c := newFakeConn(srv.hub, 32) + c := newOwnerConn(srv.hub, 32) srv.hub.register(c) sub, err := st.NewSubSession(ctx, "owner") if err != nil { @@ -101,7 +101,7 @@ func TestDeleteMessageBroadcastsDeleted(t *testing.T) { t.Parallel() ctx := context.Background() srv, ts, jwt, st := newHistoryEnv(t) - c := newFakeConn(srv.hub, 32) + c := newOwnerConn(srv.hub, 32) srv.hub.register(c) sub, err := st.NewSubSession(ctx, "owner") if err != nil { @@ -151,7 +151,7 @@ func TestDeleteMessageBroadcastsDeleted(t *testing.T) { func TestDeleteMessageUnknownIDBroadcastsNothing(t *testing.T) { t.Parallel() srv, ts, jwt, _ := newHistoryEnv(t) - c := newFakeConn(srv.hub, 32) + c := newOwnerConn(srv.hub, 32) srv.hub.register(c) tok := signAny(t, jwt, "acct", auth.RoleOwner) diff --git a/internal/services/gateway/mood_test.go b/internal/services/gateway/mood_test.go index 05eeddb..4794663 100644 --- a/internal/services/gateway/mood_test.go +++ b/internal/services/gateway/mood_test.go @@ -85,7 +85,7 @@ func TestOrchestratorPersistsEmptyMoodWhenUntracked(t *testing.T) { func TestServerBroadcastMood(t *testing.T) { t.Parallel() env := newEnv(t, nil, false) - c := newFakeConn(env.srv.hub, 32) + c := newOwnerConn(env.srv.hub, 32) env.srv.hub.register(c) env.srv.BroadcastMood("cyrene", "happy", 60) diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index bc320c2..a70f0d4 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -184,7 +184,9 @@ 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. It is wrapped in a +// 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). func newOrchestrator(s *Server) *Orchestrator { diff --git a/internal/services/gateway/session_read.go b/internal/services/gateway/session_read.go index 8e7ede3..5f2c318 100644 --- a/internal/services/gateway/session_read.go +++ b/internal/services/gateway/session_read.go @@ -83,8 +83,10 @@ func (s *Server) handleMarkSessionRead(w http.ResponseWriter, r *http.Request) { // broadcastSessionRead fans a read-cursor advance out to every client as a // `session_read` frame carrying the session's new cursor, so the same account's -// other ends clear the session's unread badge without a re-read. Like the other -// history frames it is an all-connections broadcast (the hub has no per-session +// 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. diff --git a/internal/services/gateway/ws_frame_scope.go b/internal/services/gateway/ws_frame_scope.go new file mode 100644 index 0000000..a2888b3 --- /dev/null +++ b/internal/services/gateway/ws_frame_scope.go @@ -0,0 +1,45 @@ +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: +// +// - 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. +// +// 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. +// +// 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 「语义澄清」). + +// 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. +// +// 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. +func frameReachesAll(typ string) bool { + switch typ { + case EventSessionBusy, + EventPendingAck, + EventPendingDrain, + EventSessionIdle, + EventSessionInterrupt, + EventPendingFull, + wsSessionRead: + return true + default: + return false + } +} diff --git a/internal/services/gateway/ws_frame_scope_test.go b/internal/services/gateway/ws_frame_scope_test.go new file mode 100644 index 0000000..4b8c523 --- /dev/null +++ b/internal/services/gateway/ws_frame_scope_test.go @@ -0,0 +1,326 @@ +package gateway + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "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. + +// frameScopeCase is one row of the design/11 §7.1 table. +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 +} + +// 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. +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. +const ( + scopePaceFrames = 4 + scopePacePause = 50 * time.Millisecond +) + +// scopeRead is the outcome of one connection's reader 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. +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} + } + streamed := func(kind model.StreamKind, typ string) frameScopeCase { + return frameScopeCase{ + typ: typ, + subject: wsStreamSubject, + payload: model.StreamEvent{Kind: kind, Delta: "x", SessionID: "web", TurnID: "turn-scope"}, + } + } + return []frameScopeCase{ + // 状态机信号(白名单,不含任何内容)→ 所有已认证连接。 + relayed(EventSessionBusy, true), + relayed(EventPendingAck, true), + relayed(EventPendingDrain, true), + relayed(EventSessionIdle, true), + relayed(EventSessionInterrupt, true), + relayed(EventPendingFull, true), + relayed(wsSessionRead, true), + + // 流式内容(含角色回复正文与推理内容)→ 仅 owner/admin。 + streamed(model.StreamKindTextStart, "message.start"), + streamed(model.StreamKindTextDelta, "message.delta"), + streamed(model.StreamKindTextEnd, "message.end"), + streamed(model.StreamKindThinkingStart, "thinking.start"), + streamed(model.StreamKindThinkingDelta, "thinking.delta"), + streamed(model.StreamKindThinkingEnd, "thinking.end"), + streamed(model.StreamKindToolCallStart, "tool_call.start"), + streamed(model.StreamKindToolCallDelta, "tool_call.delta"), + streamed(model.StreamKindToolCallEnd, "tool_call.end"), + streamed(model.StreamKindToolExecution, "tool_execution.update"), + relayed("stream_chunk", false), + relayed("stream_end", false), + relayed("thinking", false), + + // 持久化内容 → 仅 owner/admin。 + relayed("message_persisted", false), + relayed("message_edited", false), + relayed("message_deleted", false), + relayed("channel_persisted", false), + + // 系统内部 → 仅 owner/admin。 + relayed("graph_update", false), + relayed("subagent_event", false), + relayed("subagent.result", false), + relayed("task_output", false), + relayed("task_done", false), + relayed("automation_result", false), + relayed("tool_used", false), + relayed("capacity_awaken", false), + relayed("mode_switch", false), + + // 私有状态 → 仅 owner/admin。 + relayed("mood", false), + relayed("device_update", false), + relayed("reminder", false), + } +} + +// 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. +func TestWSFrameScopePerRole(t *testing.T) { + t.Parallel() + cases := frameScopeCases() + env := newEnv(t, nil, false) + + conns := map[auth.Role]*websocket.Conn{} + for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} { + conn := dialWS(t, env.ts.URL, env.token(t, role)) + defer conn.Close() + conns[role] = conn + } + waitConnCount(t, env.srv, len(conns)) + + all := append(frameTypes(cases, false), scopeSentinel) + public := append(frameTypes(cases, true), scopeSentinel) + wants := map[auth.Role]int{ + auth.RoleOwner: len(all), + auth.RoleAdmin: len(all), + 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. + results := make(chan scopeRead, len(conns)) + for role, want := range wants { + role, want := role, want + go func() { + types, err := readFrameTypes(conns[role], want) + results <- scopeRead{role: role, types: types, err: err} + }() + } + publishFrameScopeCases(t, env, cases) + + got := make(map[auth.Role][]string, len(conns)) + for range conns { + res := <-results + if res.err != nil { + t.Fatalf("%s reader: %v", res.role, res.err) + } + got[res.role] = res.types + } + + for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} { + assertFrameTypes(t, string(role), got[role], all) + } + gotViewer := got[auth.RoleViewer] + assertFrameTypes(t, string(auth.RoleViewer), gotViewer, public) + for _, denied := range []string{"message.delta", "thinking.delta", "message_persisted", "graph_update", "mood"} { + if containsFrameType(gotViewer, denied) { + t.Fatalf("viewer received %q; it must not (design/11 §7.1)", denied) + } + } + if !containsFrameType(gotViewer, EventSessionIdle) { + t.Fatalf("viewer missing %q, a whitelisted state signal (got %v)", EventSessionIdle, gotViewer) + } +} + +// 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. +func TestHubBroadcastScopeMatrix(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + owner := newFakeConn(hb, 4) + owner.claims = &auth.Claims{Role: auth.RoleOwner} + admin := newFakeConn(hb, 4) + admin.claims = &auth.Claims{Role: auth.RoleAdmin} + viewer := newFakeConn(hb, 4) + viewer.claims = &auth.Claims{Role: auth.RoleViewer} + for _, c := range []*wsConn{owner, admin, viewer} { + hb.register(c) + } + + for _, tc := range frameScopeCases() { + tc := tc + t.Run(tc.typ, func(t *testing.T) { + hb.broadcast(wsEvent{Type: tc.typ, SessionID: "web"}) + for _, c := range []*wsConn{owner, admin} { + if ev := readFakeFrame(t, c); ev.Type != tc.typ { + t.Fatalf("owner/admin frame = %q, want %q", ev.Type, tc.typ) + } + } + if tc.public { + if ev := readFakeFrame(t, viewer); ev.Type != tc.typ { + t.Fatalf("viewer frame = %q, want %q (whitelisted state signal)", ev.Type, tc.typ) + } + return + } + if err := connHasNoFrame(viewer); err != nil { + t.Fatalf("viewer received the restricted frame %q: %v", tc.typ, err) + } + }) + } +} + +// 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. +func TestFrameScopeAllowsExactlyTheStateSignals(t *testing.T) { + t.Parallel() + if frameReachesAll("frame_type_nobody_classified") { + t.Fatal("an unclassified frame type reached every connection; the default must be owner/admin only") + } + public := make([]string, 0, len(frameScopeCases())) + for _, tc := range frameScopeCases() { + if got := frameReachesAll(tc.typ); got != tc.public { + t.Fatalf("frame %q: gate says public=%v, design/11 §7.1 says public=%v", tc.typ, got, tc.public) + } + if tc.public { + public = append(public, tc.typ) + } + } + want := []string{ + EventSessionBusy, + EventPendingAck, + EventPendingDrain, + EventSessionIdle, + EventSessionInterrupt, + EventPendingFull, + wsSessionRead, + } + sort.Strings(public) + sort.Strings(want) + 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). +func frameTypes(cases []frameScopeCase, publicOnly bool) []string { + out := make([]string, 0, len(cases)) + for _, tc := range cases { + if publicOnly && !tc.public { + continue + } + out = append(out, tc.typ) + } + 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. +func publishFrameScopeCases(t *testing.T, env *testEnv, cases []frameScopeCase) { + t.Helper() + for i, tc := range cases { + if err := env.bus.Publish(context.Background(), tc.subject, tc.payload); err != nil { + t.Fatalf("publish %s: %v", tc.subject, err) + } + if (i+1)%scopePaceFrames == 0 { + time.Sleep(scopePacePause) + } + } + if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil { + t.Fatalf("publish sentinel: %v", err) + } +} + +// 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. +func readFrameTypes(conn *websocket.Conn, want int) ([]string, error) { + got := make([]string, 0, want) + for len(got) < want { + _ = conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + return got, fmt.Errorf("read frame %d/%d: %w (got %v)", len(got)+1, want, err, got) + } + var ev wsEvent + if err := json.Unmarshal(msg, &ev); err != nil { + return got, fmt.Errorf("decode frame %s: %w", msg, err) + } + got = append(got, ev.Type) + } + return got, nil +} + +// assertFrameTypes compares the received frame types to the expected sequence. +func assertFrameTypes(t *testing.T, what string, got, want []string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("%s frame count = %d, want %d (got %v)", what, len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("%s frame[%d] = %q, want %q (got %v)", what, i, got[i], want[i], got) + } + } +} + +// containsFrameType reports whether typ is among the received frame types. +func containsFrameType(types []string, typ string) bool { + for _, got := range types { + if got == typ { + return true + } + } + return false +} diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index e057a89..91dd516 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -48,7 +48,8 @@ type wsInbound struct { Data json.RawMessage `json:"data,omitempty"` } -// hub owns the connection set and relays bus events to every client. +// hub owns the connection set and relays bus events to clients, each frame +// gated by its type (see broadcast / ws_frame_scope.go). type hub struct { mu sync.RWMutex conns map[*wsConn]struct{} @@ -157,17 +158,30 @@ func (h *hub) connCount() int { return len(h.conns) } -// broadcast fans an event out to every registered connection without blocking on -// any single (possibly slow) client: a full send buffer drops the frame. +// 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. +// +// 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. func (h *hub) broadcast(ev wsEvent) { b, err := json.Marshal(ev) if err != nil { h.log.Warn("marshal broadcast", "err", err) return } + adminOnly := !frameReachesAll(ev.Type) h.mu.RLock() defer h.mu.RUnlock() for c := range h.conns { + if adminOnly && !ownerAdminClaims(c.claims) { + continue + } c.enqueue(b) } } diff --git a/internal/services/gateway/ws_hub_message_event.go b/internal/services/gateway/ws_hub_message_event.go index 586976c..c9bdf04 100644 --- a/internal/services/gateway/ws_hub_message_event.go +++ b/internal/services/gateway/ws_hub_message_event.go @@ -2,8 +2,9 @@ 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 all-connections fan-out as -// `message_persisted` — the hub has no per-session subscription, so the frame +// 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. const ( @@ -11,8 +12,8 @@ const ( wsMessageDeleted = "message_deleted" ) -// broadcastMessageEdited fans a content edit out to every client as a -// `message_edited` frame whose data is the updated message DTO. The frame is a +// 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. @@ -20,7 +21,7 @@ func (h *hub) broadcastMessageEdited(msg historyMessage) { h.broadcast(wsEvent{Type: wsMessageEdited, SessionID: msg.SessionID, Data: msg}) } -// broadcastMessageDeleted fans a removal out to every client as a +// 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; diff --git a/internal/services/gateway/ws_hub_stream_scope_test.go b/internal/services/gateway/ws_hub_stream_scope_test.go index 1c22928..02140b8 100644 --- a/internal/services/gateway/ws_hub_stream_scope_test.go +++ b/internal/services/gateway/ws_hub_stream_scope_test.go @@ -53,7 +53,7 @@ func newScopedStreamHub(t *testing.T) (*hub, *wsConn) { hb.streaming = true hb.start() t.Cleanup(hb.close) - c := newFakeConn(hb, 64) + c := newOwnerConn(hb, 64) hb.register(c) return hb, c } diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index 202f414..f10ec4d 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -125,7 +125,9 @@ func (h *hub) handleInterrupt(payload json.RawMessage) { } // broadcastStream maps a streamed model.StreamEvent to its downstream WS event -// type and fans it out to every client. It runs on the bus subscription worker; +// 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