From 84c124f5a9588ad626c573f6ac9acf16f2aea1ae Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 16:02:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(backthink):=20=E5=A5=B9=E4=B8=BB=E5=8A=A8?= =?UTF-8?q?=E6=89=BE=E4=BD=A0=E7=BB=88=E4=BA=8E=E7=9C=8B=E5=BE=97=E8=A7=81?= =?UTF-8?q?=20=E2=80=94=E2=80=94=20=E4=B8=BB=E5=8A=A8=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E8=90=BD=E4=B8=BB=E4=BC=9A=E8=AF=9D=20+=20=E5=89=8D=E7=AB=AF?= =?UTF-8?q?=E6=B6=88=E8=B4=B9=20+=20=E9=9D=9E=E5=89=8D=E5=8F=B0=E6=B5=8F?= =?UTF-8?q?=E8=A7=88=E5=99=A8=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修主人报的硬故障「后台思考的主动推送一个都没有推出来」（根因两条，缺一不可）： - 后端 gateway/proactive.go 只广播 proactive_message 帧、**从不写会话历史** ⇒ 刷新补不回来 - 前端**零消费点** ⇒ 帧被静默丢弃（既无气泡也无通知） 后端：新增 proactive_persist.go（**先落库、再发帧**；web/assistant/mode=chat；未接线时空操作且不谎称已落库） 载荷新增 id/createdAt（落库行权威 id）；PushResult.Detail 三态如实分开报告「投递到几个连接」与「是否落库」 前端：handlers/proactive.ts（气泡**恒落** + 按落库 id 去重 + **非前台才补通知** + 拒绝授权/畸形载荷可见降级） + lib/notify/proactive.ts（纯逻辑 isBackground/decideProactiveNotice/有界去重表） + handlers/task.ts（另三个被丢的帧 task_done/automation_result/subagent_result） + ws/resync.ts + useWsResync.ts（**重连重跑本页引导读取**，首次连接不跑） 双份核查（带三层证据）：现状不双份 —— 路由分叉「一次 Push 至多一个 sink 成功」，且 IM 路由本来就把主动消息镜像回主会话恰好一次 契约 v0.72；变异实测：只发帧不落库 / 写成 user 角色 / 去重失效 / 前台也弹 / IM 成功再打活跃 ⇒ 全部打红 --- Docs/api-client.md | 30 +- frontend/web/src/App.tsx | 13 +- frontend/web/src/lib/notify/proactive.ts | 84 ++++++ frontend/web/src/lib/useWsResync.ts | 20 ++ frontend/web/src/lib/ws.ts | 38 ++- .../web/src/lib/ws/handlers/proactive.test.ts | 253 ++++++++++++++++ frontend/web/src/lib/ws/handlers/proactive.ts | 198 +++++++++++++ frontend/web/src/lib/ws/handlers/task.test.ts | 104 +++++++ frontend/web/src/lib/ws/handlers/task.ts | 96 ++++++ frontend/web/src/lib/ws/resync.test.ts | 110 +++++++ frontend/web/src/lib/ws/resync.ts | 44 +++ internal/services/backthink/mocks_test.go | 16 +- internal/services/backthink/push.go | 68 ++++- .../backthink/push_active_delivery_test.go | 117 ++++++++ internal/services/gateway/proactive.go | 57 +++- .../services/gateway/proactive_persist.go | 51 ++++ .../gateway/proactive_persist_test.go | 280 ++++++++++++++++++ internal/services/gateway/proactive_test.go | 12 +- 18 files changed, 1545 insertions(+), 46 deletions(-) create mode 100644 frontend/web/src/lib/notify/proactive.ts create mode 100644 frontend/web/src/lib/useWsResync.ts create mode 100644 frontend/web/src/lib/ws/handlers/proactive.test.ts create mode 100644 frontend/web/src/lib/ws/handlers/proactive.ts create mode 100644 frontend/web/src/lib/ws/handlers/task.test.ts create mode 100644 frontend/web/src/lib/ws/handlers/task.ts create mode 100644 frontend/web/src/lib/ws/resync.test.ts create mode 100644 frontend/web/src/lib/ws/resync.ts create mode 100644 internal/services/backthink/push_active_delivery_test.go create mode 100644 internal/services/gateway/proactive_persist.go create mode 100644 internal/services/gateway/proactive_persist_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 939f086..b34814e 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -1,6 +1,6 @@ # Cyrene v2 客户端 API 契约 -> **契约版本：`v0.71`**（上一版 `v0.70`，其条目见 §6；`v0.69` 及更早见本文件 §6，`v0.67` 及更早的基线已归档、只读：`Docs/archive/api-client.v0.65.md`）。 +> **契约版本：`v0.72`**（上一版 `v0.71`，其条目见 §6；`v0.70` 及更早见本文件 §6，`v0.67` 及更早的基线已归档、只读：`Docs/archive/api-client.v0.65.md`）。 > **读者**：Web 前端 / 移动端原生客户端（含 Android）/ IM 桥接等**跨平台客户端开发者**。 > **唯一真源 = 当前代码**（`internal/services/gateway/*` 等）。本文档由代码逐条核对得出；与代码不符时**以代码为准**，并回改本文档。 > **旧的超详细版已归档、只读**：`Docs/archive/api-client.v0.65.md`（412 KB，含 v0.28→v0.65 全部历史条目）与 `Docs/archive/api-reference.v0.65.md`（170 KB，后端全量字段级文档）。归档的原因是**结构性臃肿 + 历史条目并列 + 长期漂移**；本文档**不复制**那段历史，需要考古时去归档。 @@ -12,6 +12,8 @@ > **v0.70 是精确化版（已实现，纯增量）**：`GET /api/v1/admin/backthink` 的 `urgency` **默认就是该轮实际生效的精确档位**（随记录落库），另加可选 `urgencyLowerBound:true` 专门标记**旧行的下限推导值**；实时 `backthink` 帧自本版起也带记录里的 `urgency`（§2.2 B8 / §4.10(b)，并更正 §4.10(b) 原「实时帧不带这两键」的表述）。**无破坏性变更**（新键可选，老客户端不读即零影响；`urgency` 的既有取值域与 `quietHoursBypassed` 语义都未改）。 > **v0.71 是链路图补版（已实现，纯增量）**：新增 1 条管理只读端点 **`GET /api/v1/admin/flow/graph`**，把 `ws.graph_update` 的**最近一轮快照**从「只能靠实时帧拿」变成「打开页面即可只读拉取」（§2.2 B13 / §4.11）。修的是「页面在轮次结束后才打开（或刷新过）⇒ `/#/admin/flow` 与后台思考行的『查看完整链路导图』永远空着」—— `graph_update` **只走实时帧**，前端图桶（`flowGraphs`）**不持久化**，此前**没有任何**补拉通路。**无破坏性变更**（新端点；`graph_update` 帧形状、角色门、发布会话范围**一字未改**）。 +> **v0.72 是主动推送可见性版（已实现，含一处行为变更）**：`proactive_message` 那条消息现在**落进主会话历史**（`web` / `assistant` / `mode:"chat"`，先落库再发帧），帧新增可选 `id`/`createdAt`（主会话历史行的权威 id 与时间戳）供客户端把实时气泡对齐到历史行（§3.4.5 与随后的「落库口径」说明）。修的是主人 2026-10-05 报的「后台思考的主动推送一个都没推出来」—— 活跃客户端路由此前**只广播帧、一行都不写**，刷新即丢且前端零消费点。**两条路由互斥 ⇒ 主会话里恰好一条（不双份）**；`proactive_message` 帧**逐字不变的部分**是既有全部键，客户端不读新键即行为与 v0.71 相同（只会多看到那条历史行）。 + --- ## 0. 基础约定 @@ -388,7 +390,7 @@ | `graph_update` | — | 实时 flow 图快照（`data.sessionId` 标会话；**无 `turnId`** ⇒ 未按路分桶）。⚠️ **纯实时、不重放**：断开/刷新期间的那一轮不会补发 ⇒ 打开页面时用 `GET /api/v1/admin/flow/graph`（§2.2 B13 / §4.11）拉**最近一轮快照** | | `capacity_awaken` | sessionId | `{fromState,toState,tool}` | | `mood` | （无 sessionId） | `{mood,personaId,affection}`；角色级情绪，仅变化时发 | -| `proactive_message` | — | `{platform,channelId,channelType,content,contentType,sender,timestamp,messageId}`（已消毒，绝不含凭据/媒体字节） | +| `proactive_message` | — | `{platform,channelId,channelType,content,contentType,sender,timestamp,messageId,id?,createdAt?}`（已消毒，绝不含凭据/媒体字节）。**v0.72**：`id`/`createdAt` 是这条消息在**主会话历史**里的权威行 id 与落库时间戳（同现/同省），落库失败时键缺席 —— 客户端据此把实时气泡对齐到历史行（按 id 去重、刷新后仍在） | | `approval_notice` | sessionId | `{actor,tool,op,detail,explain,allowed,at}`（主代理代审小字；`explain` 由后端产出、客户端原样渲染） | | `mode_switch` | — | 上行 `mode_switch` 的回放（透传 `data`） | | `observability_log` | — | `{entries:[{time,level,source,msg}...]}`，上限 200，最新在前 | @@ -397,6 +399,12 @@ > **observability_* 的订阅**：owner/admin 凭据**读取** `GET /admin/observability/{logs,metrics,trace}` 任一个即订阅；该凭据的 WS 连接全断即退订（无轮询、无常驻 goroutine）。 +> **`proactive_message` 的落库口径（v0.72，行为变更）**：后台思考决定「此刻主动开口」时，这条消息现在是**主会话（`web`）历史里的一条真实 `assistant` 消息** —— 走活跃客户端路由（在线 owner/admin 连接）的那一路此前**只广播帧、一行都不写** ⇒ 刷新/重载后什么都补不回来（既不显示气泡，也补不出历史）。本版 `SendProactive` **先落库、再发帧**： +> - 落库行：`sessionId="web"`、`role="assistant"`、`content` = 帧里的 `content`（后端已净化的同一份文本，系统元数据绝不进正文）、`platform` = 帧里的 `platform`、`mode="chat"`、`turn_id` **留空**（主动推送不属于任何一路回合）。因此 `GET /sessions/web/messages`、握手 `history` 回放、`message_persisted` 都能读到它。 +> - **`id`/`createdAt` 随帧下发**，客户端应据此把实时气泡对齐到历史行（按 id 去重；刷新后由历史行接续，不是第二个气泡）。**落库失败时这两个键缺席**，但同时：帧照发（实时气泡不丢），后端记 WARN 并返回错误 ⇒ 后台思考记录里的 `route/detail` 会写明**未落库**（不会再说成「已推送」）。 +> - **无在线 owner/admin 连接时也会落库**：推送器如实报告「没有活跃客户端」（`pushed=false`），这条话仍作为一条真实记录留在主会话里，下次打开就能看到。 +> - **IM 路由不需要客户端做任何事**：`im_bridge` 路由（IM 桥）本就把消息镜像回主会话并广播 `message_persisted`（逐条一条，**不双份**）。两条路由在一次推送里**互斥**（`web` 提示只走活跃客户端，其余先 IM、IM 投递失败才降级活跃客户端），因此同一条主动消息在主会话里**恰好一条**。 + #### 3.4.6 定向帧（不经 `hub.broadcast`，只到指定连接） | type | 说明 | @@ -814,7 +822,21 @@ > 只保留最近若干版（**v0.60 起**）；更早的全部历史见归档 `Docs/archive/api-client.v0.65.md` §5（v0.28 → v0.59）。 > 每版一句话 + 三档标注。**三档定义见 [§0.4](#04-三档标注惯例)。** -### v0.71（当前契约版本） +### v0.72（当前契约版本） + +**主动推送可见性 —— 主动消息成为主会话里的一条真实消息，并给帧补上权威行 id**（依据主人 2026-10-05 报「后台思考的主动推送一个都没推出来」；对齐 `design/12 §3` 的主动推送与 `design/16 §4.4` 的镜像口径）。 + +- **行为变更**： + ① **`proactive_message` 那条消息现在会落进主会话历史**（`web`、`role: assistant`、`mode: "chat"`、`turn_id` 留空）。此前后台思考走**活跃客户端路由**时只广播 WS 帧、**一行都不写** ⇒ 刷新/重载后既不显示气泡、也补不出历史；那句 `delivered to 2 active client(s)` 只表示「帧被 2 个连接收下」，**不代表进了会话**。顺序改为**先落库、再发帧**。 + ② **帧新增两个可选键 `id` / `createdAt`**（同现/同省）：这条消息在主会话历史里的权威行 id 与落库时间戳。客户端据此把实时气泡对齐到历史行（按 id 去重、刷新后由历史行接续）。**落库失败时两键缺席** —— 客户端不应把「没有 id」当成一条可持久化的消息。 + ③ **没有在线 owner/admin 连接时也落库**：推送器如实回 `pushed=false`（没有活跃客户端），但那句话仍作为真实记录留在主会话里。 + ④ **后台思考记录里的 `PushResult.Detail` 措辞**：活跃客户端路由改为 `delivered to N active client(s), persisted to main session`（未落库时写明 `not persisted…`，无在线客户端但已落库时写明 `no active client, recorded in main session`）—— 面板因此不会把「帧发出去了、刷新就没了」误读成「已推送」。 +- **纯增量**：帧的形状只在**新增可选键**；落库行的形状就是既有 `historyMessage`（`§4.4`），因此既有读取面（`GET /sessions/web/messages`、握手 `history` 回放、`message_persisted`）**逐字不变**。 +- **破坏性**：无。老客户端按未知键忽略 `id`/`createdAt` 即可；但它会把主动消息**多看到一次**（历史行）—— 这是本版的目的，不是缺陷。 +- **不双份（已验证）**：两条路由在一次推送里**互斥**（`web` 提示只走活跃客户端；其余先 IM 桥、IM 投递失败才降级活跃客户端），而 IM 路由本就把消息镜像回主会话 ⇒ 同一条主动消息在主会话里**恰好一条**。契约细节见 §3.4.5 后的「`proactive_message` 的落库口径」。 +- ⚠️ **诚实边界**：被频率门（`PushGate`）拒绝的推送**不落库**（限流就是「这一条不发」，绝不留下一条它没发过的记录）。 + +### v0.71（上一版） **链路图补版 —— 新增只读端点 `GET /api/v1/admin/flow/graph`，让「轮次结束后才打开的页面」也能看到最近一轮的链路图**（依据主人 2026-10-05 第二次报「`/#/admin/flow` 停在暂无流程图数据 + 后台思考行提示该会话还没有 graph_update 快照」；对齐 `design/19 §4` 的 **P3**「多轮历史回放」里的最小一步）。 @@ -994,6 +1016,8 @@ ### 7.3 诚实边界（既有事实） - `subagent_result` **不落库**：父会话转录里没有新消息、**重连回放不补发**。 +- **`proactive_message`（v0.72 起）落库**：成为主会话 `web` 的一条 `assistant` 行（见 §3.4.5 后的「落库口径」）。**仍未落库的情形**（都是如实、预期的）：① 被 `PushGate`（频率门 / 冷却 / 按风格窗口）或安静时段挡下的推送 —— 「这一条不发」就没有记录，**绝不留下一条它没发过的消息**；② 后端未接线历史存储（`HistoryStore: nil`，仅测试与异常部署）时只发帧不写库，进程重启即丢。 +- **`task_done` / `automation_result` / `subagent_result` 不是会话消息**：它们**不进消息 store、不是气泡**（后端也没有把它们写成 `chat_messages` 行；`subagent_result` 更是**错过即永久丢**）。客户端应把它们落到自己的可观测面（Web 端落在本端 WS 事件回显 + 任务日志）。 - `GET /api/v1/files/{name}` **无鉴权**（能力 URL）；文档读侧相反、**必须 Bearer**。 - `GET /admin/llm/usage` **不按 `session_id`** 过滤/分组（尽管该列已落库）。 - `POST /audio/speech` 的请求体 `voice`/`format` **当前不生效**（合成遵循服务端 `VOICE_TTS_VOICE_ID`/`VOICE_TTS_FORMAT`）。 diff --git a/frontend/web/src/App.tsx b/frontend/web/src/App.tsx index 1a25647..ccbe401 100644 --- a/frontend/web/src/App.tsx +++ b/frontend/web/src/App.tsx @@ -6,6 +6,7 @@ import { AdminApp } from "./admin/AdminApp"; import { Login } from "./components/Login"; import { parseAdminRoute, readAdminEntry, subscribeAdminRoute } from "./admin/router"; import { connectWs, disconnectWs } from "./lib/ws"; +import { useWsResync } from "./lib/useWsResync"; import { markSessionOpenedRead } from "./lib/sessionRead"; import { initTheme } from "./lib/theme"; import { @@ -33,6 +34,12 @@ export default function App(): React.ReactNode { const [menuOpen, setMenuOpen] = useState(false); const setPersonas = useAppStore((s) => s.setPersonas); const seedMoods = useAppStore((s) => s.seedMoods); + // WS 重连后重跑本页的引导读取（切片 P0 的顺带件）：握手 history 回放只补**消息**， + // 会话列表 / 角色目录 / 消息页这些初始 GET 在断线窗口里的变化没人补，页面会一直 + // 停在旧态。这里只递增一个世代号，下面的加载副作用把它列进依赖即可（首次连接不触发， + // 页面挂载时的初始加载已经做过一遍）。 + const [resyncNonce, setResyncNonce] = useState(0); + useWsResync(() => setResyncNonce((n) => n + 1)); useEffect(() => { initTheme(); @@ -84,7 +91,7 @@ export default function App(): React.ReactNode { cancelled = true; if (retry !== null) window.clearTimeout(retry); }; - }, [token, setPersonas, seedMoods]); + }, [token, setPersonas, seedMoods, resyncNonce]); const sessions = useAppStore((s) => s.sessions); const activeSessionId = useAppStore((s) => s.activeSessionId); @@ -116,7 +123,7 @@ export default function App(): React.ReactNode { return () => { cancelled = true; }; - }, [token, setSessions, setActiveSession]); + }, [token, setSessions, setActiveSession, resyncNonce]); // Whenever the focused session changes (switch, new, or first load) and it is // a session we know about, pull its initial page of history into the panel, then @@ -141,7 +148,7 @@ export default function App(): React.ReactNode { return () => { cancelled = true; }; - }, [activeSessionId, activeSessionKnown, setMessages]); + }, [activeSessionId, activeSessionKnown, setMessages, resyncNonce]); // The M9 modal UI and the admin subsystem (design/10) are two separate shells. // 面板由安全入口路径进入（见上方 route 注释），每个模块是前缀下的一段路径。 diff --git a/frontend/web/src/lib/notify/proactive.ts b/frontend/web/src/lib/notify/proactive.ts new file mode 100644 index 0000000..41bd443 --- /dev/null +++ b/frontend/web/src/lib/notify/proactive.ts @@ -0,0 +1,84 @@ +/** + * 主动消息的**浏览器通知**判定（切片 P0「她主动找你」）。 + * + * 本模块是**纯逻辑**：不碰 DOM、不碰 store、不读全局变量，因此可以在没有 jsdom 的 + * 纯 node 下逐条钉住主人裁定的口径 —— + * + * 1. **气泡恒落，通知可省**：气泡（消息 store）与前后台无关，永远落；浏览器通知 + * 只是「人不在看的时候补一发」，是提醒、不是记录。 + * 2. **只在不在前台时补一发**：判定 = Page Visibility **且** 焦点。只判可见性会漏掉 + * 「窗口被别的窗口盖住、visibilityState 仍是 visible」这一半 —— 那种时候主人 + * 根本看不见，正该提醒。 + * 3. **同一条只打扰一次**：去重表按消息 id 记（含「同一条既走实时帧、又被 IM 镜像 + * 回主会话」这种同时到达的情形）。气泡**不参与**提醒去重 —— 它是记录，去重由 + * 消息 store 按 id 负责。 + * 4. **拒绝授权 ⇒ 如实降级**：没授权就只留气泡，并让调用方留一句可见说明； + * **绝不静默丢弃**，也绝不当成故障崩掉。 + */ + +/** 浏览器通知权限（`Notification.permission` 的三态；运行时不支持该 API 时由调用方 + * 归到 `"denied"` —— 拿不到通知就是没授权，绝不假装已授权）。 */ +export type NotifyPermission = "granted" | "denied" | "default"; + +/** 一次提醒的判定结果。 */ +export type NoticeDecision = + /** 不在前台且已授权 ⇒ 补一发系统通知。 */ + | "show" + /** 同一条已经提醒过（或已在前台露过面）⇒ 不打扰第二次。 */ + | "duplicate" + /** 页面可见且聚焦 ⇒ 按裁定不弹（气泡已经落了）。 */ + | "foreground" + /** 通知权限未授权/被拒 ⇒ 如实降级为只留气泡。 */ + | "not-granted"; + +/** 「不在前台」的判定：**可见且聚焦**才算前台。`visibility` 用 + * `document.visibilityState` 的原值（"visible" | "hidden" | …），未知取值一律按 + * 「不在前台」处理 —— 提醒多发一次不可怕，人错过了才可怕。 */ +export function isBackground(visibility: string, focused: boolean): boolean { + return !(visibility === "visible" && focused); +} + +/** 决定这一刻要不要为这条主动消息补一发通知。纯函数：同样的输入永远同样的结论。 */ +export function decideProactiveNotice(input: { + /** 同一条消息此前是否已经提醒过（调用方查去重表）。 */ + seen: boolean; + /** `document.visibilityState`。 */ + visibility: string; + /** `document.hasFocus()`。 */ + focused: boolean; + /** 当前通知权限。 */ + permission: NotifyPermission; +}): NoticeDecision { + if (input.seen) return "duplicate"; + if (!isBackground(input.visibility, input.focused)) return "foreground"; + if (input.permission !== "granted") return "not-granted"; + return "show"; +} + +/** 有界的「已提醒」表。它是**本页存活期间的打扰记录**，不是消息台账：超过上限就 + * 淘汰最旧的一条（一条很老的重复消息至多多提醒一次，绝不会让表无限长）。 */ +export function createSeenNotices(limit: number): { + has(id: string): boolean; + add(id: string): void; + size(): number; +} { + const order: string[] = []; + const seen = new Set(); + return { + has(id) { + return seen.has(id); + }, + add(id) { + if (!id || seen.has(id)) return; + seen.add(id); + order.push(id); + while (order.length > limit) { + const oldest = order.shift(); + if (oldest !== undefined) seen.delete(oldest); + } + }, + size() { + return seen.size; + }, + }; +} diff --git a/frontend/web/src/lib/useWsResync.ts b/frontend/web/src/lib/useWsResync.ts new file mode 100644 index 0000000..e1e1140 --- /dev/null +++ b/frontend/web/src/lib/useWsResync.ts @@ -0,0 +1,20 @@ +import { useEffect, useRef } from "react"; +import { registerResync } from "./ws/resync"; + +/** + * 把本页的**引导读取**登记为「WS 重连后要重跑一次」的工作（切片 P0 的顺带件）。 + * + * 用法：页面的初始 GET（会话列表 / 消息页 / 角色目录 / 快照补拉 …）抽成一个函数交给 + * 本 hook 一次即可 —— 重连时它会被重跑，断线窗口里错过的变化因此补得回来；首次连接 + * **不会**重跑（页面挂载时的初始加载已经做过一遍，见 `ws.ts#handleSocketOpen`）。 + * + * 传进来的函数放在 ref 里，因此每次渲染换闭包都不会反复登记/退订；卸载时干净退订， + * 不留悬挂回调（前端审查清单：内存泄漏）。 + */ +export function useWsResync(fn: () => void): void { + const ref = useRef(fn); + useEffect(() => { + ref.current = fn; + }, [fn]); + useEffect(() => registerResync(() => ref.current()), []); +} diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index 66bf72e..7068e50 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -13,6 +13,9 @@ import { handlePersonaEvent } from "./ws/handlers/persona"; import { handleParallelEvent } from "./ws/handlers/parallel"; import { handleDeviceEvent } from "./ws/handlers/device"; import { handleApprovalEvent } from "./ws/handlers/approval"; +import { handleProactiveEvent } from "./ws/handlers/proactive"; +import { handleTaskEvent } from "./ws/handlers/task"; +import { notifyResync } from "./ws/resync"; /** * Realtime client for `/ws/chat`. Reconnects with exponential backoff and @@ -24,6 +27,9 @@ let ws: WebSocket | null = null; let shouldRun = false; let reconnectTimer: number | null = null; let attempt = 0; +/** 这条页面生命周期里是否**曾经**打开过连接。用来把「首次连接」与「重连」分开： + * 首次连接的引导读取由页面自己的挂载副作用负责，重连才需要补跑（见 handleSocketOpen）。 */ +let everOpened = false; const MAX_BACKOFF = 30_000; const BASE_BACKOFF = 1_000; @@ -101,7 +107,13 @@ export function dispatchFrame(ev: { if (handleThinkingEvent(ev, r, s)) return; if (handleToolEvent(ev, r, s)) return; if (handleMessageEvent(ev, r, s)) return; + // 主动推送（切片 P0）：气泡恒落 + 非前台补一发通知。此前这一帧**没有消费点** + // ⇒ 主人的「后台思考主动推送一个都没推出来」说的就是它。 + if (handleProactiveEvent(ev, r, s)) return; if (handleAgentEvent(ev, r, s)) return; + // 任务/结果帧（切片 P0 顺带件）：task_done / automation_result / subagent_result + // 后端一直在发，此前全落进未知帧兜底白丢；subagent_result 不落库 ⇒ 错过即永久丢。 + if (handleTaskEvent(ev, r, s)) return; if (handleSessionEvent(ev, r, s)) return; if (handlePersonaEvent(ev, r, s)) return; // 忙时并行帧组在 device 组之前：`error` 除 `chat_busy` 外一律放行，仍由 device.ts @@ -115,6 +127,26 @@ export function dispatchFrame(ev: { s.addDeviceUpdate(`[${ev.type}] ${text}`); } +/** 连接建立后的固定动作（从 `openSocket` 抽出来，好让「重连补齐」这条口径能被纯 + * 逻辑测试直接钉住 —— 项目没有 jsdom，socket 生命周期本身跑不了）。 + * + * 两件事： + * ① 每个（重）连接都重发当前模式，让后端的壳与界面一致（既有行为）； + * ② **重连**（`wasOpenBefore`，即这一页此前打开过）时重跑本页登记的引导读取 + * （`lib/ws/resync.ts`）—— 握手 `history` 回放只补**消息**，会话列表 / 角色目录 / + * 链路图快照这类初始 GET 在断线窗口里的变化没人补，页面会一直停在旧态。 + * 首次连接**不重跑**：页面挂载时的初始加载已经做过一遍。 + * + * `send` 可注入，默认走真实上行通道（socket 未打开时 `sendUp` 返回 false，不抛）。 */ +export function handleSocketOpen( + wasOpenBefore: boolean, + send: (ev: WsUpEvent) => void = sendUp +): void { + const mode = useModeStore.getState().mode; + send({ type: "mode_switch", data: { mode } }); + if (wasOpenBefore) notifyResync(); +} + function scheduleReconnect(): void { if (!shouldRun) return; if (reconnectTimer !== null) return; @@ -159,12 +191,12 @@ function openSocket(): void { s.setWsStatus(attempt === 0 ? "connecting" : "reconnecting"); ws.onopen = () => { + const reconnected = everOpened; + everOpened = true; opened = true; attempt = 0; useAppStore.getState().setWsStatus("connected"); - // Sync the current mode on every (re)connect so the agent's shell matches. - const mode = useModeStore.getState().mode; - sendUp({ type: "mode_switch", data: { mode } }); + handleSocketOpen(reconnected); }; ws.onmessage = (msg) => { diff --git a/frontend/web/src/lib/ws/handlers/proactive.test.ts b/frontend/web/src/lib/ws/handlers/proactive.test.ts new file mode 100644 index 0000000..e2a9f17 --- /dev/null +++ b/frontend/web/src/lib/ws/handlers/proactive.test.ts @@ -0,0 +1,253 @@ +// 主动推送帧（`proactive_message`）的**消费**测试（切片 P0「她主动找你」；design/12 §3）。 +// +// 为什么需要它：后端一直在发这帧，而前端**零消费点** —— 主人的原话是「后台思考的 +// 主动推送一个都没推出来」。本测试锁主人裁定的四条口径： +// ① **气泡恒落**：帧落进消息 store，且与握手回放 / IM 镜像**按落库行 id 去重** +// （同一条绝不渲染两个气泡；刷新后由历史行接续，仍是一个）； +// ② **后台非前台才补通知**：可见且聚焦不弹； +// ③ **拒绝授权 ⇒ 如实降级**：只留气泡 + 一句可见说明，绝不报错崩掉、绝不静默； +// ④ **同一条只提醒一次**（含同一条既走实时帧、又被 IM 镜像回传的情形）。 +// +// 项目没有前端测试框架/jsdom，所以这是零依赖的纯 node 检查：它驱动真实的帧解码器 +// （lib/ws.ts 的 dispatchFrame）与真实的 store，并把浏览器 API（Notification / +// document）用桩装上（handler 只在**调用时**解引用它们，因此模块加载期不碰 DOM）。 +// +// 跑法（用 vite 自带的 esbuild，不新增依赖、不改 package.json；banner 补上 store 的 +// persist 中间件在 node 下探测的 window.localStorage）： +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/lib/ws/handlers/proactive.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-proactive.test.mjs \ +// && node /tmp/cyrene-proactive.test.mjs +// +// `tsc --noEmit`（npm run build 的一部分）会对本文件做类型检查，因此它只从项目内导入、 +// 不 import node:*。 + +import { dispatchFrame } from "../../ws"; +import { useAppStore } from "../../../store/useAppStore"; +import type { ChatMessage } from "../../types"; + +let checks = 0; + +function ok(cond: boolean, what: string): void { + checks += 1; + if (!cond) throw new Error(`FAILED: ${what}`); +} + +function same(got: T, want: T, what: string): void { + ok( + JSON.stringify(got) === JSON.stringify(want), + `${what} — got ${JSON.stringify(got)}, want ${JSON.stringify(want)}` + ); +} + +const MAIN = "web"; + +/** 一条主动消息的线上帧（后端 `hub.SendProactive` 的载荷形状：落库成功时带 + * `id`/`createdAt`，失败时这两个键缺席）。 */ +function proactiveFrame(over: Record = {}): { + type: string; + data: Record; +} { + return { + type: "proactive_message", + data: { + platform: "cyrene-backthink", + content: "记得喝水哦", + contentType: "text", + sender: "Cyrene", + timestamp: "2026-10-05T10:00:00Z", + id: "row-1", + createdAt: 1_760_000_000_000, + ...over, + }, + }; +} + +function msgs(): ChatMessage[] { + return useAppStore.getState().messages; +} + +/** 每个场景前把 store 复位成「新连接、无在途路」。 */ +function reset(activeSessionId = MAIN, messages: ChatMessage[] = []): void { + useAppStore.setState({ + activeSessionId, + messages, + sessionMessages: {}, + deviceUpdates: [], + turnBindings: {}, + sessionBusy: false, + pendingCount: 0, + activeTurns: 0, + queueNotice: null, + }); +} + +/** 装一个 `Notification` 桩（并返回它记下的通知），权限由参数给出。 */ +function installNotification(perm: string): Array<{ title: string; body?: string }> { + const shown: Array<{ title: string; body?: string }> = []; + class StubNotification { + static permission = perm; + constructor(title: string, options?: { body?: string }) { + shown.push({ title, body: options?.body }); + } + } + (globalThis as unknown as { Notification?: unknown }).Notification = StubNotification; + return shown; +} + +/** 装一个 `document` 桩，只提供本 handler 用到的两个成员。 */ +function setDocument(visibility: string, focused: boolean): void { + (globalThis as unknown as { document?: unknown }).document = { + visibilityState: visibility, + hasFocus: () => focused, + }; +} + +// ── ① 气泡恒落：帧落进消息 store（主会话），字段取自落库行 ─────────────────────── +{ + reset(); + installNotification("granted"); + setDocument("visible", true); + dispatchFrame(proactiveFrame()); + same(msgs().length, 1, "主动帧落成一个气泡"); + const m = msgs()[0]; + same(m.id, "row-1", "气泡 id = 落库行 id（去重与刷新接续都靠它）"); + same(m.sessionId, MAIN, "主动消息恒落主会话"); + same(m.role, "assistant", "角色 = assistant（她单方面开口的话）"); + same(m.content, "记得喝水哦", "正文逐字来自帧"); + same(m.createdAt, 1_760_000_000_000, "时间取落库行的时间戳"); + same(useAppStore.getState().deviceUpdates.length, 0, "已消费帧，不落未知帧兜底"); +} + +// ── ① 按 id 去重：重复帧 / 与 IM 镜像（message_persisted）撞车都不新增气泡 ───────── +{ + reset(); + installNotification("granted"); + setDocument("hidden", false); + dispatchFrame(proactiveFrame()); + same(msgs().length, 1, "第一帧落成一个气泡"); + dispatchFrame(proactiveFrame()); + same(msgs().length, 1, "同 id 重复帧不新增（按 id 去重）"); + // IM 镜像路径：同一条消息被 `message_persisted` 再回传一次（同 id、同正文）。 + dispatchFrame({ + type: "message_persisted", + sessionId: MAIN, + data: { + message: { + id: "row-1", + sessionId: MAIN, + role: "assistant", + content: "记得喝水哦", + createdAt: 1_760_000_000_000, + }, + }, + }); + same(msgs().length, 1, "与 IM 镜像同时到达也只保留一个气泡"); + same(useAppStore.getState().deviceUpdates.length, 0, "两个帧都被消费，没有未知帧兜底"); +} + +// ── ② 刷新后仍在：清空 store 后由握手回放（同一个 id）接续，而不是多出一条 ──────── +{ + reset(); + installNotification("granted"); + setDocument("visible", true); + dispatchFrame(proactiveFrame()); + same(msgs().length, 1, "实时帧落成气泡"); + // 模拟刷新：store 清空（消息不持久化，后端才是真源），随后握手 history 回放同一条行。 + reset(); + dispatchFrame({ + type: "history", + sessionId: MAIN, + data: { + messages: [ + { + id: "row-1", + sessionId: MAIN, + role: "assistant", + content: "记得喝水哦", + createdAt: 1_760_000_000_000, + }, + ], + }, + }); + same(msgs().map((m) => m.id), ["row-1"], "刷新后由历史行接续（这正是「气泡恒落」的根）"); +} + +// ── ② 通知分流：前台不弹、非前台才弹 ──────────────────────────────────────────── +{ + reset(); + const shown = installNotification("granted"); + setDocument("visible", true); // 可见且聚焦 = 前台 + dispatchFrame(proactiveFrame({ id: "row-fg", content: "前台这条" })); + same(shown.length, 0, "页面可见且聚焦 ⇒ 不弹通知（气泡已经落了）"); + same(msgs().length, 1, "前台也照样落气泡（气泡与前后台无关）"); + + setDocument("hidden", false); // 非前台 + dispatchFrame(proactiveFrame({ id: "row-bg", content: "后台这条" })); + same(shown.length, 1, "非前台 ⇒ 补一发通知"); + same(shown[0].body, "后台这条", "通知正文 = 主动消息正文"); + same(msgs().length, 2, "两条都落了气泡"); +} + +// ── ② 同一条只打扰一次（重复帧 / 后台重复到达都不再弹） ───────────────────────── +{ + reset(); + const shown = installNotification("granted"); + setDocument("hidden", false); + dispatchFrame(proactiveFrame({ id: "row-once", content: "只说一次" })); + same(shown.length, 1, "第一次非前台 ⇒ 弹一发"); + dispatchFrame(proactiveFrame({ id: "row-once", content: "只说一次" })); + same(shown.length, 1, "同一条再到达 ⇒ 不打扰第二次"); + same(msgs().length, 1, "而气泡仍然只有一个（按 id 去重）"); +} + +// ── ③ 拒绝授权 ⇒ 只留气泡 + 一句可见说明，不报错、不静默 ──────────────────────── +{ + reset(); + const shown = installNotification("denied"); + setDocument("hidden", false); + dispatchFrame(proactiveFrame({ id: "row-denied", content: "通知被拒时这条" })); + same(shown.length, 0, "没有授权 ⇒ 不弹（绝不假装弹过）"); + same(msgs().length, 1, "气泡必须仍然落地"); + const updates = useAppStore.getState().deviceUpdates; + same(updates.length, 1, "留一句可见说明（不是静默丢弃）"); + ok((updates[0] ?? "").includes("通知"), "说明里点明是浏览器通知的问题"); + // 第二条再被拒时不重复说明（状态只需说明一次，不刷屏）。 + dispatchFrame(proactiveFrame({ id: "row-denied-2", content: "第二条" })); + same(useAppStore.getState().deviceUpdates.length, 1, "同一降级状态只说明一次"); + same(msgs().length, 2, "第二条气泡照落"); +} + +// ── ③ 没有通知 API 时同样是「只留气泡」而不是崩掉 ─────────────────────────────── +{ + reset(); + (globalThis as unknown as { Notification?: unknown }).Notification = undefined; + setDocument("hidden", false); + dispatchFrame(proactiveFrame({ id: "row-no-api", content: "没有通知 API" })); + same(msgs().length, 1, "环境没有 Notification 也照样落气泡、不抛错"); +} + +// ── ④ 畸形载荷不静默：落进既有的本端 WS 事件回显 ───────────────────────────────── +{ + reset(); + installNotification("granted"); + setDocument("visible", true); + dispatchFrame({ type: "proactive_message", data: { platform: "cyrene-backthink" } }); + same(msgs().length, 0, "没有正文的载荷不造气泡"); + same(useAppStore.getState().deviceUpdates.length, 1, "但要在回显里留一行（不静默丢弃）"); +} + +// ── ④ 落库失败（后端如实省略 id）：气泡仍落，本端给一个本地 id ──────────────────── +{ + reset(); + installNotification("granted"); + setDocument("visible", true); + dispatchFrame(proactiveFrame({ id: undefined, createdAt: undefined })); + same(msgs().length, 1, "没有落库行 id 时本端仍落气泡"); + ok(msgs()[0].id !== "", "本端给一个非空 id（渲染与 key 都需要它）"); +} + +console.log(`proactive_message logic test: ${checks} checks passed`); diff --git a/frontend/web/src/lib/ws/handlers/proactive.ts b/frontend/web/src/lib/ws/handlers/proactive.ts new file mode 100644 index 0000000..5a1f675 --- /dev/null +++ b/frontend/web/src/lib/ws/handlers/proactive.ts @@ -0,0 +1,198 @@ +import type { ChatMessage } from "../../types"; +import { now, uid } from "../../util"; +import { toRecord } from "../../wsParse"; +import { MAIN_SESSION_ID } from "../../../store/sessionCache"; +import { + createSeenNotices, + decideProactiveNotice, + type NotifyPermission, +} from "../../notify/proactive"; +import { frameText, type FrameEvent, type FramePayload, type StoreState } from "./shared"; + +/** + * 主动推送帧（`proactive_message`，design/12 §3；切片 P0「她主动找你」）。 + * + * 为什么需要它：后端一直在发这帧，前端**零消费点** ⇒ 它落进 `dispatchFrame` 的未知帧 + * 兜底，界面上既没有气泡、也没有通知。主人报的「后台思考的主动推送一个都没推出来」 + * 说的正是这个。本组补上两件事： + * + * ① **气泡恒落**：帧落进消息 store（按 id 去重 —— 与握手回放 / REST 历史拉取按同一个 + * 落库行 id 对齐，因此「同一条实时到了、刷新又拉回来」仍是一个气泡）。气泡是 + * **记录**，与页面在不在前台无关。 + * ② **非前台才补通知**：见 `lib/notify/proactive.ts` 的裁定。前台不弹、同一条只打扰 + * 一次、未授权就如实降级（只留气泡 + 一句可见说明），**绝不静默丢弃**。 + * + * 载荷畸形时也**不静默**：落到既有的「本端 WS 事件回显」可观测面（与未知帧同一去处）。 + * + * 命中返回 true，未命中返回 false 交给下一组。 + */ +export function handleProactiveEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { + if (ev.type !== "proactive_message") return false; + const payload = parseProactive(r); + if (!payload) { + s.addDeviceUpdate(`[proactive_message] 载荷无法解析：${rawText(ev.data)}`); + return true; + } + s.appendOrUpdateMessage(toChatMessage(payload)); + notifyProactive(payload, s); + return true; +} + +/** 一帧主动消息的可用字段（已收窄）。 */ +interface ProactiveMessage { + /** 主会话历史行的权威 id；后端落库失败时**省略**（本端那时只能造一个本地 id）。 */ + id: string; + content: string; + /** 这帧所属的会话：主动消息恒落在主会话（后端就是这么落库的），因此缺省即主会话。 */ + sessionId: string; + platform?: string; + createdAt: number; +} + +/** 收窄一帧 `proactive_message` 的载荷。没有正文即不可用（返回 null）—— 气泡里 + * 没有内容的消息不是消息。 */ +function parseProactive(r: FramePayload): ProactiveMessage | null { + const content = frameText(r?.content); + if (!content) return null; + const id = frameText(r?.id) ?? ""; + const createdAt = + typeof r?.createdAt === "number" && r.createdAt > 0 + ? r.createdAt + : parseTimestamp(r?.timestamp) ?? now(); + return { + id, + content, + sessionId: MAIN_SESSION_ID, + platform: frameText(r?.platform), + createdAt, + }; +} + +/** 把主动消息投影为 store 里的助手气泡。`id` 用落库行 id（与历史行同源，去重靠它）； + * 没有（落库失败）时给一个本端 id，气泡照落 —— 后端也说了它没留下记录，刷新不会 + * 回来，这是如实的一致行为，而不是本端隐瞒。 */ +function toChatMessage(p: ProactiveMessage): ChatMessage { + return { + id: p.id || uid(), + sessionId: p.sessionId, + role: "assistant", + content: p.content, + platform: p.platform, + createdAt: p.createdAt, + }; +} + +/** `timestamp` 是 Go `time.Time` 的 RFC3339 串；解析不出时不编值（返回 undefined， + * 由调用方回落本地时钟）。 */ +function parseTimestamp(v: unknown): number | undefined { + if (typeof v !== "string" || !v) return undefined; + const t = Date.parse(v); + return Number.isNaN(t) ? undefined : t; +} + +/** 本页存活期间的「已提醒」表：同一条消息只打扰一次（含同一条既走实时帧、又被 IM + * 镜像回主会话这种同时到达的情形）。气泡不参与去重 —— 那是消息 store 按 id 的事。 */ +const seenNotices = createSeenNotices(200); + +/** 最近一次已经说明过的降级状态：通知没开是**状态**，不是每次推送都要报告一遍的事故。 */ +let degradeNote: NotifyPermission | null = null; + +/** 按 `lib/notify/proactive.ts` 的裁定决定要不要补一发系统通知，并把「同一条只打扰 + * 一次」记进本页的去重表。 */ +function notifyProactive(p: ProactiveMessage, s: StoreState): void { + const key = p.id || `${p.createdAt}|${p.content}`; + const permission = notifyPermission(); + const decision = decideProactiveNotice({ + seen: seenNotices.has(key), + visibility: documentVisibility(), + focused: documentFocused(), + permission, + }); + // 无论结论如何都记下「这条已经露过面」：前台看过、或提醒过，都不该被第二次打扰。 + seenNotices.add(key); + if (decision === "not-granted") { + noteDegrade(s, permission); + return; + } + if (decision !== "show") return; + if (!showSystemNotification(p.content, key)) { + // 构造通知本身失败（浏览器策略/权限收紧等）也算降级：留一句可见说明， + // 绝不让它冒泡进帧处理把气泡也带崩。 + noteDegrade(s, notifyPermission()); + } +} + +/** 通知没开时留一句**可见说明**（同一状态只说一次，不刷屏）：让主人知道「不是她没说， + * 是浏览器通知没开」，而不是以为主动消息根本没来。 */ +function noteDegrade(s: StoreState, permission: NotifyPermission): void { + if (degradeNote === permission) return; + degradeNote = permission; + s.addDeviceUpdate( + permission === "denied" + ? "浏览器通知未授权（已拒绝）：主动消息只出现在气泡里，可在本站的站点设置里允许通知。" + : "浏览器通知尚未授权：主动消息只出现在气泡里，可在本站允许通知。" + ); +} + +/** `Notification` 构造器（不存在时 null）。**调用时**才解引用全局，因此纯 node + * 测试可以安全导入本模块（模块加载期不碰 DOM）。 */ +type NotificationCtor = (new (title: string, options?: { body?: string; tag?: string }) => unknown) & { + permission?: string; +}; + +function notificationCtor(): NotificationCtor | null { + const g = globalThis as unknown as { Notification?: NotificationCtor }; + return typeof g.Notification === "function" ? g.Notification : null; +} + +/** 当前通知权限。运行时没有该 API ⇒ 归到 `"denied"`（拿不到通知就是没授权， + * 绝不假装已授权）。 */ +function notifyPermission(): NotifyPermission { + const ctor = notificationCtor(); + if (!ctor) return "denied"; + if (ctor.permission === "granted") return "granted"; + if (ctor.permission === "denied") return "denied"; + return "default"; +} + +/** `document.visibilityState`；取不到时按 `"visible"`（= 不去打扰）。 */ +function documentVisibility(): string { + const g = globalThis as unknown as { document?: { visibilityState?: string } }; + const v = g.document?.visibilityState; + return typeof v === "string" ? v : "visible"; +} + +/** `document.hasFocus()`；取不到或抛错时按「已聚焦」处理（= 不去打扰）。 */ +function documentFocused(): boolean { + const g = globalThis as unknown as { document?: { hasFocus?: () => boolean } }; + const hasFocus = g.document?.hasFocus; + if (typeof hasFocus !== "function") return true; + try { + return hasFocus(); + } catch { + return true; + } +} + +/** 弹一发系统通知；失败（含构造抛错）返回 false，由调用方降级为「只留气泡 + + * 一句可见说明」。 */ +function showSystemNotification(body: string, tag: string): boolean { + const ctor = notificationCtor(); + if (!ctor) return false; + try { + new ctor("昔涟", { body, tag }); + return true; + } catch { + return false; + } +} + +/** 把无法解析的载荷转成一行可读文本（本端事件回显用）。 */ +function rawText(v: unknown): string { + if (typeof v === "string") return v; + try { + return JSON.stringify(v) ?? String(v); + } catch { + return String(v); + } +} diff --git a/frontend/web/src/lib/ws/handlers/task.test.ts b/frontend/web/src/lib/ws/handlers/task.test.ts new file mode 100644 index 0000000..4a9706c --- /dev/null +++ b/frontend/web/src/lib/ws/handlers/task.test.ts @@ -0,0 +1,104 @@ +// 任务 / 结果帧的**消费**测试（切片 P0 的顺带件）：`task_done` / `automation_result` / +// `subagent_result`。 +// +// 为什么需要它：这三个帧后端一直在发（`ws_bridge.go` 的 `eventBridge` 把它们成形为 +// 下行帧），前端却**零消费点** ⇒ 全部落进未知帧兜底，等于白丢。其中 `subagent_result` +// 更严重：它**不落库**（父会话转录里没有新消息、重连回放不补发）⇒ 错过即永久丢。 +// 本测试锁「至少不静默」： +// ① 三条各自落进**本端 WS 事件回显**（`deviceUpdates`），而不是被无声丢掉； +// ② 带 id 的结论顺手写进**任务日志**（`taskLogs`，子代理页读它）； +// ③ 载荷残缺时回显那一行仍然发（「这一批跑完了」这件事不能因为字段缺而消失）。 +// +// 跑法： +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/lib/ws/handlers/task.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-task.test.mjs \ +// && node /tmp/cyrene-task.test.mjs + +import { dispatchFrame } from "../../ws"; +import { useAppStore } from "../../../store/useAppStore"; + +let checks = 0; + +function ok(cond: boolean, what: string): void { + checks += 1; + if (!cond) throw new Error(`FAILED: ${what}`); +} + +function reset(): void { + useAppStore.setState({ deviceUpdates: [], taskLogs: {} }); +} + +function updates(): string[] { + return useAppStore.getState().deviceUpdates; +} + +// ── task_done：回显一行 + 任务日志按 taskId 收一行 ────────────────────────────── +{ + reset(); + dispatchFrame({ + type: "task_done", + data: { id: "task-9", kind: "shell", status: "failed", exitCode: 1, error: "boom" }, + }); + ok(updates().length === 1, "task_done 落进本端事件回显"); + ok(updates()[0].includes("task-9") && updates()[0].includes("failed"), "回显点名任务与终态"); + ok(updates()[0].includes("boom"), "失败原因也在回显里（不吞错误）"); + const log = useAppStore.getState().taskLogs["task-9"] ?? ""; + ok(log.includes("task_done"), "任务日志按 taskId 收到一行"); +} + +// ── automation_result：回显一行（含规则、动作与成败） ─────────────────────────── +{ + reset(); + dispatchFrame({ + type: "automation_result", + data: { kind: "schedule", rule_id: "morning-brief", action: "tts", output: "ok" }, + }); + ok(updates().length === 1, "automation_result 落进本端事件回显"); + ok(updates()[0].includes("morning-brief") && updates()[0].includes("tts"), "回显点名规则与动作"); + ok(updates()[0].includes("完成"), "成功时说明是完成"); + + reset(); + dispatchFrame({ + type: "automation_result", + data: { kind: "reminder", rule_id: "r-1", action: "notify", error: "channel offline" }, + }); + ok(updates()[0].includes("失败") && updates()[0].includes("channel offline"), "失败时如实写明原因"); +} + +// ── subagent_result：回显一批的摘要 + 逐代理结论进任务日志 ────────────────────── +{ + reset(); + dispatchFrame({ + type: "subagent_result", + sessionId: "web", + data: { + sessionId: "web", + status: "done", + summary: "两路都跑完了", + agents: [ + { id: "a-1", goal: "查天气", status: "done", summary: "晴", text: "明天晴，25 度" }, + { id: "a-2", goal: "查日程", status: "failed", summary: "读不到日历" }, + ], + }, + }); + ok(updates().length === 1, "subagent_result 落进本端事件回显（不落库的帧更不能静默丢）"); + ok(updates()[0].includes("2 个") && updates()[0].includes("两路都跑完了"), "回显写明批量与汇总"); + const a1 = useAppStore.getState().taskLogs["a-1"] ?? ""; + const a2 = useAppStore.getState().taskLogs["a-2"] ?? ""; + ok(a1.includes("明天晴，25 度"), "子代理的结论正文进了它自己的任务日志"); + ok(a2.includes("读不到日历"), "没有 text 时退回 summary（结论不能丢在载荷形状上）"); +} + +// ── 载荷残缺：回显那一行仍然发（「跑完了」这件事不能因为字段缺而消失） ────────── +{ + reset(); + dispatchFrame({ type: "subagent_result", data: { status: "done" } }); + ok(updates().length === 1, "没有 agents 数组时仍然留一行回显"); + ok(useAppStore.getState().taskLogs[""] === undefined, "不写空 key 的任务日志"); +} + +console.log(`task result frames logic test: ${checks} checks passed`); diff --git a/frontend/web/src/lib/ws/handlers/task.ts b/frontend/web/src/lib/ws/handlers/task.ts new file mode 100644 index 0000000..d639295 --- /dev/null +++ b/frontend/web/src/lib/ws/handlers/task.ts @@ -0,0 +1,96 @@ +import { toRecord } from "../../wsParse"; +import { frameText, type FrameEvent, type FramePayload, type StoreState } from "./shared"; + +/** + * 任务 / 结果帧处理组（切片 P0 的顺带件）：`task_done`、`automation_result`、 + * `subagent_result`。 + * + * 为什么需要它：这三个帧后端**一直在发**（`internal/services/gateway/ws_bridge.go` + * 的 `eventBridge` 把它们从 bus 主题成形并投递给 owner/admin 连接），前端却零消费点 + * ⇒ 全部落进未知帧兜底，等于白丢。其中 `subagent_result` 更严重：它**不落库** + * （design/05 §9 的诚实边界：父会话转录里没有新消息、重连回放不补发）⇒ 错过即永久丢。 + * + * 本组的取舍是「**至少不静默**」：三条都进既有的本端 WS 事件回显（`deviceUpdates`）， + * 并把带 id 的结论顺手写进任务日志（`taskLogs`，子代理页读它）。它们不进消息 store + * —— 后端没把它们写成会话消息，前端也不该凭空造气泡。 + * + * 命中返回 true，未命中返回 false 交给下一组。 + */ +export function handleTaskEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { + switch (ev.type) { + case "task_done": + return handleTaskDone(r, s); + case "automation_result": + return handleAutomationResult(r, s); + case "subagent_result": + return handleSubagentResult(r, s); + default: + return false; + } +} + +/** `task_done`：后台任务到达终态（载荷 = 后端 `task.DoneEvent`）。任务日志按 + * `taskId` 收一行，回显里再留一行摘要 —— 与 `task_output` 的日志口径同源。 */ +function handleTaskDone(r: FramePayload, s: StoreState): boolean { + const id = frameText(r?.id) ?? ""; + const status = frameText(r?.status) ?? "unknown"; + const err = frameText(r?.error); + const line = `[task_done] 任务 ${id || "(未知)"} 结束：${status}${err ? `（${err}）` : ""}`; + if (id) s.appendTaskLog(id, `${line}\n`); + s.addDeviceUpdate(line); + return true; +} + +/** `automation_result`：一次自动化/提醒规则跑完（载荷 = 后端 `automations.Result`）。 */ +function handleAutomationResult(r: FramePayload, s: StoreState): boolean { + const kind = frameText(r?.kind) ?? "automation"; + const rule = frameText(r?.rule_id) ?? frameText(r?.ruleId) ?? "(未知规则)"; + const action = frameText(r?.action); + const err = frameText(r?.error); + s.addDeviceUpdate( + `[automation_result] ${kind} ${rule}${action ? ` → ${action}` : ""}：${err ? `失败（${err}）` : "完成"}` + ); + return true; +} + +/** 一条子代理结论（载荷 `agents[]` 的可用字段）。 */ +interface SubagentOutcome { + id: string; + status: string; + summary: string; + text: string; +} + +/** `subagent_result`：一个后台子代理批次的最终结果（**不落库**，错过即永久丢）。 */ +function handleSubagentResult(r: FramePayload, s: StoreState): boolean { + const sessionId = frameText(r?.sessionId); + const summary = frameText(r?.summary) ?? ""; + const agents = parseSubagentAgents(r?.agents); + s.addDeviceUpdate( + `[subagent_result] ${sessionId ? `${sessionId} ` : ""}子代理批次完成` + + `${agents.length ? `（${agents.length} 个）` : ""}${summary ? `：${summary}` : ""}` + ); + for (const a of agents) { + const body = a.text || a.summary; + if (a.id && body) s.appendTaskLog(a.id, `[subagent_result] ${a.status}：${body}\n`); + } + return true; +} + +/** 收窄载荷里的 `agents[]`；不是数组或没有可用字段时返回空数组（回显那行仍然发， + * 因此「这一批跑完了」这件事不会因为载荷残缺而被静默丢掉）。 */ +function parseSubagentAgents(v: unknown): SubagentOutcome[] { + if (!Array.isArray(v)) return []; + const out: SubagentOutcome[] = []; + for (const item of v) { + const rec = toRecord(item); + if (!rec) continue; + out.push({ + id: frameText(rec.id) ?? "", + status: frameText(rec.status) ?? "done", + summary: frameText(rec.summary) ?? "", + text: frameText(rec.text) ?? "", + }); + } + return out; +} diff --git a/frontend/web/src/lib/ws/resync.test.ts b/frontend/web/src/lib/ws/resync.test.ts new file mode 100644 index 0000000..376d5bc --- /dev/null +++ b/frontend/web/src/lib/ws/resync.test.ts @@ -0,0 +1,110 @@ +// 「重连后重跑当前页引导读取」的逻辑测试（切片 P0 的顺带件）。 +// +// 为什么需要它：`ws.ts` 的 `onopen` 此前只重发 `mode_switch`，于是断线窗口里 +// **非消息**的实时状态永久缺失 —— 握手 `history` 回放只补消息，会话列表 / 角色目录 / +// 「最近一轮」链路图快照这类**引导读取**在断线期间的变化没人补，页面会一直停在旧态。 +// 本测试锁三件事： +// ① **重连**（曾打开过）会重跑登记项；**首次连接不会**（页面挂载时已经加载过）； +// ② `mode_switch` 这条既有行为在两种情况都保留； +// ③ 单个登记项抛错不影响其它项，也不冒泡（页面某处补拉失败不该拖垮实时通道）。 +// +// 项目没有前端测试框架/jsdom，所以这是零依赖的纯 node 检查：它驱动真实的登记表 +// （lib/ws/resync.ts）与真实的连接建立动作（lib/ws.ts 的 handleSocketOpen）。 +// +// 跑法： +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/lib/ws/resync.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-resync.test.mjs \ +// && node /tmp/cyrene-resync.test.mjs +// +// `tsc --noEmit` 会对本文件做类型检查，因此它只从项目内导入、不 import node:*。 + +import { handleSocketOpen } from "../ws"; +import { notifyResync, registerResync, resyncHandlerCount } from "./resync"; +import type { WsUpEvent } from "../types"; + +let checks = 0; + +function ok(cond: boolean, what: string): void { + checks += 1; + if (!cond) throw new Error(`FAILED: ${what}`); +} + +/** 收集一次重连动作发出的所有上行帧。 */ +function recorder(): { events: WsUpEvent[]; send: (ev: WsUpEvent) => void } { + const events: WsUpEvent[] = []; + return { events, send: (ev) => events.push(ev) }; +} + +// ── ① 首次连接：不重跑引导读取（页面挂载时的初始加载已经做过一遍），但模式照发 ──── +{ + let ran = 0; + const unregister = registerResync(() => { + ran += 1; + }); + const { events, send } = recorder(); + handleSocketOpen(false, send); + ok(ran === 0, "首次连接不重跑引导读取"); + ok(events.length === 1 && events[0].type === "mode_switch", "首次连接仍重发当前模式"); + unregister(); + notifyResync(); + ok(ran === 0, "退订之后不再被调用（组件卸载不留悬挂回调）"); +} + +// ── ② 重连：重跑登记项，且模式照发 ───────────────────────────────────────────── +{ + let ran = 0; + const unregister = registerResync(() => { + ran += 1; + }); + const { events, send } = recorder(); + handleSocketOpen(true, send); + ok(ran === 1, "重连重跑一次本页的引导读取"); + ok(events.length === 1 && events[0].type === "mode_switch", "重连仍重发当前模式"); + handleSocketOpen(true, send); + ok(ran === 2, "每次重连都重跑（不是只补一次）"); + unregister(); +} + +// ── ③ 一个登记项抛错不影响其它项，也不冒泡 ──────────────────────────────────── +{ + let ok2 = 0; + const off1 = registerResync(() => { + throw new Error("这一页的补拉失败了"); + }); + const off2 = registerResync(() => { + ok2 += 1; + }); + let threw = false; + try { + handleSocketOpen(true, () => {}); + } catch { + threw = true; + } + ok(!threw, "单个登记项抛错绝不冒泡进 socket 生命周期"); + ok(ok2 === 1, "其它登记项照常被重跑"); + off1(); + off2(); + ok(resyncHandlerCount() === 0, "全部退订后登记表为空"); +} + +// ── ④ 登记表幂等：同一个函数登记两次只跑一次（Set 语义） ──────────────────────── +{ + let ran = 0; + const fn = (): void => { + ran += 1; + }; + const off1 = registerResync(fn); + const off2 = registerResync(fn); + ok(resyncHandlerCount() === 1, "同一个函数只登记一次"); + notifyResync(); + ok(ran === 1, "只重跑一次"); + off1(); + off2(); + ok(resyncHandlerCount() === 0, "两次退订都不会残留"); +} + +console.log(`ws reconnect resync logic test: ${checks} checks passed`); diff --git a/frontend/web/src/lib/ws/resync.ts b/frontend/web/src/lib/ws/resync.ts new file mode 100644 index 0000000..f2da578 --- /dev/null +++ b/frontend/web/src/lib/ws/resync.ts @@ -0,0 +1,44 @@ +/** + * 「重连后重跑当前页引导读取」的通用登记表（切片 P0 的顺带件）。 + * + * 为什么需要它：`ws.ts` 的 `onopen` 此前只重发一帧 `mode_switch`，于是**断线窗口**里 + * 发生的变化没人补 —— 握手 `history` 回放只补**消息**，而会话列表、角色目录、 + * 「最近一轮」链路图快照这类**引导读取**（页面的初始 GET）只有在重新挂载时才会跑， + * 页面会一直停在旧态，直到手动刷新。修掉它只需要一个通用钩子：各页把「本页的引导 + * 读取」登记一次，重连时统一重跑。 + * + * 本模块刻意**不 import react**（hook 在 `lib/useWsResync.ts`）：登记表本身是纯逻辑， + * 因此能在无 jsdom 的纯 node 下钉住「重连真的会重跑、首次连接不会白跑一遍」。 + */ + +/** 一件重连后要重跑的引导读取。 */ +type ResyncFn = () => void; + +const handlers = new Set(); + +/** 登记一件重连后要重跑的引导读取，返回退订函数（组件卸载时调用）。 */ +export function registerResync(fn: ResyncFn): () => void { + handlers.add(fn); + return () => { + handlers.delete(fn); + }; +} + +/** 重跑全部登记项（**只在重连后**调用，见 `ws.ts#handleSocketOpen`）。 + * + * 单个登记项抛错绝不影响其它项，也绝不冒泡进 socket 生命周期 —— 页面里某一处补拉 + * 失败不该让整条实时通道跟着出问题（它自己的加载/错误态照旧各自负责）。 */ +export function notifyResync(): void { + for (const fn of Array.from(handlers)) { + try { + fn(); + } catch { + /* 一处失败不影响其它页面的重跑 */ + } + } +} + +/** 当前登记的项数（供测试与排查「到底有没有人登记」）。 */ +export function resyncHandlerCount(): number { + return handlers.size; +} diff --git a/internal/services/backthink/mocks_test.go b/internal/services/backthink/mocks_test.go index 42dfa27..0ae4a03 100644 --- a/internal/services/backthink/mocks_test.go +++ b/internal/services/backthink/mocks_test.go @@ -72,19 +72,21 @@ func (p *mockPusher) count() int { p.mu.Lock(); defer p.mu.Unlock(); return p.ca // mockActive 实现 ActiveClient。 type mockActive struct { - mu sync.Mutex - n int - err error - calls int - last adapter.UnifiedMessage + mu sync.Mutex + n int + persisted bool + err error + calls int + last adapter.UnifiedMessage } -func (m *mockActive) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) { +func (m *mockActive) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (ProactiveDelivery, error) { m.mu.Lock() m.calls++ m.last = msg + n, persisted, err := m.n, m.persisted, m.err m.mu.Unlock() - return m.n, m.err + return ProactiveDelivery{Connections: n, Persisted: persisted}, err } // mockIMAdapter 实现 IMAdapter。 diff --git a/internal/services/backthink/push.go b/internal/services/backthink/push.go index 3852dea..1cbf43c 100644 --- a/internal/services/backthink/push.go +++ b/internal/services/backthink/push.go @@ -69,13 +69,35 @@ func (p PushResult) String() string { return fmt.Sprintf("pushed=%v route=%s bypass=%v detail=%q", p.Pushed, p.Route, p.Bypassed, p.Detail) } +// ProactiveDelivery 是一次活跃客户端投递的结果（切片 P0「她主动找你」）。 +// 它把**两件本就独立**的事分开报告：帧送到了几条连接，以及这条消息有没有 +// 成为主会话里的一条真实消息。 +// +// 为什么要分开：修复前活跃客户端路由只广播 WS 帧、不落库，于是面板上的 +// 「delivered to 2 active client(s)」被读成「已推送」，而刷新后什么都没剩 +// —— 那正是主人报的「后台思考的主动推送一个都没有推出来」。落库与投递各自 +// 成败，记录里就必须各自可见。 +type ProactiveDelivery struct { + // Connections 是接受该帧的在线 owner/admin 连接数。0 意味着此刻没有 + // 在线目标；它**不代表消息丢了**——消息可能已经落库（见 Persisted）。 + Connections int + // Persisted 报告该消息是否已成为主会话历史（`history.MainSessionID`）里的 + // 一条真实消息，因而刷新/重载后仍在。 + Persisted bool +} + // ActiveClient 把主动消息投递给所有当前活跃的 Web/原生 -// 客户端，并报告有多少个收到。它是首选的推送目标。 +// 客户端，并报告投递与落库两件事。它是首选的推送目标。 // 生产中由 gateway 的 WebSocket hub 实现；测试注入 mock。 type ActiveClient interface { // SendProactive 把 msg 推给每个活跃客户端，并返回 - // 接受它的客户端数量。 - SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) + // ProactiveDelivery。 + // + // 实现契约（切片 P0）：**先落库、再发帧**，且发帧的载荷带上落库行的 id —— + // 客户端据此把实时气泡对齐到历史行（按 id 去重、刷新后仍在）。落库失败时 + // 仍要发出帧（实时气泡不能丢），但必须把错误返回且 Persisted=false， + // 让调用方如实记录「发出去了、但没留下记录」。 + SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (ProactiveDelivery, error) } // IMAdapter 是 IM 桥降级：它把主动消息经 IM @@ -246,19 +268,43 @@ func (p *pusher) exportAsync(msg adapter.UnifiedMessage) { }() } -// pushActive 只投递给活跃 Web/原生客户端。 +// pushActive 只投递给活跃 Web/原生客户端。它按 ProactiveDelivery 如实报告 +// 「送到了几条连接」与「有没有落进主会话」（切片 P0）：投递数为 0 时消息仍可能 +// 已经是一条真实记录，两者绝不互相冒充。 func (p *pusher) pushActive(ctx context.Context, msg adapter.UnifiedMessage) PushResult { - if p.active != nil { - n, err := p.active.SendProactive(ctx, msg) - if err != nil { - p.log.Warn("backthink active client push failed", "error", err) - } else if n > 0 { - return PushResult{Pushed: true, Route: RouteActiveClient, Detail: fmt.Sprintf("delivered to %d active client(s)", n)} - } + if p.active == nil { + return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client"} + } + d, err := p.active.SendProactive(ctx, msg) + if err != nil { + p.log.Warn("backthink active client push failed", "error", err) + } + if d.Connections > 0 { + return PushResult{Pushed: true, Route: RouteActiveClient, Detail: activeDeliveryDetail(d)} + } + if d.Persisted { + // 此刻没有在线客户端，但这条话已经成为主会话里的一条真实记录：打开/ + // 刷新页面就能看到，只是没有实时帧可发。如实分别报告这两件事，面板才不会 + // 把「没推出去」与「没留下」混为一谈。 + return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client, recorded in main session"} + } + if err != nil { + return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client", Err: err.Error()} } return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client"} } +// activeDeliveryDetail 是活跃客户端路由的 Detail 措辞：投递数与「有没有落进 +// 主会话」两件事实都在里面，且**绝不**只说「已推送」（那正是被误读成「刷新还在」 +// 的原话）。 +func activeDeliveryDetail(d ProactiveDelivery) string { + state := "not persisted to main session" + if d.Persisted { + state = "persisted to main session" + } + return fmt.Sprintf("delivered to %d active client(s), %s", d.Connections, state) +} + // pushIM 只投递给 IM 桥。 func (p *pusher) pushIM(ctx context.Context, msg adapter.UnifiedMessage) PushResult { if p.im == nil { diff --git a/internal/services/backthink/push_active_delivery_test.go b/internal/services/backthink/push_active_delivery_test.go new file mode 100644 index 0000000..9f36286 --- /dev/null +++ b/internal/services/backthink/push_active_delivery_test.go @@ -0,0 +1,117 @@ +package backthink + +import ( + "context" + "errors" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// 本文件锁定切片 P0「她主动找你」的**活跃客户端路由口径**： +// +// - PushResult.Detail 必须把「送到了几条连接」与「有没有落进主会话」**分开** +// 报告 —— 修复前那句「delivered to 2 active client(s)」被读者当成「已推送 +// （刷新也在）」，而当时一行都没落库； +// - 一次推送**至多让一个 sink 成功**（IM 桥与活跃客户端互斥），这正是 +// 「主会话里不会出现双份」的结构性依据：两条路由各自落库一次，而一次推送 +// 只走其中一条。 + +// TestPusher_ActiveDetailReportsPersistence 验证 Detail 分别报告投递数与落库。 +func TestPusher_ActiveDetailReportsPersistence(t *testing.T) { + t.Parallel() + tests := []struct { + name string + active *mockActive + wantPushed bool + wantIn string + wantNotIn string + }{ + { + name: "投递成功且已落库", + active: &mockActive{n: 2, persisted: true}, + wantPushed: true, + wantIn: "persisted to main session", + wantNotIn: "not persisted", + }, + { + name: "投递成功但未落库（故障可见）", + active: &mockActive{n: 2, persisted: false}, + wantPushed: true, + wantIn: "not persisted to main session", + }, + { + name: "无在线客户端但已落库", + active: &mockActive{n: 0, persisted: true}, + wantPushed: false, + wantIn: "recorded in main session", + }, + } + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + p := newPusher(tc.active, nil, &mockGate{allow: true}, discardLog()) + res := p.Push(context.Background(), PushRequest{ + ChannelHint: channelWeb, + Message: adapter.UnifiedMessage{Content: "记得喝水哦"}, + }) + if res.Pushed != tc.wantPushed { + t.Fatalf("Pushed = %v, want %v (res=%+v)", res.Pushed, tc.wantPushed, res) + } + if !strings.Contains(res.Detail, tc.wantIn) { + t.Fatalf("Detail = %q, want it to contain %q", res.Detail, tc.wantIn) + } + if tc.wantNotIn != "" && strings.Contains(res.Detail, tc.wantNotIn) { + t.Fatalf("Detail = %q, must not contain %q", res.Detail, tc.wantNotIn) + } + }) + } +} + +// TestPusher_OnePushSucceedsOnAtMostOneSink 验证一次推送**至多让一个 sink 成功** +// —— 互斥是「主会话里不会双份」的结构性依据。IM 失败后的活跃客户端降级是同一 +// 次推送内的兜底：失败的 IM 什么都没写，因此整体仍只有一条记录。 +func TestPusher_OnePushSucceedsOnAtMostOneSink(t *testing.T) { + t.Parallel() + tests := []struct { + name string + hint string + imErr error + activeN int + wantIMCalls int + wantActiveCall int + }{ + {name: "web 提示只打活跃客户端", hint: channelWeb, activeN: 1, wantIMCalls: 0, wantActiveCall: 1}, + {name: "auto 提示 IM 可用时只打 IM", hint: channelAuto, activeN: 2, wantIMCalls: 1, wantActiveCall: 0}, + {name: "IM 失败时降级到活跃客户端", hint: channelAuto, imErr: errors.New("no owner IM channel"), activeN: 1, wantIMCalls: 1, wantActiveCall: 1}, + } + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + active := &mockActive{n: tc.activeN, persisted: true} + im := &mockIMAdapter{err: tc.imErr} + p := newPusher(active, im, &mockGate{allow: true}, discardLog()) + + res := p.Push(context.Background(), PushRequest{ + ChannelHint: tc.hint, + Message: adapter.UnifiedMessage{Content: "记得喝水哦"}, + }) + if im.count() != tc.wantIMCalls || active.calls != tc.wantActiveCall { + t.Fatalf("sink calls: im=%d (want %d) active=%d (want %d)", + im.count(), tc.wantIMCalls, active.calls, tc.wantActiveCall) + } + // 成功的 sink 至多一个：能落库的路径因此也至多一条。 + imOK := tc.imErr == nil && tc.wantIMCalls == 1 + activeOK := tc.activeN > 0 && tc.wantActiveCall == 1 + if imOK && activeOK { + t.Fatalf("both sinks succeeded in one push (res=%+v)", res) + } + if !imOK && !activeOK && res.Pushed { + t.Fatalf("Pushed = true without a successful sink (res=%+v)", res) + } + }) + } +} diff --git a/internal/services/gateway/proactive.go b/internal/services/gateway/proactive.go index 47484f9..b406ac8 100644 --- a/internal/services/gateway/proactive.go +++ b/internal/services/gateway/proactive.go @@ -8,6 +8,8 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" ) // wsProactiveMessage 是自主主动推送的 WS 事件类型 @@ -27,21 +29,46 @@ type proactivePayload struct { Sender string `json:"sender,omitempty"` Timestamp time.Time `json:"timestamp,omitempty"` MessageID string `json:"messageId,omitempty"` + // ID 是这条主动消息在**主会话历史**里的权威行 id（切片 P0「气泡恒落」）。 + // 客户端据此把实时气泡对齐到落库行：与握手回放 / REST 历史拉取按 id 去重 + // （同一条绝不渲染两个气泡），刷新后仍在。落库失败时该键**省略** —— + // 客户端因此知道这条只有实时帧、没有历史行，而不是拿到一个悬空的 id。 + ID string `json:"id,omitempty"` + // CreatedAt 是落库行的时间戳（Unix 毫秒），与 ID 同时出现/同时省略。它是 + // 权威的排序时间（与 `timestamp` 是同一次推送的两个视角：前者是历史行的 + // 时间，后者是这条统一消息自身的时间）。 + CreatedAt int64 `json:"createdAt,omitempty"` } -// SendProactive 把 msg 投递给每个存活的 owner/admin 连接，并返回 -// 有多少个接受了它。计数为零意味着没有 owner/admin 客户端在线， -// 因此调用方回退到 IM bridge。只有 owner/admin 连接会被作为 -// 目标（红线：主动推送仅 owner）；viewer 或未认证连接 -// 绝不被触达。它绝不阻塞：发送缓冲满时对该 -// 慢客户端丢弃该帧，而不是卡住后台思考推送。 -func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) { +// SendProactive 把 msg 投递给每个存活的 owner/admin 连接，并返回投递与落库 +// 两件事的结果（切片 P0，`backthink.ProactiveDelivery`）。 +// +// 顺序是刻意的 —— **先落库、再发帧**：主动消息必须成为主会话 +// （`history.MainSessionID`）里的一条真实消息（assistant 角色），否则刷新/ +// 重载后什么都补不回来（这正是「气泡恒落」的根）；而帧要带上落库行的 id， +// 客户端才能把实时气泡对齐到历史行。 +// +// 落库失败**不静默**：记一条带来源的 WARN，并把错误返回给调用方，同时 +// Persisted=false —— 帧照发（实时气泡不能丢），但记录里绝不说「已推送」。 +// +// 计数为零意味着没有 owner/admin 客户端在线；消息此时仍会作为一条真实记录 +// 留在主会话里（下次打开就能看到），调用方据此回退到 IM bridge 把它**投出去**。 +// 只有 owner/admin 连接会被作为目标（红线：主动推送仅 owner）；viewer 或未认证 +// 连接绝不被触达。它绝不阻塞：发送缓冲满时对该慢客户端丢弃该帧， +// 而不是卡住后台思考推送。 +func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (backthink.ProactiveDelivery, error) { if err := ctx.Err(); err != nil { - return 0, err + return backthink.ProactiveDelivery{}, err } - b, err := json.Marshal(wsEvent{Type: wsProactiveMessage, Data: proactivePayloadFrom(msg)}) + persisted, persistErr := h.persistProactiveToMain(ctx, msg) + if persistErr != nil { + h.log.Warn("persist proactive message to main session failed", + "err", persistErr, "platform", msg.Platform) + } + b, err := json.Marshal(wsEvent{Type: wsProactiveMessage, Data: proactivePayloadFrom(msg, persisted)}) if err != nil { - return 0, fmt.Errorf("gateway: marshal proactive: %w", err) + return backthink.ProactiveDelivery{Persisted: persisted.ID != ""}, + fmt.Errorf("gateway: marshal proactive: %w", err) } h.mu.RLock() defer h.mu.RUnlock() @@ -56,17 +83,19 @@ func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (in default: } } - return n, nil + return backthink.ProactiveDelivery{Connections: n, Persisted: persisted.ID != ""}, persistErr } // Server.SendProactive 通过委托给 hub 满足后台思考的 ActiveClient // 接缝，因此组合根能把网关自身交给汇聚点设置器。 -func (s *Server) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) { +func (s *Server) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (backthink.ProactiveDelivery, error) { return s.hub.SendProactive(ctx, msg) } // proactivePayloadFrom 把 UnifiedMessage 投影为消毒后的线上载荷。 -func proactivePayloadFrom(msg adapter.UnifiedMessage) proactivePayload { +// persisted 是刚落库的主会话消息（未接线历史存储或落库失败时为零值）， +// 它的 id/时间戳随之进入载荷，使客户端能把实时气泡对齐到历史行。 +func proactivePayloadFrom(msg adapter.UnifiedMessage, persisted history.Message) proactivePayload { return proactivePayload{ Platform: msg.Platform, ChannelID: msg.ChannelID, @@ -76,6 +105,8 @@ func proactivePayloadFrom(msg adapter.UnifiedMessage) proactivePayload { Sender: msg.Sender.DisplayName, Timestamp: msg.Timestamp, MessageID: msg.MessageID, + ID: persisted.ID, + CreatedAt: persisted.CreatedAt, } } diff --git a/internal/services/gateway/proactive_persist.go b/internal/services/gateway/proactive_persist.go new file mode 100644 index 0000000..f0c3b0d --- /dev/null +++ b/internal/services/gateway/proactive_persist.go @@ -0,0 +1,51 @@ +package gateway + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// 本文件承载主动推送的**落库**那一半（切片 P0「她主动找你」）。它从 +// proactive.go 拆出来，好让「帧长什么样」与「记录怎么写」各自成篇 +// （AGENTS §10 模块按职责拆源文件）。 +// +// 为什么必须有这一半：修复前活跃客户端路由只把 `proactive_message` 帧 +// 广播给在线连接（fire-and-forget），**一行都不写** ⇒ 刷新/重载后什么都补不回来， +// 界面既不显示气泡也补不出历史。主人的原话是「后台思考的主动推送一个都没推出来」。 +// 主动消息是**她真实说过的一句话**，因此它必须与 IM 镜像路径走同一条去向： +// 主会话历史里的一条 assistant 消息。 + +// persistProactiveToMain 把一条主动消息写进主会话（`history.MainSessionID`）， +// 并返回落库后的完整消息（含权威 id/时间戳），供帧载荷携带。 +// +// 三条口径： +// - **角色恒为 assistant**：这是她单方面开口说的话，不是主人的输入 —— 塞成 +// user 角色会污染「谁说了什么」，也会在历史里冒充主人发言（红线）。 +// - **正文用调用方已经净化过的文本**（backthink 的 stripInternalMarkers 出口）， +// 系统元数据（决策标记、渠道选择）绝不进正文，也绝不新增任何隐藏字段 —— +// 注入面因此与「气泡里看到的那句话」逐字相同。 +// - **turn_id 留空**：主动推送不属于任何一路回合（她单方面开口，没有对应的 +// 入站回合），空值诚实地表示这一点（与 channelbridge 的同一取舍）。 +// +// 未接线历史存储时它是空操作并返回零值（`Persisted=false`）—— 与编排器 / +// IM 镜像路径同一取舍：没存储不是错误，但**绝不能谎称已落库**。写库失败返回 +// 包装后的错误，由调用方让它可见（WARN + 返回给推送服务）。 +func (h *hub) persistProactiveToMain(ctx context.Context, msg adapter.UnifiedMessage) (history.Message, error) { + if h.history == nil { + return history.Message{}, nil + } + full, err := h.history.AppendFull(ctx, history.Message{ + SessionID: history.MainSessionID, + Role: history.RoleAssistant, + Content: msg.Content, + Platform: msg.Platform, + Mode: modeChat, + }) + if err != nil { + return history.Message{}, fmt.Errorf("gateway: persist proactive to main session: %w", err) + } + return full, nil +} diff --git a/internal/services/gateway/proactive_persist_test.go b/internal/services/gateway/proactive_persist_test.go new file mode 100644 index 0000000..67b392f --- /dev/null +++ b/internal/services/gateway/proactive_persist_test.go @@ -0,0 +1,280 @@ +package gateway + +import ( + "context" + "errors" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// 本文件锁定切片 P0 的**落库**那一半：主动消息无论走活跃客户端路由 +// （gateway hub）还是 IM 桥路由（channelbridge），都必须成为主会话 +// （`history.MainSessionID`）里**恰好一条** assistant 消息。 +// +// 为什么需要它：修复前活跃客户端路由只广播 `proactive_message` 帧、一行都不写， +// 于是「帧被 2 个连接收下」被读成「已推送」，刷新后什么都不剩（主人报的正是 +// 「后台思考的主动推送一个都没推出来」）。另一半风险是**双份** —— IM 适配器 +// 出站消息本来就会被镜像回主会话（channelbridge/proactive.go 的 persistToMain）， +// 若活跃路由也写一次、而一条推送同时走了两条路，主会话里就会出现两条。 +// 因此这里两个路由各自跑一遍，逐个断言「恰好一条」，并额外断言 +// 一条推送只会触发**一路**写入。 + +// proactiveStoreMsg 构造一条最小化的自主主动消息，等同于 backthink 的 +// buildProactiveMessage 在无渠道决策（通用哨兵值）时的产物。 +func proactiveStoreMsg(content string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "cyrene-backthink", + ChannelType: adapter.ChannelDirect, + Content: content, + ContentType: adapter.ContentText, + Sender: adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"}, + } +} + +// mainSessionRows 读回主会话的落库行（最新优先）。 +func mainSessionRows(t *testing.T, store *history.Store) []history.Message { + t.Helper() + rows, err := store.List(context.Background(), history.MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list main session: %v", err) + } + return rows +} + +// TestHub_SendProactive_PersistsAssistantMessageToMainSession 验证活跃客户端 +// 路由把主动消息写成主会话里的一条 assistant 消息（「气泡恒落」的根：刷新后 +// 仍在），且发出去的帧带上落库行的权威 id/时间戳 —— 客户端据此把实时气泡 +// 对齐到历史行，而不是让它刷新即丢。 +func TestHub_SendProactive_PersistsAssistantMessageToMainSession(t *testing.T) { + t.Parallel() + store := newHistoryStoreForGateway(t) + hb := newTestHub(nil, nil) + hb.history = store + owner := newOwnerConn(hb, 8) + hb.register(owner) + + delivery, err := hb.SendProactive(context.Background(), proactiveStoreMsg("记得喝水哦")) + if err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if delivery.Connections != 1 { + t.Fatalf("connections = %d, want 1", delivery.Connections) + } + if !delivery.Persisted { + t.Fatalf("Persisted = false, want true (主动消息必须落进主会话，否则刷新就没了)") + } + + rows := mainSessionRows(t, store) + if len(rows) != 1 { + t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条（>1 = 双份）", len(rows)) + } + row := rows[0] + if row.Role != history.RoleAssistant { + t.Fatalf("role = %q, want %q（她单方面开口的话，绝不能冒充主人的发言）", row.Role, history.RoleAssistant) + } + if row.SessionID != history.MainSessionID { + t.Fatalf("session = %q, want %q", row.SessionID, history.MainSessionID) + } + if row.Content != "记得喝水哦" { + t.Fatalf("content = %q, want the proactive text verbatim", row.Content) + } + if row.TurnID != "" { + t.Fatalf("turn_id = %q, want empty（主动推送不属于任何一路回合）", row.TurnID) + } + + ev := readFakeFrame(t, owner) + if ev.Type != wsProactiveMessage { + t.Fatalf("frame type = %q, want %q", ev.Type, wsProactiveMessage) + } + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame data type = %T, want map[string]any", ev.Data) + } + if data["id"] != row.ID { + t.Fatalf("frame id = %v, want the persisted row id %q", data["id"], row.ID) + } + if got, _ := data["createdAt"].(float64); int64(got) != row.CreatedAt { + t.Fatalf("frame createdAt = %v, want %d", data["createdAt"], row.CreatedAt) + } +} + +// failingHistory 让 AppendFull 失败，用于验证「落库失败**可见**」：帧照发 +// （实时气泡不能丢）、错误如实返回、Persisted=false（绝不谎报已推送）。 +type failingHistory struct { + HistoryStore + err error +} + +func (f failingHistory) AppendFull(context.Context, history.Message) (history.Message, error) { + return history.Message{}, f.err +} + +// TestHub_SendProactive_PersistFailureIsVisible 验证落库失败时不做假成功： +// 错误被返回、Persisted=false、Detail 的措辞由调用方如实反映，而帧仍然发出。 +func TestHub_SendProactive_PersistFailureIsVisible(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + hb.history = failingHistory{err: errors.New("db down")} + owner := newOwnerConn(hb, 8) + hb.register(owner) + + delivery, err := hb.SendProactive(context.Background(), proactiveStoreMsg("在吗")) + if err == nil { + t.Fatalf("err = nil, want the persist failure to be reported") + } + if delivery.Persisted { + t.Fatalf("Persisted = true, want false（没落库就是没落库）") + } + if delivery.Connections != 1 { + t.Fatalf("connections = %d, want 1（落库失败不该吞掉实时帧）", delivery.Connections) + } + var frameData any + if ev := readFakeFrame(t, owner); ev.Type != wsProactiveMessage { + t.Fatalf("frame type = %q, want %q", ev.Type, wsProactiveMessage) + } else { + frameData = ev.Data + } + data, ok := frameData.(map[string]any) + if !ok { + t.Fatalf("frame data type = %T, want map[string]any", frameData) + } + if _, hasID := data["id"]; hasID { + t.Fatalf("frame carries id %v, want the key omitted（没有落库行就没有权威 id）", data["id"]) + } +} + +// fakeIMAdapter 是 IM 路由测试所需的最小适配器（只用到连接态与出站）。 +type fakeIMAdapter struct { + platform string + sends []adapter.UnifiedMessage +} + +func (f *fakeIMAdapter) Name() string { return f.platform } +func (f *fakeIMAdapter) Config() map[string]any { return map[string]any{} } +func (f *fakeIMAdapter) Connect(context.Context) error { return nil } +func (f *fakeIMAdapter) Disconnect(context.Context) error { return nil } +func (f *fakeIMAdapter) IsConnected() bool { return true } +func (f *fakeIMAdapter) HealthCheck(context.Context) error { return nil } +func (f *fakeIMAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) { + <-ctx.Done() +} +func (f *fakeIMAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error { + f.sends = append(f.sends, resp.Messages...) + return nil +} +func (f *fakeIMAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error { + f.sends = append(f.sends, msg) + return nil +} +func (f *fakeIMAdapter) SetMessageHandler(adapter.MessageHandler) {} +func (f *fakeIMAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() } +func (f *fakeIMAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { + return nil, nil +} + +// stubBridgeChatter 满足 channelbridge.Chatter（本测试不会跑回合）。 +type stubBridgeChatter struct{} + +func (stubBridgeChatter) Run(context.Context, string, []model.Message) (*loop.Result, error) { + return &loop.Result{Text: "hi"}, nil +} + +// stubBridgeSessions 满足 channelbridge.SessionStore（本测试不会走入站路径）。 +type stubBridgeSessions struct{} + +func (stubBridgeSessions) GetOrCreate(context.Context, adapter.UnifiedMessage, *user.OwnerProfile) (session.Session, error) { + return session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}, nil +} + +// newProactiveTestBridge 装配一条接在**真历史存储**上的 IM 桥（一个已连接、 +// 在白名单里的主人渠道），使 IM 路由的落库能被真库断言。 +func newProactiveTestBridge(t *testing.T, store *history.Store, onPersisted *[]history.Message) (*channelbridge.Bridge, *fakeIMAdapter) { + t.Helper() + owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "telegram", AccountID: "100"}}} + var hook func(history.Message) + if onPersisted != nil { + hook = func(m history.Message) { *onPersisted = append(*onPersisted, m) } + } + b, err := channelbridge.New(channelbridge.Options{ + Log: gwTestLogger(), + Loop: stubBridgeChatter{}, + Sessions: stubBridgeSessions{}, + Owner: owner, + History: store, + OnPersisted: hook, + }) + if err != nil { + t.Fatalf("new bridge: %v", err) + } + ad := &fakeIMAdapter{platform: "telegram"} + b.Add(ad) + return b, ad +} + +// TestChannelBridge_SendProactive_MirrorsExactlyOnce 是**现状核查**的可执行 +// 证据：IM 路由本来就会把主动消息镜像回主会话（且只镜像一次）。因此在这条 +// 路由上再加一次落库就是双份 —— 活跃路由的落库正是靠「两条路由互斥」避开它的。 +func TestChannelBridge_SendProactive_MirrorsExactlyOnce(t *testing.T) { + t.Parallel() + store := newHistoryStoreForGateway(t) + var persisted []history.Message + bridge, ad := newProactiveTestBridge(t, store, &persisted) + + if err := bridge.SendProactive(context.Background(), proactiveStoreMsg("记得喝水哦")); err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if len(ad.sends) != 1 { + t.Fatalf("IM sends = %d, want 1", len(ad.sends)) + } + rows := mainSessionRows(t, store) + if len(rows) != 1 { + t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条（>1 = 双份）", len(rows)) + } + if rows[0].Role != history.RoleAssistant || !strings.Contains(rows[0].Content, "记得喝水哦") { + t.Fatalf("row = %+v, want an assistant message carrying the proactive text", rows[0]) + } + if len(persisted) != 1 || persisted[0].ID != rows[0].ID { + t.Fatalf("OnPersisted 应恰好收到刚落地的那一条（Web 端据此实时镜像），得到 %+v", persisted) + } +} + +// TestProactiveRoutes_EachRouteAddsExactlyOneRow 把两条路由对**同一条**主动 +// 消息各跑一次：主会话里应恰好两条（每条路由各一条）。它同时钉住两件事 —— +// ① 两条路由都真的落库（不再有「只发帧不落库」的那条）；② 没有一条路由写两份 +// （若活跃路由也走 IM 镜像那一次，这里会是 3 条）。 +func TestProactiveRoutes_EachRouteAddsExactlyOneRow(t *testing.T) { + t.Parallel() + store := newHistoryStoreForGateway(t) + hb := newTestHub(nil, nil) + hb.history = store + hb.register(newOwnerConn(hb, 8)) + bridge, _ := newProactiveTestBridge(t, store, nil) + + msg := proactiveStoreMsg("两条路各跑一次") + if _, err := hb.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("active route err = %v", err) + } + if got := len(mainSessionRows(t, store)); got != 1 { + t.Fatalf("活跃路由之后主会话有 %d 条，want 1", got) + } + if err := bridge.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("im route err = %v", err) + } + rows := mainSessionRows(t, store) + if len(rows) != 2 { + t.Fatalf("两条路由各跑一次后主会话有 %d 条，want 2（3 条 = 有一条路由写了双份）", len(rows)) + } + for _, row := range rows { + if row.Role != history.RoleAssistant { + t.Fatalf("row role = %q, want assistant", row.Role) + } + } +} diff --git a/internal/services/gateway/proactive_test.go b/internal/services/gateway/proactive_test.go index 3218100..a1e33ac 100644 --- a/internal/services/gateway/proactive_test.go +++ b/internal/services/gateway/proactive_test.go @@ -39,8 +39,8 @@ func TestHub_SendProactive_OwnerAndAdmin(t *testing.T) { if err != nil { t.Fatalf("SendProactive err = %v", err) } - if n != 2 { - t.Fatalf("count = %d, want 2 (owner+admin; viewer and bare excluded)", n) + if n.Connections != 2 { + t.Fatalf("count = %d, want 2 (owner+admin; viewer and bare excluded)", n.Connections) } for _, c := range []*wsConn{owner, admin} { @@ -82,8 +82,8 @@ func TestHub_SendProactive_NoOnlineOwner(t *testing.T) { if err != nil { t.Fatalf("SendProactive err = %v", err) } - if n != 0 { - t.Fatalf("count = %d, want 0", n) + if n.Connections != 0 { + t.Fatalf("count = %d, want 0", n.Connections) } } @@ -113,8 +113,8 @@ func TestServer_SendProactive_DelegatesToHub(t *testing.T) { if err != nil { t.Fatalf("SendProactive err = %v", err) } - if n != 1 { - t.Fatalf("count = %d, want 1", n) + if n.Connections != 1 { + t.Fatalf("count = %d, want 1", n.Connections) } ev := readFakeFrame(t, c) if ev.Type != "proactive_message" {