diff --git a/Docs/api-client.md b/Docs/api-client.md index ab2c22f..4350f05 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -29,8 +29,8 @@ | 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}`。**会话持久化(design/16)**:注入 HistoryStore 后返回持久化会话元数据 `{data:[{id,type("main"|"sub"),title?,createdAt,pinned,lastMessage?}]}`(**排序(D5):`pinned=true` 优先,其次主会话 `id="web"` 恒在前,再按 `createdAt` 升序**;次级 `web:`;`pinned` 恒返回 bool;`lastMessage` 为**最后一条消息预览** `{content,role(user|assistant),createdAt}`,content 折叠空白截前 60 rune,空会话省略);未注入时退化回渠道会话列表 | | 5 | `/api/v1/sessions` | POST | JWT(owner/admin) | ✅ | **新建次级会话** → `201 {data:{sessionId}}`(`sessionId="web:"`);未注入 HistoryStore 回 `501 not_implemented` | | 6 | `/api/v1/sessions/{id}/messages` | GET | JWT(owner/admin) | ✅ | **热分页拉历史**:`?limit=&before=&platform=&after=`(A5)默认 `limit=50`(上限 200)。**`before`**=上一页最旧 `createdAt`(毫秒),按时间**倒序**;**`after`**=<`createdAt` 毫秒>,按时间**升序**返回 `created_at > after` 的消息(**正向增量**,客户端把上次看到的 `createdAt` 当游标补断线期间漏的,解决重连只回放 50 条的缺口);**`after` 与 `before` 互斥**,同传回 `400`;负数/非数字回 `400`。`platform` 为**可选**来源平台过滤(design/16 §4.4)——Web 端不传(=全量共享主会话),IM 适配器传**本平台**(如 `telegram`)只拉本渠道消息;`deviceId` 为**客户端具体端标识**;`senderName` 为**第三方发送者昵称**;**`q`**(D2)=**会话内内容搜索**(大小写不敏感 ASCII LIKE,`%`/`_`/`\` 已转义为字面量),仍按时间**倒序**、`before`/`limit` 同分页语义;`q` 与 `after` **互斥**(同传 `400`),空 `q` 视为未传(走普通分页),搜索**严格限定在 `{id}` 会话内**(绝不跨会话,群聊/陌生人隔离会话不会外泄);非法 `limit`/`before`/`after` 回 `400`;未注入 HistoryStore 回 `501` | -| 6a | `/api/v1/messages/{id}` | PATCH | JWT(owner/admin) | ✅ | **编辑单条消息**(D3):body `{content}`(非空、≤4000 rune,空/纯空白/超长回 `400`)→ `{data:{message}}`(更新后的完整消息 DTO,含 `id/sessionId/role/content/.../createdAt`;**只改 `content`**,role/时间戳/平台保持不变);消息不存在回 `404 not_found`;未注入 HistoryStore 回 `501`。**重发不单独设端点**——客户端用「编辑后的内容再 `POST /chat`」即可(避免重复轮次语义) | -| 6b | `/api/v1/messages/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除单条消息**(D3)→ `{data:{deleted:}}`;不存在回 `404 not_found`;未注入 HistoryStore 回 `501` | +| 6a | `/api/v1/messages/{id}` | PATCH | JWT(owner/admin) | ✅ | **编辑单条消息**(D3):body `{content}`(非空、≤4000 rune,空/纯空白/超长回 `400`)→ `{data:{message}}`(更新后的完整消息 DTO,含 `id/sessionId/role/content/.../createdAt`;**只改 `content`**,role/时间戳/平台保持不变);消息不存在回 `404 not_found`;未注入 HistoryStore 回 `501`。**成功后广播下行 `message_edited`**(§3.8,账号其它端免刷新同步)。**重发不单独设端点**——客户端用「编辑后的内容再 `POST /chat`」即可(避免重复轮次语义) | +| 6b | `/api/v1/messages/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除单条消息**(D3)→ `{data:{deleted:}}`;不存在回 `404 not_found`;未注入 HistoryStore 回 `501`。**成功后广播下行 `message_deleted`**(§3.8,账号其它端按 `id` 移除) | | 7 | `/api/v1/sessions/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除次级会话及其消息** → `{data:{deleted:}}`;主会话不可删回 `403 forbidden`;未注入 HistoryStore 回 `501` | | 8 | `/api/v1/sessions/{id}` | PATCH | JWT(owner/admin) | ✅ | **会话元数据更新**(A4 改名 + D5 置顶):body `{title?,pinned?}`(**至少一个字段**,都缺回 `400`)。`title` trim、≤80 rune,空/超长回 `400`;`pinned` 为 bool(置顶/取消,主会话与次级会话均可)。响应只回显**本次变更**的字段:`{data:{id,title?}}` / `{data:{id,pinned?}}` / 两者都改则都有(**title-only 保持 A4 原 `{id,title}` 形状**);不存在回 `404`;主会话(`id="web"`)可改名/置顶但不可删;未注入 HistoryStore 回 `501`。另:会话**首条用户消息到达时自动命名** title(取前 20 rune 折叠空白;纯空白回落 `主会话`/`次级会话`),列表不再空名 | | 9 | `/api/v1/personas` | GET | JWT | ✅ | `{data:[persona...]}` | @@ -138,7 +138,7 @@ ## 3. WebSocket `/ws/chat`(实时) - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`;未认证在握手前回 `401` JSON。 -- **下行(服务端→客户端)**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`(`turnId`/`seq` 为流式关联字段,见 §3.1;`sessionId` 为**会话持久化(design/16)新增**,能确定会话的事件带上,供客户端按激活会话过滤)——`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)。**会话编排状态(design/15,均为广播,`data.sessionId` 区分会话)**:`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`(见 §3.2)。语音回复走 `voice_output`(定向回请求连接,非广播)。 +- **下行(服务端→客户端)**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`(`turnId`/`seq` 为流式关联字段,见 §3.1;`sessionId` 为**会话持久化(design/16)新增**,能确定会话的事件带上,供客户端按激活会话过滤)——`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)。**会话编排状态(design/15,均为广播,`data.sessionId` 区分会话)**:`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`(见 §3.2)。语音回复走 `voice_output`(定向回请求连接,非广播)。 > **`history` 回放事件(design/16,新增)**:WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,platform?,deviceId?,createdAt}]}}`——回放主会话最新 50 条(`thinking` 已剥离,不泄给非 owner;默认主会话,次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取)。客户端据此在重连/新开连接后补全上下文。消息体含 `senderName?`(第三方昵称,design/16 §4.6,主人私聊为空)。 @@ -205,6 +205,13 @@ - **客户端处理**:`mood` → `setMood(data.mood)`(`src/lib/ws.ts`),顶栏/侧栏显示「心情 · 」;消息体内 `mood` 由聊天气泡显示(`chat.tsx`)。 - **状态持久性**:情绪与好感度为**进程内、按角色**状态,**重启回到角色配置的 `initialState`**(design/01 未要求落库;如需跨重启保留,待主人确认后加表)。 +### 3.8 消息编辑 / 删除实时事件(`message_edited` / `message_deleted`,D3 实时同步) +- **触发**:`PATCH /api/v1/messages/{id}`(改 content)或 `DELETE /api/v1/messages/{id}` **成功写库后**由网关广播(与 `message_persisted` 同一条 hub 广播路径,**全连接广播 + 客户端按会话过滤**——hub 无按会话订阅;消息体的唯一真源始终在后端,这两个事件只驱动渲染同步,不携带新真源)。 +- **`message_edited`**:`{type:"message_edited", sessionId, data:{id, content, mode?, mood?, createdAt?, ...}}`——`data` 为**编辑后的完整消息 DTO**(与 §3.3 的 `message.message` 同形,至少含 `id`/`content`/`sessionId`/`createdAt`,另带 `role`/`mode?`/`mood?`/`image?`/`platform?`/`deviceId?`);`thinking` 已剥离(与 `message_persisted` 一致)。**只改 `content`**,`role`/时间戳/平台等保持原值。 +- **`message_deleted`**:`{type:"message_deleted", sessionId, data:{id}}`——只带被删除消息的 `id`(`sessionId` 为该消息所属会话,由后端在删除前读取后填入;客户端按它过滤,**不**据此删除别的会话的行)。 +- **客户端处理**:与 `message_persisted` **同一过滤规则**——仅当 `sessionId === activeSessionId` 时处理(`message_deleted` 的 `sessionId` 缺失时按 `id` 兜底:消息 `id` 全局唯一,只会命中它自己那条)。`message_edited` → **按 `id` 原位更新该条的 `content`**(不新增行:找不到该 `id` 即忽略,避免把别的会话的消息注入当前列表);`message_deleted` → **按 `id` 从列表移除**。两个操作都**不触发整页重拉**——这正是它相对「重拉绕行」的收益。 +- **失败不广播**:`404`(消息不存在)/ `400`(内容非法)/ `501`(未注入 HistoryStore)**不**产生任何事件,客户端不会因一次失败请求而丢掉本地消息。 + ## 4. 客户端接入指引(Web / 移动 / 原生 / IM bridge 通用) ### 4.1 一次性接入流程(三步走) @@ -219,9 +226,9 @@ - **热分页**:`limit` 默认 50(上限 200),`before` = 当前最旧一条的 `createdAt`(毫秒),向上翻页。`platform` 可选(IM 适配器传本平台只拉本渠道;Web/通用客户端不传 = 全量共享主会话)。 - **搜索**:`q` 非空 = **会话内**内容搜索(倒序,仍支持 `before`/`limit` 分页);空 `q` = 普通分页;`q` 与 `after` 互斥。**不跨会话**。 - 返回**按时间倒序**,需 `reverse()` 成旧→新用于渲染。 -- **编辑 / 删除单条消息**:`PATCH /api/v1/messages/{id}` `{content}`(≤4000 rune)→ `{data:{message}}`;`DELETE /api/v1/messages/{id}` → `{data:{deleted:}}`。不存在回 `404`。**重发**用「编辑后内容再 `POST /chat`」即可(无独立重发端点)。 +- **编辑 / 删除单条消息**:`PATCH /api/v1/messages/{id}` `{content}`(≤4000 rune)→ `{data:{message}}`;`DELETE /api/v1/messages/{id}` → `{data:{deleted:}}`。不存在回 `404`。**重发**用「编辑后内容再 `POST /chat`」即可(无独立重发端点)。两者成功后都会广播(`message_edited` / `message_deleted`,§3.8),同账号其它端**免刷新**同步。 +- **实时增量**:连接 WS 后,消息落库会广播 `message_persisted`(§3.3)——**用其中的 `id` 去重**,`sessionId` 匹配激活会话才渲染;新开/重连连接首帧即 `history` 回放(主会话最近 50 条)。**消息被编辑/删除**则分别广播 `message_edited` / `message_deleted`(§3.8)——同一 `sessionId` 过滤规则,前者按 `id` 原位改 `content`、后者按 `id` 移除,都不需要重拉整页。 - **发消息**:`POST /api/v1/chat` `{text, deviceId?, image?, stream:true}`。`stream:true` 走 WS 流式;非流式返回完整 `reply`(会话正忙时 `202 queued`、排队满 `409 chat_busy`,见 §3.2)。 -- **实时增量**:连接 WS 后,消息落库会广播 `message_persisted`(§3.3)——**用其中的 `id` 去重**,`sessionId` 匹配激活会话才渲染;新开/重连连接首帧即 `history` 回放(主会话最近 50 条)。 ### 4.3 deviceId 与 platform 规范(design/16 §4.4-4.5) - **`platform`**(来源平台):`web` / 渠道如 `telegram`、`weixin-ilink`、`qq-obv11`。Web 端恒 `web`;IM 渠道消息带各自平台。**同一平台多个端 → 相同 platform、不同 deviceId。** @@ -240,6 +247,7 @@ - 语音:上行 `voice_input`(base64 音频或已转写 text),下行 `voice_output`(合成音频 base64 + text)。 ## 5. 变更记录 +- `v0.34`(**D3 收尾:消息编辑 / 删除的 WS 实时广播**,对客户端**新增两个下行事件、无 REST 契约变更**):`PATCH /api/v1/messages/{id}` 成功后广播 **`message_edited`**(`{type, sessionId, data:<更新后的完整消息DTO>}`,`thinking` 已剥离)、`DELETE /api/v1/messages/{id}` 成功后广播 **`message_deleted`**(`{type, sessionId, data:{id}}`);两者复用 `message_persisted` 的同一 hub 广播路径(**全连接广播 + 客户端按 `sessionId` 过滤**,hub 无按会话订阅),客户端**按 `id` 原位改 `content` / 按 `id` 移除**,不再需要「改/删后重拉」的绕行。`message_deleted` 的 `sessionId` 由后端在删除前读取消息所属会话填入(`HistoryStore` 新增 `GetMessage`;读不到时客户端按全局唯一的 `id` 兜底)。`404`/`400`/`501` 等失败路径**不广播**。Web 端已消费(`lib/ws.ts` + `store/useAppStore` 新增 `updateMessageById`/`removeMessageById`)。详见 §3.8。 - `v0.33`(**A2 记忆异步提取 / A4 可观测性 Tracing 接线 / A6 通知页后端**):① **A6 新增 `GET /api/v1/admin/notifications`**(owner/admin 读,**全新端点**):把**现有源**(审计 `audit_logs` + backthink 主动推送记录)聚合为**统一只读通知 feed**,`type`/`before`/`limit` 过滤与游标见 §2「通知聚合契约」;**不新建表**,未注入的源该类型为空、不伪造;未读游标后置。② **A4 trace 端点由恒空变真实数据**(**无契约变更**,`GET /admin/observability/trace` 与 `metrics` 的 `llmCalls`/`subagentEvents` 从此反映 loop/工具/编排的真实 span):此前 `TraceSink` 从未接线、环形缓冲无人喂;现在组合根把 loop 产生的 span 扇出到网关观测缓冲。响应字段/形状不变。③ **A2 LLM 异步记忆提取接线**(**对客户端无变更**):loop 回合结束后改由 LLM 提取器**后台异步**抽取记忆(不阻塞回复、失败仅记日志),**群聊/陌生人仍只读不写**;无可用模型凭据时回落确定性 `ManualExtractor`(`记住 X` 路径不变)。 - `v0.32`(**A1 情绪系统接线:情绪/好感度真跑起来**):后端把既有的 `persona/emotion` 状态机接进 loop 回合边界——每轮结束按该轮用户消息的情绪信号移动**该角色自己**的情绪状态机(沿配置转移表走一步)并调整好感度(按角色 `affection.delta`);无信号轮次衰减一步趋向 `calm`。① **消息 `mood` 不再恒空**:assistant 消息落库带该轮情绪,`message_persisted` / `history` 回放 / `GET /sessions/{id}/messages` 的 `mood` 为真实值(user 消息不带)。② **新增下行 `mood` 事件**(`{data:{mood,personaId,affection}}`,**仅情绪变化时**广播,见 §3.7),前端顶栏「心情 · x」与消息气泡接真数据。③ **隔离**:群聊/陌生人(`MemWritable=false`)情绪只读不写——不改情绪/好感度、不发事件、该轮 `mood` 空。④ **持久性**:情绪/好感度为进程内按角色状态,重启回 `initialState`(未落库;design/01 未要求)。**触发词表与「每轮一步」为 design/01 未规定处的最小实现,待主人确认**。**契约新增(向后兼容):`mood` 事件 + 消息 `mood` 变真实值。** - `v0.31`(**Android 协同批次 ③b:B5 剩余 501 + D1 推送出口钩子**):① **B5-知识库删除由 501 改为真实实现**:`POST /api/v1/admin/rag/delete` body `{"id"}` → `200 {data:{id,deleted:true}}`——从项目索引删除该文档条目并**级联删除全部章节**(单事务);不存在回 `404 not_found`,`id` 空回 `400`,未注入 store 回 `501`;路由由 admin 收紧为 **owner-only**(与其它 admin 写一致)+ 审计 `rag.delete`。② **B5-新增用户 / IoT 场景 / IoT 规则仍诚实 501**:无用户表(登录仅 `.env` 单 admin)、IoT 无场景/规则模型与存储(`iot.Provider` 只有设备列表/控制/状态),错误消息明确缺什么,**不伪造 store/假成功**。③ **D1 推送出口钩子**:主动消息在 WS/IM 投递之外异步镜像到外部推送出口,**自建统一推送服务优先 → 个推(Getui)兜底**(个推 SDK 未接入,仅预留 `GETUI_*` 字段,不可达时日志注明待接入);`.env` 新增 `PUSH_EXPORTER_ENABLED/URL/TOKEN/TIMEOUT_SEC`,token 只走 `Authorization` 头、不落日志;**对客户端无新端点**(详见 §3.6)。 diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index 3bf7e4b..ff0149b 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -1,4 +1,4 @@ -import { useAppStore } from "../store/useAppStore"; +import { messageEventApplies, useAppStore } from "../store/useAppStore"; import { useModeStore } from "../store/useModeStore"; import { getAuthToken, isTokenExpired, onAuthExpired, setAuthToken } from "./api"; import { @@ -13,7 +13,7 @@ import { playVoiceAudio, toRecord, } from "./wsParse"; -import { MODES, type AwakenState, type Mode, type WsUpEvent } from "./types"; +import { MODES, type AwakenState, type ChatMessage, type Mode, type WsUpEvent } from "./types"; /** * Realtime client for `/ws/chat`. Reconnects with exponential backoff and @@ -41,7 +41,43 @@ function wsUrl(): string { return `${proto}://${window.location.host}/ws/chat${qs}`; } -function dispatch(ev: { type: string; data: unknown }): void { +/** The fields of a `message_edited` payload the chat needs: the edited message's + * id and its new content. `sessionId` is carried inside the broadcast DTO (and on + * the frame envelope) so the frame can be filtered; `mode`/`mood` are optional — + * an edit never changes them, but the DTO carries them for completeness. */ +interface MessageEdited { + id: string; + content: string; + sessionId?: string; + mode?: Mode; + mood?: string; +} + +/** Narrow a `message_edited` payload (the updated message DTO), or null when it is + * unusable. `content` is required: an edit always carries the new text, and a + * frame without it must not blank out a rendered bubble. */ +function parseMessageEdited(data: unknown): MessageEdited | null { + const r = toRecord(data); + if (!r || typeof r.id !== "string" || r.id === "" || typeof r.content !== "string") { + return null; + } + return { + id: r.id, + content: r.content, + sessionId: typeof r.sessionId === "string" ? r.sessionId : undefined, + mode: + typeof r.mode === "string" && MODES.includes(r.mode as Mode) + ? (r.mode as Mode) + : undefined, + mood: typeof r.mood === "string" ? r.mood : undefined, + }; +} + +/** Route one decoded downstream frame into the store. Exported (with a `frame` + * shaped first argument) so the frame→store mapping can be exercised as a pure + * logic test without a browser: this project has no jsdom/test runner, so the + * socket lifecycle around it stays untested while the mapping does not have to. */ +export function dispatchFrame(ev: { type: string; sessionId?: string; data: unknown }): void { const s = useAppStore.getState(); const r = toRecord(ev.data); @@ -87,6 +123,32 @@ function dispatch(ev: { type: string; data: unknown }): void { } break; } + case "message_edited": { + // design/16 §4.2 (D3 edit path) — the stored message's content was changed + // and broadcast with the updated DTO. The backend row stays the single source + // of truth; this only re-renders the bubble, and only in the session it + // belongs to (same filter as message_persisted). Patch by id — an id we do not + // hold is another session's (or an already-gone) message, never a new row. + const edit = parseMessageEdited(ev.data); + if (!edit) break; + if (messageEventApplies(edit.sessionId ?? ev.sessionId, s.activeSessionId)) { + const patch: Partial = { content: edit.content }; + if (edit.mode) patch.mode = edit.mode; + if (edit.mood) patch.mood = edit.mood; + s.updateMessageById(edit.id, patch); + } + break; + } + case "message_deleted": { + // design/16 §4.2 (D3 delete path) — the message was removed server-side, so + // drop it by id instead of waiting for a manual re-read. The id is globally + // unique; sessionId is advisory and, when present, must be the active one. + const id = r && typeof r.id === "string" ? r.id : ""; + if (id && messageEventApplies(ev.sessionId, s.activeSessionId)) { + s.removeMessageById(id); + } + break; + } case "channel_persisted": { // design/16 §4.6 §4 — an isolated (group/stranger) channel message was // persisted and broadcast for the IM adapter log page. The web chat has no @@ -305,7 +367,13 @@ function openSocket(): void { const parsed = JSON.parse(String(msg.data)) as unknown; const r = toRecord(parsed); if (!r) return; - dispatch({ type: typeof r.type === "string" ? r.type : "", data: r.data }); + dispatchFrame({ + type: typeof r.type === "string" ? r.type : "", + // The frame envelope's sessionId is what scopes a payload that does not + // carry its own (see the `message_deleted` case in dispatch). + sessionId: typeof r.sessionId === "string" ? r.sessionId : undefined, + data: r.data, + }); } catch { // ignore malformed frames } diff --git a/frontend/web/src/store/messageSync.test.ts b/frontend/web/src/store/messageSync.test.ts new file mode 100644 index 0000000..7953f2a --- /dev/null +++ b/frontend/web/src/store/messageSync.test.ts @@ -0,0 +1,113 @@ +// Logic test for the D3 realtime-sync frames (`message_edited` / `message_deleted`). +// The project has no frontend test runner and no jsdom, so this is a dependency-free +// plain-node check: it drives the real frame decoder (`dispatchFrame` in lib/ws.ts) +// and the real store, and throws on the first failure. +// +// Run it with the esbuild that already ships with vite (no new dependency, no +// package.json change); the banner supplies the `window.localStorage` the store's +// persist middleware probes, which is simply absent under node: +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/store/messageSync.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-message-sync.test.mjs \ +// && node /tmp/cyrene-message-sync.test.mjs +// +// `tsc --noEmit` (part of `npm run build`) type-checks this file, which is why it +// imports nothing from node:*. + +import { dispatchFrame } from "../lib/ws"; +import { messageEventApplies, useAppStore } from "./useAppStore"; +import type { ChatMessage } from "../lib/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"; +const OTHER = "web:sub-1"; + +/** Reset the store to a focused session holding a fixed message list. */ +function seed(messages: ChatMessage[], activeSessionId: string): void { + useAppStore.setState({ activeSessionId, messages }); +} + +function messages(): ChatMessage[] { + return useAppStore.getState().messages; +} + +function base(): ChatMessage[] { + return [ + { id: "m1", sessionId: MAIN, role: "user", content: "old", mode: "chat", createdAt: 1000 }, + { id: "m2", sessionId: MAIN, role: "assistant", content: "reply", createdAt: 2000 }, + ]; +} + +// ① Session filter: a frame scoped to another conversation never touches this one. +ok(messageEventApplies(MAIN, MAIN), "same session applies"); +ok(!messageEventApplies(OTHER, MAIN), "another session does not apply"); +ok(messageEventApplies(undefined, MAIN), "an unscoped (id-only) frame applies"); + +// ② message_edited patches the addressed bubble in place, by id, in the active +// session — and leaves its siblings alone. +seed(base(), MAIN); +dispatchFrame({ type: "message_edited", sessionId: MAIN, data: { id: "m1", sessionId: MAIN, content: "new", mode: "chat", createdAt: 1000 } }); +same( + messages().map((m) => m.content), + ["new", "reply"], + "edit patches only the addressed message" +); +same(messages().find((m) => m.id === "m1")?.mode, "chat", "edit keeps the message mode"); +ok(messages().find((m) => m.id === "m1")?.createdAt === 1000, "edit keeps createdAt"); + +// ③ An edit frame for another session is ignored (no cross-session write) … +seed(base(), MAIN); +dispatchFrame({ type: "message_edited", sessionId: OTHER, data: { id: "m3", sessionId: OTHER, content: "elsewhere" } }); +same(messages().map((m) => m.content), ["old", "reply"], "edit for another session does not apply"); + +// … and an edit for an id we do not hold never appends a row. +seed(base(), MAIN); +dispatchFrame({ type: "message_edited", sessionId: MAIN, data: { id: "m-ghost", sessionId: MAIN, content: "ghost" } }); +same(messages().length, 2, "edit for an unknown id does not append"); + +// A malformed frame (no content) must not blank out a rendered bubble. +seed(base(), MAIN); +dispatchFrame({ type: "message_edited", sessionId: MAIN, data: { id: "m1", sessionId: MAIN } }); +same(messages()[0].content, "old", "content-less edit frame is dropped"); + +// ④ message_deleted removes the addressed bubble in the active session only. +seed(base(), MAIN); +dispatchFrame({ type: "message_deleted", sessionId: MAIN, data: { id: "m1" } }); +same(messages().map((m) => m.id), ["m2"], "delete removes the addressed message"); + +// A delete frame for another session is ignored. +seed(base(), MAIN); +dispatchFrame({ type: "message_deleted", sessionId: OTHER, data: { id: "m1" } }); +same(messages().map((m) => m.id), ["m1", "m2"], "delete for another session does not apply"); + +// An unscoped delete still applies (the id is globally unique, so it can only ever +// match its own message). +seed(base(), MAIN); +dispatchFrame({ type: "message_deleted", data: { id: "m2" } }); +same(messages().map((m) => m.id), ["m1"], "unscoped delete applies by id"); + +// A delete for an id we do not hold is a no-op. +seed(base(), MAIN); +dispatchFrame({ type: "message_deleted", sessionId: MAIN, data: { id: "m-ghost" } }); +same(messages().length, 2, "delete for an unknown id is a no-op"); + +// The store actions are also safe on their own (defensive: duplicate frames). +seed(base(), MAIN); +useAppStore.getState().updateMessageById("m-ghost", { content: "x" }); +useAppStore.getState().removeMessageById("m-ghost"); +same(messages().length, 2, "store id-actions no-op on an unknown id"); + +console.log(`message_edited/message_deleted logic test: ${checks} checks passed`); diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index 70e7c31..88ef240 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -73,6 +73,13 @@ export interface AppStore { * it also adopts a same-role local optimistic bubble (rewriting its uid to the * backend id) so a message never renders twice (design/16 §4.3 dedup). */ appendOrUpdateMessage: (msg: ChatMessage) => void; + /** Patch one already-rendered message in place by id — the `message_edited` WS + * path. A no-op when the id is not in the active session, so an edit broadcast + * for another conversation can never inject a row here. */ + updateMessageById: (id: string, patch: Partial) => void; + /** Drop one message by id — the `message_deleted` WS path. A no-op when the id + * is absent (another session's message, or a duplicate frame). */ + removeMessageById: (id: string) => void; addUserMessage: (text: string, mode?: Mode, image?: string, deviceId?: string) => void; beginAssistant: (mode?: Mode, deviceId?: string) => string; @@ -192,6 +199,21 @@ export const useAppStore = create()( return { messages: [...s.messages, msg] }; }), + updateMessageById: (id, patch) => + set((s) => { + const idx = s.messages.findIndex((m) => m.id === id); + if (idx === -1) return s; + const messages = s.messages.slice(); + messages[idx] = { ...messages[idx], ...patch }; + return { messages }; + }), + + removeMessageById: (id) => + set((s) => { + const messages = s.messages.filter((m) => m.id !== id); + return messages.length === s.messages.length ? s : { messages }; + }), + addUserMessage: (text, mode, image, deviceId) => set((s) => ({ messages: [ @@ -383,3 +405,20 @@ export function lastStreamingMessage(s: AppStore): ChatMessage | undefined { const last = s.messages[s.messages.length - 1]; return last?.role === "assistant" ? last : undefined; } + +/** Does an edit/delete frame scoped to `sessionId` belong to the conversation the + * UI is showing? `message_edited` / `message_deleted` carry the owning session so + * the rule matches `message_persisted`: only the active session's messages are + * rendered, so another conversation's frame must not touch this list. + * + * A frame without a session id is id-scoped (the backend could not resolve the + * conversation — see the gateway's optional message lookup). It still applies: + * message ids are globally unique, so the by-id update/removal can only ever match + * a message of its own session, and an id we do not hold is a no-op either way. */ +export function messageEventApplies( + sessionId: string | undefined, + activeSessionId: string +): boolean { + if (!sessionId) return true; + return sessionId === activeSessionId; +} diff --git a/internal/runtime/history/message_ops.go b/internal/runtime/history/message_ops.go index f058240..f9bb2df 100644 --- a/internal/runtime/history/message_ops.go +++ b/internal/runtime/history/message_ops.go @@ -9,10 +9,11 @@ import ( ) // This file owns the single-message operations behind the Android collaboration batch -// ③a (D2 search, D3 edit/delete). Both are strictly session-scoped or id-scoped: a -// message is only ever reached through the session it belongs to (Search) or its own -// opaque id (UpdateMessage/DeleteMessage), so an isolated group/stranger session can -// never surface in another session's results (project red line). +// ③a (D2 search, D3 edit/delete) plus the id-scoped read the D delete broadcast needs. +// All are strictly session-scoped or id-scoped: a message is only ever reached through +// the session it belongs to (Search) or its own opaque id (GetMessage/UpdateMessage/ +// DeleteMessage), so an isolated group/stranger session can never surface in another +// session's results (project red line). // Search returns up to limit messages for sessionID whose content contains query, // newest first. The match is a case-insensitive ASCII LIKE (SQLite's default for @@ -91,6 +92,14 @@ func (s *Store) DeleteMessage(ctx context.Context, id string) error { return nil } +// GetMessage reads the message with id, returning ErrMessageNotFound when no such +// row exists. It is the id-scoped read the transport uses before removing a +// message: DeleteMessage leaves nothing to ask afterwards, yet the broadcast that +// tells other clients to drop the row is session-scoped. +func (s *Store) GetMessage(ctx context.Context, id string) (Message, error) { + return s.messageByID(ctx, id) +} + // messageByID reads one message by id, returning ErrMessageNotFound when the row is // absent. func (s *Store) messageByID(ctx context.Context, id string) (Message, error) { diff --git a/internal/services/gateway/history_write.go b/internal/services/gateway/history_write.go index ca6ca14..edc56a7 100644 --- a/internal/services/gateway/history_write.go +++ b/internal/services/gateway/history_write.go @@ -1,6 +1,7 @@ package gateway import ( + "context" "encoding/json" "errors" "net/http" @@ -14,7 +15,9 @@ import ( // This file owns the session/message write endpoints added for the Android // collaboration batch ③a: PATCH /sessions/{id} (A4 rename + D5 pin), PATCH // /messages/{id} (D3 edit) and DELETE /messages/{id} (D3 delete). All are owner/admin -// history writes; the reads and DTOs live in history.go. +// history writes; the reads and DTOs live in history.go. The D3 edit/delete paths also +// broadcast their outcome over the WS hub (message_edited / message_deleted, see +// ws_hub.go) so the account's other browsers stay in sync without a manual re-read. // patchSessionRequest is the PATCH /api/v1/sessions/{id} body. Both fields are pointers // so an absent field is distinguishable from an explicit zero value: title renames the @@ -145,11 +148,20 @@ func (s *Server) handlePatchMessage(w http.ResponseWriter, r *http.Request) { writeError(w, apperr.Wrap(apperr.ErrInternal, "message_failed", "failed to update message")) return } + // Broadcast the edit so every other browser of the account re-renders the same + // bubble without a refresh (the message_persisted realtime-sync contract, D3 edit + // path). Thinking is stripped from the frame to match the persisted/replay + // contract; the REST reply keeps it because the messages routes are owner/admin-only. + edited := toHistoryMessage(msg) + edited.Thinking = "" + s.hub.broadcastMessageEdited(edited) writeJSON(w, http.StatusOK, toHistoryMessage(msg)) } // handleDeleteMessage removes a single message (D3). An unknown id answers 404, a -// missing HistoryStore 501. It is owner/admin-gated. +// missing HistoryStore 501. It is owner/admin-gated. A successful delete is +// broadcast as a `message_deleted` frame so every browser of the account drops the +// row without a refresh; the owning session is read before the row disappears. func (s *Server) handleDeleteMessage(w http.ResponseWriter, r *http.Request) { if s.history == nil { writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置(未注入 HistoryStore)", HTTPCode: http.StatusNotImplemented}) @@ -160,6 +172,7 @@ func (s *Server) handleDeleteMessage(w http.ResponseWriter, r *http.Request) { writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "message id is required")) return } + sessionID := s.messageSession(r.Context(), id) if err := s.history.DeleteMessage(r.Context(), id); err != nil { if errors.Is(err, history.ErrMessageNotFound) { writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "message %q not found", id)) @@ -169,5 +182,31 @@ func (s *Server) handleDeleteMessage(w http.ResponseWriter, r *http.Request) { writeError(w, apperr.Wrap(apperr.ErrInternal, "message_failed", "failed to delete message")) return } + s.hub.broadcastMessageDeleted(sessionID, id) writeJSON(w, http.StatusOK, map[string]any{"deleted": id}) } + +// messageLookup is the optional narrower view of HistoryStore used by DELETE to learn +// a message's owning session before it is removed (afterwards there is nothing left to +// ask). *history.Store implements it; a store that does not simply yields an unscoped +// frame, which the client resolves by the globally unique message id. +type messageLookup interface { + GetMessage(ctx context.Context, id string) (history.Message, error) +} + +// messageSession returns the session id owning message id, or "" when the store cannot +// answer (no lookup surface, unknown id, or a read error). A blank session is safe for +// the broadcast: the message id is globally unique, so a client that falls back to an +// id-only filter can only ever touch that message's own conversation. +func (s *Server) messageSession(ctx context.Context, id string) string { + lookup, ok := s.history.(messageLookup) + if !ok { + return "" + } + msg, err := lookup.GetMessage(ctx, id) + if err != nil { + s.log.Debug("history message session lookup", "err", err, "message", id) + return "" + } + return msg.SessionID +} diff --git a/internal/services/gateway/history_write_broadcast_test.go b/internal/services/gateway/history_write_broadcast_test.go new file mode 100644 index 0000000..bed4e46 --- /dev/null +++ b/internal/services/gateway/history_write_broadcast_test.go @@ -0,0 +1,165 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// These tests cover the D3 realtime-sync contract: a successful message edit or +// delete must reach the account's other browsers as a WS frame, not only the client +// that issued the REST call (otherwise every other browser has to re-read to converge). + +// TestPatchMessageBroadcastsEdited covers ①: PATCH /api/v1/messages/{id} answers 200 +// and pushes exactly one `message_edited` frame scoped to the edited message's own +// session, carrying the new content (and the mode/createdAt a client renders with). +// The frame's data is the updated message DTO with thinking stripped, matching the +// message_persisted contract; the REST reply keeps thinking (owner/admin-only read). +func TestPatchMessageBroadcastsEdited(t *testing.T) { + t.Parallel() + ctx := context.Background() + srv, ts, jwt, st := newHistoryEnv(t) + c := newFakeConn(srv.hub, 32) + srv.hub.register(c) + sub, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub: %v", err) + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + seedMessage := func(sessionID string) history.Message { + t.Helper() + return seedFullMessage(t, st, history.Message{ + SessionID: sessionID, Role: history.RoleUser, Content: "old body", + Mode: "chat", Thinking: "internal chain", CreatedAt: 1000, + }) + } + for _, tc := range []struct { + name string + sessionID string + }{ + {name: "main session", sessionID: history.MainSessionID}, + {name: "sub session", sessionID: sub}, + } { + t.Run(tc.name, func(t *testing.T) { + msg := seedMessage(tc.sessionID) + resp, body := doReq(t, http.MethodPatch, ts.URL+"/api/v1/messages/"+msg.ID, map[string]any{"content": "new body"}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("patch status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + // The REST reply is the full DTO for the issuing client. + var reply historyMessage + if err := json.Unmarshal(decodeBody(t, body)["data"], &reply); err != nil { + t.Fatalf("decode reply: %v (body=%s)", err, body) + } + if reply.Content != "new body" || reply.Thinking != "internal chain" { + t.Fatalf("reply = %+v, want edited content + kept thinking", reply) + } + + ev := readFakeFrame(t, c) + if ev.Type != "message_edited" { + t.Fatalf("frame type = %q, want message_edited (frame=%+v)", ev.Type, ev) + } + if ev.SessionID != tc.sessionID { + t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, tc.sessionID) + } + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame data = %+v, want object", ev.Data) + } + if data["id"] != msg.ID || data["content"] != "new body" { + t.Fatalf("frame data = %+v, want id=%s content=new body", data, msg.ID) + } + if data["sessionId"] != tc.sessionID { + t.Fatalf("frame data.sessionId = %v, want %q", data["sessionId"], tc.sessionID) + } + if data["mode"] != "chat" { + t.Fatalf("frame data.mode = %v, want chat", data["mode"]) + } + if _, isNum := data["createdAt"].(float64); !isNum { + t.Fatalf("frame data.createdAt = %v, want a number", data["createdAt"]) + } + if _, leaked := data["thinking"]; leaked { + t.Fatalf("frame data = %+v, thinking must be stripped from the broadcast", data) + } + if err := connHasNoFrame(c); err != nil { + t.Fatal(err) + } + }) + } +} + +// TestDeleteMessageBroadcastsDeleted covers ②: DELETE /api/v1/messages/{id} answers +// 200, really removes the row, and pushes one `message_deleted` frame carrying the id +// plus the owning session (read before the row disappears) so a remote browser can +// drop it without re-reading. +func TestDeleteMessageBroadcastsDeleted(t *testing.T) { + t.Parallel() + ctx := context.Background() + srv, ts, jwt, st := newHistoryEnv(t) + c := newFakeConn(srv.hub, 32) + srv.hub.register(c) + sub, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub: %v", err) + } + msg := seedFullMessage(t, st, history.Message{SessionID: sub, Role: history.RoleUser, Content: "to be removed", CreatedAt: 2000}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodDelete, ts.URL+"/api/v1/messages/"+msg.ID, nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("delete status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + + ev := readFakeFrame(t, c) + if ev.Type != "message_deleted" { + t.Fatalf("frame type = %q, want message_deleted (frame=%+v)", ev.Type, ev) + } + if ev.SessionID != sub { + t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, sub) + } + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame data = %+v, want object", ev.Data) + } + if data["id"] != msg.ID { + t.Fatalf("frame data.id = %v, want %s", data["id"], msg.ID) + } + if err := connHasNoFrame(c); err != nil { + t.Fatal(err) + } + + // The broadcast must not claim a delete that did not happen. + remaining, err := st.List(ctx, sub, 0, 50, "") + if err != nil { + t.Fatalf("list after delete: %v", err) + } + for _, m := range remaining { + if m.ID == msg.ID { + t.Fatalf("message %s still stored after a successful delete", msg.ID) + } + } +} + +// TestDeleteMessageUnknownIDBroadcastsNothing pins the failure side: a 404 delete +// (unknown id) must not push a `message_deleted` frame, or a client would drop a +// message that still exists. +func TestDeleteMessageUnknownIDBroadcastsNothing(t *testing.T) { + t.Parallel() + srv, ts, jwt, _ := newHistoryEnv(t) + c := newFakeConn(srv.hub, 32) + srv.hub.register(c) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodDelete, ts.URL+"/api/v1/messages/msg-does-not-exist", nil, tok) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("delete status = %d, want 404 (body=%s)", resp.StatusCode, body) + } + if err := connHasNoFrame(c); err != nil { + t.Fatal(err) + } +} diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index 1d23bc3..2b73642 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -188,6 +188,35 @@ func (h *hub) broadcast(ev wsEvent) { } } +// 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 +// carries SessionID and a client filters it by its active session, exactly as it +// does for message_persisted. +const ( + wsMessageEdited = "message_edited" + 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 +// 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. +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 +// `message_deleted` frame carrying the removed id (plus the owning sessionId for +// client-side session filtering), so every browser of the account drops the row +// without a refresh. sessionID may be empty when the store could not report it; +// clients then fall back to the globally unique id. +func (h *hub) broadcastMessageDeleted(sessionID, id string) { + h.broadcast(wsEvent{Type: wsMessageDeleted, SessionID: sessionID, Data: map[string]any{"id": id}}) +} + // relay is the bus handler for the ws.> pattern and the producer bridge // patterns. The ws.chat subject is a streaming trigger (run the loop and emit // chunks); a ws.* subject becomes an outbound event of its design type; a