From 37e98ec584db01621f04aa2ce0bc8a5887f150c1 Mon Sep 17 00:00:00 2001 From: Aska Date: Mon, 7 Sep 2026 04:51:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20IM=E5=B9=B3=E5=8F=B0=E5=88=97=20+=20dev?= =?UTF-8?q?iceID=E6=A0=87=E8=AF=86=20+=20=E4=BC=9A=E8=AF=9D=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8C=96=E5=AE=8C=E5=96=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - agent-96: chat_messages 加 platform 列(迁移0004) + IM主人私聊落主会话web(带platform) + 前端按平台过滤契约 + message_persisted 带 platform(复用 BroadcastPersisted) - agent-97: chat_messages 加 device_id 列(迁移0005) + history.Message.DeviceID + Web发消息带deviceId(localStorage getClientId) + IM persistToMain 带 deviceID + 前端气泡展示 SourceLabel + 修主会话基础2个bug(fn scanMessage语法/DDL缺列) - design/16 §4.4(平台) + §4.5(deviceID) 已入档 - 实机: POST /chat 带 deviceId 读回 user/assistant 均带 deviceId+platform; 后端 build/vet/test 绿 + 前端 build 绿 - 红线: 身份靠 actor 元数据; 禁止 shell 读 .env(JWT冒烟由主会话跑) --- Docs/api-client.md | 12 +- cmd/cyrene/wire.go | 23 ++- frontend/web/src/components/chat.tsx | 21 ++- frontend/web/src/lib/api.ts | 33 +++- frontend/web/src/lib/client.ts | 46 +++++ frontend/web/src/lib/types.ts | 12 ++ frontend/web/src/lib/useChat.ts | 8 +- frontend/web/src/lib/ws.ts | 2 + frontend/web/src/store/useAppStore.ts | 12 +- .../0004_chat_history_platform.down.sql | 25 +++ .../0004_chat_history_platform.up.sql | 13 ++ .../db/migrations/0005_chat_device.down.sql | 4 + .../db/migrations/0005_chat_device.up.sql | 10 + internal/runtime/history/store.go | 58 ++++-- internal/runtime/history/store_test.go | 141 +++++++++++++- .../services/channelbridge/channelbridge.go | 95 +++++++--- .../channelbridge/channelbridge_test.go | 173 +++++++++++++++++- internal/services/channelbridge/persist.go | 55 ++++++ internal/services/gateway/history.go | 7 +- .../services/gateway/history_handler_test.go | 35 ++++ internal/services/gateway/rest.go | 7 +- internal/services/gateway/server.go | 29 +-- .../services/gateway/session_orchestrator.go | 46 +++-- .../gateway/session_orchestrator_test.go | 34 +++- internal/services/gateway/session_persist.go | 28 +-- internal/services/gateway/ws_hub.go | 15 +- 26 files changed, 828 insertions(+), 116 deletions(-) create mode 100644 frontend/web/src/lib/client.ts create mode 100644 internal/db/migrations/0004_chat_history_platform.down.sql create mode 100644 internal/db/migrations/0004_chat_history_platform.up.sql create mode 100644 internal/db/migrations/0005_chat_device.down.sql create mode 100644 internal/db/migrations/0005_chat_device.up.sql create mode 100644 internal/services/channelbridge/persist.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 2c78bdf..f0429e1 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -21,10 +21,10 @@ |---|------|------|------|:---:|------| | 1 | `/api/v1/auth/login` | POST | 否 | ✅ | `{username,password}` → `{data:{token,jwt}}`（24h，owner） | | 2 | `/api/v1/health` | GET | 否 | ✅ | 健康检查 `{data:{status:"ok"}}` | -| 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,stream?}` → `{data:{reply}}`；`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL，入视觉真推理（design/06）。**会话编排（design/15）**：非流式且该会话「正忙」时**不阻塞**，返回 `202` `{data:{queued:true,queueDepth}}`（稍后按上下文续接）；待处理队列满时返回 `409` `{error:{code:"chat_busy"}}` | +| 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,deviceId?,stream?}` → `{data:{reply}}`；`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL，入视觉真推理（design/06）；`deviceId` 为**客户端生成的具体端标识**（design/16 §4.5，如 `web:chrome:a1b2c3`，可选），落库到消息、便于主人知道从哪个端发来。**会话编排（design/15）**：非流式且该会话「正忙」时**不阻塞**，返回 `202` `{data:{queued:true,queueDepth}}`（稍后按上下文续接）；待处理队列满时返回 `409` `{error:{code:"chat_busy"}}` | | 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}`。**会话持久化（design/16）**：注入 HistoryStore 后返回持久化会话元数据 `{data:[{id,type("main"|"sub"),title?,createdAt}]}`（主会话 `id="web"` 恒在首位，次级 `web:`）；未注入时退化回渠道会话列表 | | 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=` 默认 `limit=50`（上限 200）、`before`=上一页最旧 `createdAt`（毫秒），按时间倒序 `{data:[{id,sessionId,role,content,mode?,mood?,image?,thinking?,createdAt}]}`；非法 `limit`/`before` 回 `400`；未注入 HistoryStore 回 `501` | +| 6 | `/api/v1/sessions/{id}/messages` | GET | JWT(owner/admin) | ✅ | **热分页拉历史**：`?limit=&before=&platform=` 默认 `limit=50`（上限 200）、`before`=上一页最旧 `createdAt`（毫秒），按时间倒序 `{data:[{id,sessionId,role,content,mode?,mood?,image?,thinking?,platform?,deviceId?,createdAt}]}`；`platform` 为**可选**来源平台过滤（design/16 §4.4）——Web 端不传（=全量共享主会话），IM 适配器传**本平台**（如 `telegram`）只拉本渠道消息；`deviceId` 为**客户端具体端标识**（design/16 §4.5，如 `web:chrome:a1b2c3` / `weixin-ilink:conf0:wx_uid`）；非法 `limit`/`before` 回 `400`；未注入 HistoryStore 回 `501` | | 7 | `/api/v1/sessions/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除次级会话及其消息** → `{data:{deleted:}}`；主会话不可删回 `403 forbidden`；未注入 HistoryStore 回 `501` | | 8 | `/api/v1/personas` | GET | JWT | ✅ | `{data:[persona...]}` | @@ -89,7 +89,7 @@ - **握手**: `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` / `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`（定向回请求连接，非广播）。 - > **`history` 回放事件（design/16，新增）**：WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,createdAt}]}}`——回放主会话最新 50 条（`thinking` 已剥离，不泄给非 owner；默认主会话，次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取）。客户端据此在重连/新开连接后补全上下文。 + > **`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` 拉取）。客户端据此在重连/新开连接后补全上下文。 ### 3.1 结构化流事件（`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*`） - **事件壳**：`{type, turnId?, seq?, data}`, `data` 为**类型化结构**（非 `any`）。`seq` 为 hub 级单调递增序号，供客户端将一次突发的流式事件排序关联。 @@ -117,8 +117,8 @@ - **上下文续接**：忙完重开时新回合**带上一轮上下文**（`{mode, queueDepth}` 之外的续接由服务端完成），非清空重来（design/15 §2.2）。 ### 3.3 消息持久化增量事件（`message_persisted`，design/16） -- **触发**：每一条 user / assistant 消息**落库后**由编排层广播（`persistUserMessages` 在提交时写用户消息、`persistAssistantReply` 在回合落定时写助手消息）。 -- **载荷**：`{type:"message_persisted", sessionId, data:{message:{id, sessionId, role, content, mode?, mood?, image?, createdAt}}}`——`role` ∈ `user|assistant`；`id` 为**后端生成的消息 id**（客户端据此去重）；`thinking` 已剥离（design/16 §6）。 +- **触发**：每一条 user / assistant 消息**落库后**由编排层广播——Web 主会话路径走 `persistUserMessages`/`persistAssistantReply`，**IM 渠道主人私聊**（design/16 §4.4）由 channelbridge 把消息**落主会话 `web`**（带 `platform` 标记）后复用**同一** `message_persisted` 广播（同一 hub 的 `BroadcastPersisted`），保证 Web 端实时镜像 IM 消息。 +- **载荷**：`{type:"message_persisted", sessionId, data:{message:{id, sessionId, role, content, mode?, mood?, image?, platform?, deviceId?, createdAt}}}`——`role` ∈ `user|assistant`；`id` 为**后端生成的消息 id**（客户端据此去重）；`platform` 为**来源平台**（`web` 或渠道如 `telegram`/`weixin-ilink`/`qq-obv11`）；`deviceId` 为**具体端/客户端标识**（design/16 §4.5，如 `web:chrome:a1b2c3` / `weixin-ilink:conf0:wx_uid`）；`thinking` 已剥离（design/16 §6）。 - **客户端处理**：仅当 `data.message.sessionId === activeSessionId`（激活会话）时处理；按 `id` upsert——本端已预建的**同 role 乐观气泡**（本地 `uid` 无后端 `id`）被**归一**（`uid` → 后端 `id`），异端无本地气泡则直接追加，两者最终收敛为**同一条 `id` 相同的消息**，「A 发消息、A 与 B 都无需刷新且各出现一次」。 ## 4. 客户端接入指引 @@ -127,6 +127,8 @@ - **离线**: PWA / 原生端建议缓存 `sessions`/`personas` 列表。 ## 5. 变更记录 +- `v0.13`（每条消息带 deviceID/client 标识，design/16 §4.5）：**`chat_messages` 增 `device_id` 列**（迁移 0005，默认 `''`）；`POST /api/v1/chat` 请求体新增**可选 `deviceId`**（客户端生成的端标识，如 `web:chrome:a1b2c3`）；消息 DTO（REST 消息、WS `history` 回放、WS `message_persisted`）均带 **`deviceId`**。`platform`（平台）与 `deviceId`（具体端）互补：同一平台多个端 → 相同 `platform`、不同 `deviceId`。**Web 端**由 `lib/client.ts` 用 `localStorage` 持久化每个浏览器一个的 `web::<随机短id>`，发送时自动带上，聊天气泡默认小字标注来源（仅当 `deviceId` 存在才显示）；**IM 端**按 `:<本地配置唯一id>:<用户平台唯一uid>` 构造（当前 `UnifiedMessage` 尚无 deviceId 概念，channelbridge 以 `::` 占位，`<本地配置唯一id>` 待适配器提供后替换）；Web 端聊天不回流 IM。 +- `v0.12`（IM 渠道并入主会话 + 平台标记，design/16 §4.4）：**`chat_messages` 增 `platform` 列**（迁移 0004，默认 `'web'`）；`GET /api/v1/sessions/{id}/messages` 支持可选 **`?platform=`** 过滤（Web 端不传=全量共享主会话；IM 适配器传本平台只拉本渠道）；消息 DTO（REST 与 WS `message_persisted`）均带 `platform`。**IM 渠道主人私聊**（`!IsGroup && Owner`）经 channelbridge 落**主会话 `web`**（`platform=<渠道>`）并广播 `message_persisted` → Web 端免刷新镜像；**群聊/陌生人**保持独立隔离上下文、不落主会话。Web 端聊天（`platform=web`）不回流 IM。 - `v0.11`（跨端实时同步，design/16 §4.2）：WS 下行新增 **`message_persisted`** 事件——每条 user/assistant 消息落库后广播 `{type, sessionId, data:{message:{id,sessionId,role,content,mode?,mood?,image?,createdAt}}}`（`id` 为后端生成，`thinking` 已剥离）。客户端按 `sessionId === activeSessionId` 过滤 + 按 `id` upsert（含本端乐观气泡 `uid`→后端 `id` 归一）去重，实现「A 发消息、A 与 B 都无需刷新且各出现一次」。后端 `history.Store` 新增 `AppendFull` 供编排层取含 `id` 的落库消息。 - `v0.10`（会话消息持久化 — WS 契约，design/16，后端-2）：WS 下行帧新增可选 `sessionId` 字段（能确定会话的事件带上）；新增 **`history` 回放事件**——WS 握手成功即回放主会话最新 50 条（`{type:"history", sessionId:"web", data:{messages:[...]}}`，`thinking` 已剥离）。`stream_end`/`stream_chunk`/会话编排状态帧（`session_busy` 等）在能确定 session 时带 `sessionId`；`message.delta` 等结构化流帧因 `model.StreamEvent` 无 session 上下文不带（即"若知道就带上"边界）。 - `v0.9`（会话消息持久化 + 主/次级会话，design/16，后端-1）：新增 `POST /api/v1/sessions`（新建次级会话）、`GET /api/v1/sessions/{id}/messages?limit=&before=`（游标热分页拉历史，时间倒序）、`DELETE /api/v1/sessions/{id}`（删次级会话及其消息）；`GET /api/v1/sessions` 在注入 `HistoryStore` 后改返回持久化会话元数据数组（主 + 次级，`type` 区分），未注入时保持原渠道会话列表向后兼容。读消息/写会话为 owner/admin（用户侧私有历史，thinking 不达 viewer）；会话元数据列表保持任意认证角色可读。 diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 47e3937..3844288 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -23,12 +23,12 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader" personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/platform" - "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" @@ -426,6 +426,12 @@ func (a *app) wireAuxiliary() error { } a.sessions = sess + hist, herr := history.NewStore(context.Background(), a.db, a.log) + if herr != nil { + return fmt.Errorf("history store: %w", herr) + } + a.history = hist + automation, err := automations.NewService(automations.Config{}, automations.Deps{ Bus: a.bus, Loop: a.loop, @@ -481,6 +487,7 @@ func (a *app) wireChannels() error { Loop: a.loop, Sessions: a.sessions, Owner: a.owner, + History: a.history, }) if berr != nil { return fmt.Errorf("channel bridge: %w", berr) @@ -512,11 +519,9 @@ func (a *app) wireFrontend() error { a.log.Warn("audit store unavailable, falling back to in-memory", "error", aerr) auditStore = nil } - hist, herr := history.NewStore(context.Background(), a.db, a.log) - if herr != nil { - return fmt.Errorf("history store: %w", herr) - } - a.history = hist + // a.history is built in wireAuxiliary so both the channel bridge (wireChannels) + // and the gateway share the same persistent store. History was a no-op dependency + // before the IM path, so nothing downstream depends on its creation order change. gw, err := gateway.New(a.cfg, gateway.Deps{ Log: logger.WithSource(a.log, "gateway"), Auth: jwt, @@ -543,6 +548,12 @@ func (a *app) wireFrontend() error { return fmt.Errorf("gateway: %w", err) } gw.SetVoice(a.voice) + if a.channelBridge != nil { + // Route the channel bridge's owner-private persisted messages into the same + // `message_persisted` broadcast the orchestrator uses, so the Web client mirrors + // IM messages with one, uniform frame (design/16 §4.4). + a.channelBridge.SetOnPersisted(a.gw.BroadcastPersisted) + } a.gw = gw return nil } diff --git a/frontend/web/src/components/chat.tsx b/frontend/web/src/components/chat.tsx index d2c0f23..f482cfd 100644 --- a/frontend/web/src/components/chat.tsx +++ b/frontend/web/src/components/chat.tsx @@ -116,6 +116,19 @@ function statusLabel(status: ToolStatus): string { } } +/** Small muted source label for a persisted message (design/16 §4.5): shows the + * concrete client the message came from, e.g. "来自 web:chrome:a1b2c3". Rendered + * only when a deviceId is present, so legacy/unlabelled messages stay clean. The + * Web client defaults to this label; other clients decide themselves. */ +function SourceLabel({ deviceId }: { deviceId?: string }): React.ReactNode { + if (!deviceId) return null; + return ( +
+ 来自 {deviceId} +
+ ); +} + /** A single message bubble. Per design/07 §1 the bubble shape is driven by the * message's own `mode`: the same role renders differently under different modes, * and a single session may mix shapes. Falls back to `chat` when mode is absent @@ -188,6 +201,7 @@ function AssistantParaBubble({ {full && message.mood ? (
心情 · {message.mood}
) : null} + {full ? : null} {full ? : null} {full ? : null} {full ? message.tools?.map((t, i) => ) : null} @@ -208,7 +222,7 @@ function UserParaBubble({ full: boolean; }): React.ReactNode { return ( -
+
{full && message.image ? ( {renderContent(text)}
+ {full ? : null}
); } @@ -281,7 +296,7 @@ function UserBubble({ message, mode }: { message: ChatMessage; mode: Mode }): Re {renderContent(message.content)} ) : null; return ( -
+
{message.image ? ( +
); } @@ -316,6 +332,7 @@ function AssistantBubble({ {message.mood ? (
心情 · {message.mood}
) : null} + {message.tools?.map((t, i) => )} diff --git a/frontend/web/src/lib/api.ts b/frontend/web/src/lib/api.ts index 5687cdc..8ca2592 100644 --- a/frontend/web/src/lib/api.ts +++ b/frontend/web/src/lib/api.ts @@ -31,6 +31,7 @@ import type { SessionMeta, SessionRecord, } from "./types"; +import { getClientId } from "./client"; /** * REST client for the Cyrene v2 gateway. The gateway wraps every success in @@ -160,21 +161,24 @@ export function health(): Promise<{ status: string }> { return request<{ status: string }>("/health"); } -/** POST /api/v1/chat — one synchronous agent turn. */ +/** POST /api/v1/chat — one synchronous agent turn. The body carries the sender's + * persistent per-browser deviceId (design/16 §4.5) so the persisted user and + * assistant messages are tagged with the concrete client. */ export function chatSync( sessionId: string, text: string ): Promise { return request("/chat", { method: "POST", - body: JSON.stringify({ sessionId, text }), + body: JSON.stringify({ sessionId, text, deviceId: getClientId() }), }); } /** POST /api/v1/chat with stream:true — triggers streaming over the WS channel. * `image` (a base64 data: URI or URL) is attached for the vision path; the * gateway's chat body currently has no image field, so an unconfigured backend - * simply ignores it. */ + * simply ignores it. `deviceId` mirrors the client's per-browser id onto the + * persisted message (design/16 §4.5). */ export function chatStream( sessionId: string, text: string, @@ -182,7 +186,7 @@ export function chatStream( ): Promise<{ stream: true }> { return request<{ stream: true }>("/chat", { method: "POST", - body: JSON.stringify({ sessionId, text, image, stream: true }), + body: JSON.stringify({ sessionId, text, image, stream: true, deviceId: getClientId() }), }); } @@ -197,6 +201,10 @@ export interface MessageDto { mood?: string; image?: string; thinking?: string; + /** Source-platform tag (design/16 §4.4): "web" or an IM channel platform. */ + platform?: string; + /** Concrete client/device that produced the message (design/16 §4.5). */ + deviceId?: string; createdAt: number; } @@ -211,6 +219,8 @@ function toChatMessage(m: MessageDto): ChatMessage { mood: m.mood, image: m.image, thinking: m.thinking, + platform: m.platform, + deviceId: m.deviceId, createdAt: m.createdAt, }; } @@ -228,18 +238,23 @@ export function createSubSession(title?: string): Promise { }); } -/** GET /api/v1/sessions/{id}/messages?before=&limit= — hot-paginated history. - * The wire returns newest-first, but callers render chronologically (oldest - * first), so we reverse here: page 1 is the most recent `limit` messages in - * display order, and `before` (the oldest message's createdAt) walks upward. */ +/** GET /api/v1/sessions/{id}/messages?before=&limit=&platform= — hot-paginated + * history, optionally filtered to one source platform (design/16 §4.4). The wire + * returns newest-first, but callers render chronologically (oldest first), so we + * reverse here: page 1 is the most recent `limit` messages in display order, and + * `before` (the oldest message's createdAt) walks upward. The Web client omits + * `platform` for the full shared main session; an IM adapter passes its own + * platform to see only its lane. */ export async function fetchMessages( sessionId: string, before?: number, - limit = 50 + limit = 50, + platform?: string ): Promise { const qs = new URLSearchParams(); qs.set("limit", String(limit)); if (before !== undefined) qs.set("before", String(before)); + if (platform) qs.set("platform", platform); const list = await request( `/sessions/${encodeURIComponent(sessionId)}/messages?${qs.toString()}` ); diff --git a/frontend/web/src/lib/client.ts b/frontend/web/src/lib/client.ts new file mode 100644 index 0000000..7e26502 --- /dev/null +++ b/frontend/web/src/lib/client.ts @@ -0,0 +1,46 @@ +// Per-browser persistent client id (design/16 §4.5). Every chat message the Web +// client sends carries this id so the agent knows which browser/device the owner +// is chatting from — e.g. "web:chrome:a1b2c3". It survives reloads and is shared +// by all tabs of the same browser (localStorage), distinguishing multiple browsers +// logged into the same account. + +const CLIENT_ID_KEY = "cyrene.clientId"; + +/** Short, human-readable browser key derived from the navigator user agent, so the + * deviceID reads like "chrome" rather than a raw UA string. */ +function browserKey(): string { + const ua = navigator.userAgent; + if (/edg\//i.test(ua)) return "edge"; + if (/firefox\//i.test(ua)) return "firefox"; + if (/chrome\//i.test(ua)) return "chrome"; + if (/safari\//i.test(ua)) return "safari"; + return "browser"; +} + +/** Eight lowercase hex characters of random entropy (one short id token). */ +function randomToken(): string { + if (typeof crypto !== "undefined" && "getRandomValues" in crypto) { + const bytes = new Uint8Array(4); + crypto.getRandomValues(bytes); + let out = ""; + for (const b of bytes) out += b.toString(16).padStart(2, "0"); + return out; + } + return Math.random().toString(16).slice(2, 10); +} + +/** Persistent, per-browser client id: "web::". Stored in + * localStorage so every tab of the same browser shares one id and it survives + * reloads. Degrades to a fresh id when localStorage is unavailable (privacy mode + * / a page that blocked storage): the message is still tagged, just not stable. */ +export function getClientId(): string { + const existing = localStorage.getItem(CLIENT_ID_KEY); + if (existing && existing.startsWith("web:")) return existing; + const id = `web:${browserKey()}:${randomToken()}`; + try { + localStorage.setItem(CLIENT_ID_KEY, id); + } catch { + // Storage unavailable: fall back to the freshly generated id for this call. + } + return id; +} diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index a5997a7..3a2a04c 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -76,6 +76,18 @@ export interface ChatMessage { * as an image card. Absent for legacy persisted messages, so renders fall back * to plain text. */ image?: string; + /** Source-platform tag of the message (design/16 §4.4): "web" for messages that + * originate on the Web main session, or an IM channel (e.g. "telegram", + * "weixin-ilink", "qq-obv11") for messages mirrored in from an IM adapter. The + * Web client renders the full session; an IM adapter filters to its own platform. + * Absent on legacy/client-optimistic messages. */ + platform?: string; + /** Concrete client/device that produced the message (design/16 §4.5), e.g. + * "web:chrome:a1b2c3" for a browser or "weixin-ilink:conf0:wx_uid" for an IM + * channel. Complements platform (which only names the platform). The Web client + * defaults to labeling the bubble source when this is present; absent on + * legacy/client-optimistic messages kept without it. */ + deviceId?: string; createdAt: number; } diff --git a/frontend/web/src/lib/useChat.ts b/frontend/web/src/lib/useChat.ts index d327961..f32220e 100644 --- a/frontend/web/src/lib/useChat.ts +++ b/frontend/web/src/lib/useChat.ts @@ -3,6 +3,7 @@ import { chatStream } from "./api"; import { connectWs, isWsOpen } from "./ws"; import { useAppStore } from "../store/useAppStore"; import { useModeStore } from "../store/useModeStore"; +import { getClientId } from "./client"; import { now } from "./util"; /** Drives one user turn: push the user bubble, blanket a streaming assistant @@ -38,8 +39,11 @@ export function useSendChat(): { // The gateway's chat body still requires a non-empty text, so a pure image // turn (vision) falls back to a default prompt to keep the trigger valid. const prompt = trimmed || (image ? "分析这张图片" : ""); - addUserMessage(trimmed, undefined, image); - beginAssistant(mode); + // Tag this browser's persistent client id onto the optimistic bubbles so the + // source label shows immediately and matches the persisted turn (design/16 §4.5). + const clientId = getClientId(); + addUserMessage(trimmed, undefined, image, clientId); + beginAssistant(mode, clientId); try { await chatStream(sessionId, prompt, image); diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index 674cefd..bfaf573 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -174,6 +174,8 @@ function parsePersistedMessage(v: unknown): ChatMessage | null { mode: typeof r.mode === "string" && MODES.includes(r.mode as Mode) ? (r.mode as Mode) : undefined, mood: typeof r.mood === "string" ? r.mood : undefined, image: typeof r.image === "string" ? r.image : undefined, + platform: typeof r.platform === "string" && r.platform ? r.platform : undefined, + deviceId: typeof r.deviceId === "string" && r.deviceId ? r.deviceId : undefined, createdAt: typeof r.createdAt === "number" ? r.createdAt : now(), }; } diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index 46e4336..72b761c 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -66,8 +66,8 @@ export interface AppStore { * backend id) so a message never renders twice (design/16 §4.3 dedup). */ appendOrUpdateMessage: (msg: ChatMessage) => void; - addUserMessage: (text: string, mode?: Mode, image?: string) => void; - beginAssistant: (mode?: Mode) => string; + addUserMessage: (text: string, mode?: Mode, image?: string, deviceId?: string) => void; + beginAssistant: (mode?: Mode, deviceId?: string) => string; appendChunk: (delta: string) => void; endStream: () => void; setThinking: (text: string) => void; @@ -172,6 +172,8 @@ export const useAppStore = create()( mode: msg.mode ?? m.mode, mood: msg.mood ?? m.mood, image: msg.image ?? m.image, + platform: msg.platform ?? m.platform, + deviceId: msg.deviceId ?? m.deviceId, createdAt: msg.createdAt ?? m.createdAt, }; return { messages }; @@ -181,7 +183,7 @@ export const useAppStore = create()( return { messages: [...s.messages, msg] }; }), - addUserMessage: (text, mode, image) => + addUserMessage: (text, mode, image, deviceId) => set((s) => ({ messages: [ ...s.messages, @@ -191,12 +193,13 @@ export const useAppStore = create()( content: text, mode, image, + deviceId, createdAt: now(), }, ], })), - beginAssistant: (mode) => { + beginAssistant: (mode, deviceId) => { const id = uid(); set((s) => ({ messages: [ @@ -208,6 +211,7 @@ export const useAppStore = create()( mode, streaming: true, tools: [], + deviceId, createdAt: now(), }, ], diff --git a/internal/db/migrations/0004_chat_history_platform.down.sql b/internal/db/migrations/0004_chat_history_platform.down.sql new file mode 100644 index 0000000..f8091cc --- /dev/null +++ b/internal/db/migrations/0004_chat_history_platform.down.sql @@ -0,0 +1,25 @@ +-- 0004_chat_history_platform: rolls back the platform column added by the up +-- migration. SQLite's ALTER DROP COLUMN is version-dependent (3.35+), so the column +-- is removed by recreating the table without it — universally safe and lossless +-- (the platform tag is not needed once the column is gone). + +CREATE TABLE chat_messages_new ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + mode TEXT, + mood TEXT, + image TEXT, + thinking TEXT, + created_at INTEGER NOT NULL +); + +INSERT INTO chat_messages_new (id, session_id, role, content, mode, mood, image, thinking, created_at) + SELECT id, session_id, role, content, mode, mood, image, thinking, created_at FROM chat_messages; + +DROP TABLE chat_messages; +ALTER TABLE chat_messages_new RENAME TO chat_messages; + +CREATE INDEX IF NOT EXISTS idx_chat_msgs_session_created + ON chat_messages (session_id, created_at DESC); diff --git a/internal/db/migrations/0004_chat_history_platform.up.sql b/internal/db/migrations/0004_chat_history_platform.up.sql new file mode 100644 index 0000000..b34452b --- /dev/null +++ b/internal/db/migrations/0004_chat_history_platform.up.sql @@ -0,0 +1,13 @@ +-- 0004_chat_history_platform: tag every chat message with its source platform +-- (DevDocs/design/16 §4.4). The Web main session is shared across platforms: the +-- Web client renders the full session, while each IM adapter filters to its own +-- platform lane. Messages that originate on the Web carry 'web'; IM owner-private +-- messages carry the channel platform (e.g. 'weixin-ilink', 'qq-obv11', 'telegram'). +-- +-- The 0003 chat_messages table already exists at this point, so ALTER ADD COLUMN is +-- the correct, non-destructive step. The canonical schema (including this column) is +-- also re-issued idempotently by internal/runtime/history.NewStore so a fresh +-- in-memory store used by tests stays self-contained; the migration is the +-- authoritative path for an existing database. + +ALTER TABLE chat_messages ADD COLUMN platform TEXT NOT NULL DEFAULT 'web'; diff --git a/internal/db/migrations/0005_chat_device.down.sql b/internal/db/migrations/0005_chat_device.down.sql new file mode 100644 index 0000000..bd0870f --- /dev/null +++ b/internal/db/migrations/0005_chat_device.down.sql @@ -0,0 +1,4 @@ +-- 0005_chat_device down: drop the device_id column. +-- SQLite DROP COLUMN is supported on modern versions; rebuild safely when needed. + +ALTER TABLE chat_messages DROP COLUMN device_id; diff --git a/internal/db/migrations/0005_chat_device.up.sql b/internal/db/migrations/0005_chat_device.up.sql new file mode 100644 index 0000000..c0073e1 --- /dev/null +++ b/internal/db/migrations/0005_chat_device.up.sql @@ -0,0 +1,10 @@ +-- 0005_chat_device: tag every chat message with a short deviceID/client identifier +-- (DevDocs/design/16 §4.5). deviceID is the concrete client (browser/app/IM channel +-- binding) that produced the message, complementing `platform` which only names +-- the platform. It lets the agent know exactly HOW the owner is chatting (e.g. +-- "web:chrome:a1b2c3" for a browser, "weixin-ilink:conf0:wx_uid_123" for IM). +-- +-- The 0004 chat_messages table already exists at this point, so ALTER ADD COLUMN is +-- the correct, non-destructive step. + +ALTER TABLE chat_messages ADD COLUMN device_id TEXT NOT NULL DEFAULT ''; diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 14b1477..479aebf 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -37,6 +37,11 @@ const ( // SessionMain and SessionSub classify a persistent session. SessionMain = "main" SessionSub = "sub" + + // PlatformWeb is the source-platform tag for messages that originate on the Web + // main session (design/16 §4.4). It is both the column default and the value + // assigned when a caller leaves Message.Platform empty. + PlatformWeb = "web" ) // MainSessionID is the fixed id of the one main session (design/16 §2.1). Sub session @@ -71,6 +76,17 @@ type Message struct { Mood string Image string Thinking string + // Platform tags the source platform of a message (design/16 §4.4). The Web main + // session is shared across platforms: messages that originate on the Web carry + // "web", while IM-owner-private messages carry the channel platform (e.g. + // "telegram", "qq-obv11", "weixin-ilink"). An empty Platform defaults to "web" so + // legacy/explicitly-Web writes stay on the Web lane without a separate assignment. + Platform string + // DeviceID is a short, client-produced identifier for the concrete device/app + // that sent the message (design/16 §4.5) — e.g. "web:chrome:a1b2c3" for a + // browser, "weixin-ilink:conf0:wx_uid_123" for IM. It complements Platform: two + // clients on the same platform share Platform but differ in DeviceID. + DeviceID string CreatedAt int64 } @@ -92,8 +108,12 @@ type HistoryStore interface { AppendFull(ctx context.Context, msg Message) (Message, error) // List returns up to limit messages for sessionID, newest first. before is a // millisecond timestamp cursor: when nonzero, only messages older than before are - // returned (inclusive-free, matching the "before" bound on the frontend). - List(ctx context.Context, sessionID string, before int64, limit int) ([]Message, error) + // returned (inclusive-free, matching the "before" bound on the frontend). platform + // is an optional source-platform filter (design/16 §4.4): when non-empty only + // messages carrying that platform are returned; when empty all platforms are + // returned (the Web client requests the full main session, an IM adapter asks for + // its own platform only). + List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error) // Recent returns the most recent n messages for sessionID in oldest-first order, // for WS replay / context tail usage. Recent(ctx context.Context, sessionID string, n int) ([]Message, error) @@ -175,11 +195,14 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { if msg.CreatedAt == 0 { msg.CreatedAt = time.Now().UnixMilli() } + if msg.Platform == "" { + msg.Platform = PlatformWeb + } _, err := s.db.ExecContext(ctx, `INSERT INTO chat_messages - (id, session_id, role, content, mode, mood, image, thinking, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + (id, session_id, role, content, mode, mood, image, thinking, platform, device_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, - msg.Image, msg.Thinking, msg.CreatedAt) + msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.CreatedAt) if err != nil { return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) } @@ -188,17 +211,24 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { // List returns up to limit messages for sessionID, newest first. When before is nonzero // it bounds the query to messages strictly older than before, so a client can page upward -// by passing the oldest message's created_at from the previous page. The returned slice -// is non-nil. limit is clamped to [1, maxPageLimit]. -func (s *Store) List(ctx context.Context, sessionID string, before int64, limit int) ([]Message, error) { +// by passing the oldest message's created_at from the previous page. When platform is +// non-empty it further restricts the page to messages carrying that source platform +// (design/16 §4.4: the Web client omits platform for the full main session, an IM +// adapter passes its own platform to see only its lane). The returned slice is non-nil. +// limit is clamped to [1, maxPageLimit]. +func (s *Store) List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error) { limit = clampLimit(limit) - query := `SELECT id, session_id, role, content, mode, mood, image, thinking, created_at + query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, created_at FROM chat_messages WHERE session_id = ?` args := []any{sessionID} if before > 0 { query += ` AND created_at < ?` args = append(args, before) } + if platform != "" { + query += ` AND platform = ?` + args = append(args, platform) + } query += ` ORDER BY created_at DESC LIMIT ?` args = append(args, limit) @@ -228,7 +258,7 @@ func (s *Store) Recent(ctx context.Context, sessionID string, n int) ([]Message, if n < 1 { n = defaultPageLimit } - rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, created_at + rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, created_at FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC LIMIT ?`, sessionID, n) if err != nil { return nil, fmt.Errorf("history: recent %s: %w", sessionID, err) @@ -342,11 +372,11 @@ func (s *Store) ensureMainSession(ctx context.Context) error { // scanMessage reads one chat_messages row. message columns are scanned positionally. func scanMessage(sc rowScanner) (Message, error) { var m Message - var mode, mood, image, thinking string - if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &image, &thinking, &m.CreatedAt); err != nil { + var mode, mood, image, thinking, platform, deviceID string + if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &image, &thinking, &platform, &deviceID, &m.CreatedAt); err != nil { return Message{}, err } - m.Mode, m.Mood, m.Image, m.Thinking = mode, mood, image, thinking + m.Mode, m.Mood, m.Image, m.Thinking, m.Platform, m.DeviceID = mode, mood, image, thinking, platform, deviceID return m, nil } @@ -401,6 +431,8 @@ CREATE TABLE IF NOT EXISTS chat_messages ( mood TEXT, image TEXT, thinking TEXT, + platform TEXT NOT NULL DEFAULT 'web', + device_id TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL )` diff --git a/internal/runtime/history/store_test.go b/internal/runtime/history/store_test.go index 0046bc7..d809cb9 100644 --- a/internal/runtime/history/store_test.go +++ b/internal/runtime/history/store_test.go @@ -51,7 +51,7 @@ func TestAppendAndListNewestFirst(t *testing.T) { } } - msgs, err := st.List(ctx, MainSessionID, 0, 50) + msgs, err := st.List(ctx, MainSessionID, 0, 50, "") if err != nil { t.Fatalf("list: %v", err) } @@ -67,7 +67,7 @@ func TestAppendAndListNewestFirst(t *testing.T) { if err := st.Append(ctx, empty); err != nil { t.Fatalf("append no ts: %v", err) } - msgs, err = st.List(ctx, MainSessionID, 0, 50) + msgs, err = st.List(ctx, MainSessionID, 0, 50, "") if err != nil { t.Fatalf("list after append: %v", err) } @@ -96,7 +96,7 @@ func TestAppendFullReturnsID(t *testing.T) { } // The message is actually persisted under the returned id. - msgs, err := st.List(ctx, MainSessionID, 0, 50) + msgs, err := st.List(ctx, MainSessionID, 0, 50, "") if err != nil { t.Fatalf("list: %v", err) } @@ -114,6 +114,133 @@ func TestAppendFullReturnsID(t *testing.T) { } } +func TestAppendFullListPlatformRoundTrip(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + // An IM owner-private message carries its channel platform through the round + // trip: AppendFull → List. + full, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "从微信来", Platform: "weixin-ilink"}) + if err != nil { + t.Fatalf("append full: %v", err) + } + if full.Platform != "weixin-ilink" { + t.Fatalf("AppendFull platform = %q, want weixin-ilink", full.Platform) + } + msgs, err := st.List(ctx, MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 || msgs[0].Platform != "weixin-ilink" || msgs[0].Content != "从微信来" { + t.Fatalf("list round trip = %+v, want platform weixin-ilink", msgs) + } + + // A message without an explicit platform defaults to the web lane. + web, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleAssistant, Content: "ok"}) + if err != nil { + t.Fatalf("append web default: %v", err) + } + if web.Platform != PlatformWeb { + t.Fatalf("default platform = %q, want web", web.Platform) + } +} + +func TestAppendFullListDeviceIDRoundTrip(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + // A Web message tagged with a browser deviceID survives the AppendFull → List + // round trip (design/16 §4.5). + full, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "来自浏览器", DeviceID: "web:chrome:a1b2c3"}) + if err != nil { + t.Fatalf("append full: %v", err) + } + if full.DeviceID != "web:chrome:a1b2c3" { + t.Fatalf("AppendFull DeviceID = %q, want web:chrome:a1b2c3", full.DeviceID) + } + msgs, err := st.List(ctx, MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 || msgs[0].DeviceID != "web:chrome:a1b2c3" || msgs[0].Content != "来自浏览器" { + t.Fatalf("list round trip = %+v, want DeviceID web:chrome:a1b2c3", msgs) + } + + // Recent (WS replay / context tail) reads the same device_id column. + recent, err := st.Recent(ctx, MainSessionID, 5) + if err != nil { + t.Fatalf("recent: %v", err) + } + if len(recent) != 1 || recent[0].DeviceID != "web:chrome:a1b2c3" { + t.Fatalf("recent = %+v, want DeviceID web:chrome:a1b2c3", recent) + } + + // A message without an explicit deviceID persists the column default (''). + om, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleAssistant, Content: "ok"}) + if err != nil { + t.Fatalf("append empty device: %v", err) + } + if om.DeviceID != "" { + t.Fatalf("empty DeviceID = %q, want empty", om.DeviceID) + } +} + +func TestListPlatformFilter(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + // Seed across the web lane (explicit + default) and one IM lane. + for _, m := range []Message{ + {SessionID: MainSessionID, Role: RoleUser, Content: "web1", Platform: PlatformWeb, CreatedAt: 1000}, + {SessionID: MainSessionID, Role: RoleUser, Content: "tg", Platform: "telegram", CreatedAt: 2000}, + {SessionID: MainSessionID, Role: RoleUser, Content: "web2", CreatedAt: 3000}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + + // No platform filter → all three (the empty-platform message defaulted to web). + all, err := st.List(ctx, MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list all: %v", err) + } + if len(all) != 3 { + t.Fatalf("all = %d, want 3", len(all)) + } + + // Filter to the IM lane → only its message. + tg, err := st.List(ctx, MainSessionID, 0, 50, "telegram") + if err != nil { + t.Fatalf("list telegram: %v", err) + } + if len(tg) != 1 || tg[0].Content != "tg" { + t.Fatalf("telegram page = %+v, want [tg]", tg) + } + + // Filter to the web lane → web1 + web2 (the one left without an explicit + // platform defaulted to web on insert). + web, err := st.List(ctx, MainSessionID, 0, 50, PlatformWeb) + if err != nil { + t.Fatalf("list web: %v", err) + } + if len(web) != 2 { + t.Fatalf("web page = %d, want 2 (%+v)", len(web), web) + } + + // The filter composes with the before cursor. + older, err := st.List(ctx, MainSessionID, 2000, 50, PlatformWeb) + if err != nil { + t.Fatalf("list web before: %v", err) + } + if len(older) != 1 || older[0].Content != "web1" { + t.Fatalf("web before page = %+v, want [web1]", older) + } +} + func TestListPaginationBefore(t *testing.T) { t.Parallel() st := newStore(t) @@ -126,7 +253,7 @@ func TestListPaginationBefore(t *testing.T) { } // First page: newest 2. - page, err := st.List(ctx, MainSessionID, 0, 2) + page, err := st.List(ctx, MainSessionID, 0, 2, "") if err != nil { t.Fatalf("first page: %v", err) } @@ -134,7 +261,7 @@ func TestListPaginationBefore(t *testing.T) { t.Fatalf("first page = %+v, want [6000 5000]", page) } // Cursor on the page's oldest (5000) → older page [4000 3000]. - next, err := st.List(ctx, MainSessionID, page[1].CreatedAt, 2) + next, err := st.List(ctx, MainSessionID, page[1].CreatedAt, 2, "") if err != nil { t.Fatalf("next page: %v", err) } @@ -142,7 +269,7 @@ func TestListPaginationBefore(t *testing.T) { t.Fatalf("next page = %+v, want [4000 3000]", next) } // Past the start → empty. - tail, err := st.List(ctx, MainSessionID, 1000, 50) + tail, err := st.List(ctx, MainSessionID, 1000, 50, "") if err != nil { t.Fatalf("tail: %v", err) } @@ -220,7 +347,7 @@ func TestDeleteSession(t *testing.T) { if err := st.DeleteSession(ctx, sub); err != nil { t.Fatalf("delete sub: %v", err) } - msgs, err := st.List(ctx, sub, 0, 50) + msgs, err := st.List(ctx, sub, 0, 50, "") if err != nil { t.Fatalf("list after delete: %v", err) } diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 34daff5..98eb475 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -24,6 +24,7 @@ import ( "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/logger" + "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/user" @@ -72,21 +73,30 @@ type SessionStore interface { // Options carries the bridge's injected dependencies. Loop and Sessions are // required; Owner may be nil (then no account is recognized as owner, the safe -// default-deny posture from design/11 §0). +// default-deny posture from design/11 §0). History and OnPersisted are optional +// (design/16 §4.4): when a history store is wired, owner-private messages are +// routed into the main web session and tagged with the channel platform, and every +// persisted message is handed to OnPersisted so the gateway can broadcast a +// `message_persisted` event; when both are absent the bridge runs exactly as +// before (isolated sess-xxx, no chat_messages writes). type Options struct { - Log *slog.Logger - Loop Chatter - Sessions SessionStore - Owner *user.OwnerProfile + Log *slog.Logger + Loop Chatter + Sessions SessionStore + Owner *user.OwnerProfile + History history.HistoryStore + OnPersisted func(history.Message) } // Bridge owns the channel adapter lifecycle and the inbound message pipeline. It // is safe for concurrent use. type Bridge struct { - log *slog.Logger - run Chatter - sessions SessionStore - owner *user.OwnerProfile + log *slog.Logger + run Chatter + sessions SessionStore + owner *user.OwnerProfile + history history.HistoryStore + onPersisted func(history.Message) mu sync.RWMutex adapters map[string]adapter.PlatformAdapter @@ -114,14 +124,16 @@ func New(opts Options) (*Bridge, error) { log = logger.New(logger.Options{Out: nil}) } return &Bridge{ - log: logger.WithSource(log, "services/channelbridge"), - run: opts.Loop, - sessions: opts.Sessions, - owner: opts.Owner, - adapters: make(map[string]adapter.PlatformAdapter), - connected: make(map[string]bool), - cancel: func() {}, - sem: make(chan struct{}, maxConcurrent), + log: logger.WithSource(log, "services/channelbridge"), + run: opts.Loop, + sessions: opts.Sessions, + owner: opts.Owner, + history: opts.History, + onPersisted: opts.OnPersisted, + adapters: make(map[string]adapter.PlatformAdapter), + connected: make(map[string]bool), + cancel: func() {}, + sem: make(chan struct{}, maxConcurrent), }, nil } @@ -353,28 +365,52 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) { "session", sess.SessionID, "actor_role", sess.ActorRole, "env", sess.Env(), "mem_writable", sess.MemWritable) + // design/16 §4.4 + design/02 §3: an owner private chat is routed INTO the single + // main web session (shared across platforms) and persisted to chat_messages with + // the channel platform, so the Web client mirrors it; group chats and strangers + // keep their isolated sess-xxx context and are NOT persisted to the main session. + sessID := sess.SessionID + isOwnerPrivate := !sess.IsGroup && sess.Owner + platform := "" + if isOwnerPrivate { + sessID = history.MainSessionID + platform = msg.Platform + } + + // Persist the inbound user message into the main web session before the turn, so + // the Web client sees it even if the channel reply fails (design/16 §4.4). + if isOwnerPrivate { + b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg)) + } + // Thread the resolved isolation boundary into the loop's memory gates: the // owner's private conversation reads+writes memory, group chats and strangers // are read-only isolation (design/02 §3, design/11 §2). runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable} - res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sess.SessionID, []model.Message{ + res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, []model.Message{ model.TextMessage(model.RoleUser, text), }) if err != nil { b.log.Warn("agent run failed", "platform", msg.Platform, - "session", sess.SessionID, "error", err) + "session", sessID, "error", err) return } reply := strings.TrimSpace(res.Text) if reply == "" { - b.log.Debug("no reply produced", "platform", msg.Platform, "session", sess.SessionID) + b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID) return } + // Persist the assistant reply into the main web session (design/16 §4.4) and + // broadcast, then deliver it back to the channel — so the Web client mirrors the + // IM reply too. + if isOwnerPrivate { + b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg)) + } if err := b.sendReply(ctx, msg, reply); err != nil { if errors.Is(err, adapter.ErrUnsupported) { b.log.Debug("channel does not support outbound reply, skipped", - "platform", msg.Platform, "session", sess.SessionID) + "platform", msg.Platform, "session", sessID) return } b.log.Warn("send reply failed", "platform", msg.Platform, "error", err) @@ -422,3 +458,20 @@ func cfgStr(m map[string]any, key string) string { } return "" } + +// deviceIDFor builds the client deviceID tag for an inbound IM message +// (design/16 §4.5): :<本地配置唯一id>:<用户平台唯一uid>. UnifiedMessage has no +// explicit deviceId field yet, so the local config id and the user platform uid are +// taken from the normalized channel binding: ChannelID stands in for the local +// adapter binding and SenderID is the owner's platform uid. When SenderID is absent +// the tag degrades to : so it stays non-empty and readable. +// +// Extension point: when an adapter/UnifiedMessage grows a concrete deviceId (or the +// bridge gains a local config id), replace the ChannelID stand-in here and return +// that value directly. +func deviceIDFor(msg adapter.UnifiedMessage) string { + if msg.SenderID != "" { + return msg.Platform + ":" + msg.ChannelID + ":" + msg.SenderID + } + return msg.Platform + ":" + msg.ChannelID +} diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index 538cd02..13ca338 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -2,10 +2,12 @@ package channelbridge import ( "context" + "sync" "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/user" @@ -113,8 +115,9 @@ func TestProcessRunsPipelineAndReplies(t *testing.T) { if sess.msg.SenderID != "100" { t.Errorf("resolved sender = %q, want 100", sess.msg.SenderID) } - if chat.sessionID != "sess-1" { - t.Errorf("loop session id = %q, want sess-1", chat.sessionID) + // An owner private chat routes into the single main web session (design/16 §4.4). + if chat.sessionID != history.MainSessionID { + t.Errorf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID) } if len(chat.messages) != 1 || chat.messages[0].Role != model.RoleUser { t.Fatalf("loop messages = %+v, want single user message", chat.messages) @@ -160,8 +163,8 @@ func TestProcessLoopErrorDoesNotSend(t *testing.T) { b.process(context.Background(), directMessage()) - if chat.sessionID != "sess-1" { - t.Errorf("loop should still run for valid text, session = %q", chat.sessionID) + if chat.sessionID != history.MainSessionID { + t.Errorf("loop should still run for valid text on the main session, session = %q", chat.sessionID) } if len(ad.sends) != 0 { t.Errorf("send count = %d, want 0 on loop error", len(ad.sends)) @@ -207,3 +210,165 @@ func TestProcessEmptyReplySendsNothing(t *testing.T) { t.Errorf("send count = %d, want 0 for empty reply", len(ad.sends)) } } + +// captureHistory records every Append for asserting the bridge writes owner-private +// messages into the main web session (design/16 §4.4). +type captureHistory struct { + mu sync.Mutex + appends []history.Message +} + +func (c *captureHistory) Append(_ context.Context, msg history.Message) error { + _, err := c.AppendFull(context.Background(), msg) + return err +} + +func (c *captureHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.appends = append(c.appends, msg) + return msg, nil +} + +func (c *captureHistory) List(context.Context, string, int64, int, string) ([]history.Message, error) { + return nil, nil +} +func (c *captureHistory) Recent(context.Context, string, int) ([]history.Message, error) { + return nil, nil +} +func (c *captureHistory) DeleteSession(context.Context, string) error { return nil } +func (c *captureHistory) NewSubSession(context.Context, string) (string, error) { + return "", nil +} +func (c *captureHistory) ListSessions(context.Context) ([]history.SessionMeta, error) { + return nil, nil +} + +func (c *captureHistory) records() []history.Message { + c.mu.Lock() + defer c.mu.Unlock() + return append([]history.Message(nil), c.appends...) +} + +func TestOwnerPrivatePersistsToMainSessionWithPlatform(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "微信回复"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + hist := &captureHistory{} + var persisted []history.Message + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + History: hist, + OnPersisted: func(m history.Message) { persisted = append(persisted, m) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + msg := directMessage() // Platform = "telegram" + b.process(context.Background(), msg) + + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + // Both land on the main web session and carry the channel platform. + if recs[0].SessionID != history.MainSessionID || recs[0].Role != history.RoleUser || + recs[0].Content != "hello" || recs[0].Platform != "telegram" { + t.Fatalf("user record = %+v, want web/telegram/hello", recs[0]) + } + if recs[1].SessionID != history.MainSessionID || recs[1].Role != history.RoleAssistant || + recs[1].Content != "微信回复" || recs[1].Platform != "telegram" { + t.Fatalf("assistant record = %+v, want web/telegram/微信回复", recs[1]) + } + // Every persisted message was handed to the broadcast hook. + if len(persisted) != 2 { + t.Fatalf("persisted notifications = %d, want 2", len(persisted)) + } + // And the loop ran against the main session, not the isolated sess-1. + if chat.sessionID != history.MainSessionID { + t.Fatalf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID) + } +} + +func TestOwnerPrivatePersistsToMainWithDeviceID(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "微信回复"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + History: hist, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // directMessage() → Platform telegram / ChannelID 100 / SenderID 100, so the + // deviceID is built as :: (design/16 §4.5). + b.process(context.Background(), directMessage()) + + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + // Both the user turn and the assistant reply carry the derived deviceID and the + // channel platform, so the Web client can label the source end. + for i, rec := range recs { + if rec.DeviceID != "telegram:100:100" { + t.Fatalf("record[%d] DeviceID = %q, want telegram:100:100", i, rec.DeviceID) + } + if rec.Platform != "telegram" { + t.Fatalf("record[%d] Platform = %q, want telegram", i, rec.Platform) + } + } + if recs[0].Role != history.RoleUser || recs[0].Content != "hello" { + t.Fatalf("user record = %+v", recs[0]) + } + if recs[1].Role != history.RoleAssistant || recs[1].Content != "微信回复" { + t.Fatalf("assistant record = %+v", recs[1]) + } +} + +func TestGroupMessageNotPersistedToMainSession(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "群聊回复"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: true, MemWritable: false}} + hist := &captureHistory{} + b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist}) + if err != nil { + t.Fatalf("New: %v", err) + } + + msg := adapter.UnifiedMessage{ + Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup, + GroupID: "-20", GroupName: "Gang", SenderID: "9", Content: "hello all", + ContentType: adapter.ContentText, + } + b.process(context.Background(), msg) + + // Group chat stays isolated: nothing is written to the main web session and the + // loop runs against the isolated session id. + if recs := hist.records(); len(recs) != 0 { + t.Fatalf("group appends = %+v, want none persisted", recs) + } + if chat.sessionID != "sess-g" { + t.Fatalf("group loop session id = %q, want sess-g", chat.sessionID) + } +} + +func TestNoHistoryStoreKeepsBridgeUnchanged(t *testing.T) { + t.Parallel() + // A bridge built without a history store persists nothing but still runs/replies. + b, chat, _, ad := newTestBridge(t) + b.process(context.Background(), directMessage()) + if len(ad.sends) != 1 { + t.Fatalf("send count = %d, want 1", len(ad.sends)) + } + if chat.sessionID != history.MainSessionID { + t.Fatalf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID) + } +} diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go new file mode 100644 index 0000000..22e54e6 --- /dev/null +++ b/internal/services/channelbridge/persist.go @@ -0,0 +1,55 @@ +package channelbridge + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// This file holds the channelbridge's chat-history persistence hooks (design/16 +// §4.4): an owner private chat is routed INTO the single main web session and +// tagged with the channel platform, then broadcast so the Web client mirrors it +// without a refresh. Kept separate from channelbridge.go so the bridge's inbound +// pipeline file stays under the module size bound. + +// SetOnPersisted wires (or replaces) the transport hook called after an +// owner-private message is persisted into the main web session. The gateway is +// built after the bridge in the composition root, so the hook is attached post-New +// through this setter; it is a no-op-safe call and is safe for concurrent use with +// in-flight processing. +func (b *Bridge) SetOnPersisted(fn func(history.Message)) { + b.mu.Lock() + defer b.mu.Unlock() + b.onPersisted = fn +} + +// notifyPersisted hands a just-persisted message to the transport hook when one is +// wired, so the gateway can broadcast `message_persisted` for cross-end sync. +func (b *Bridge) notifyPersisted(msg history.Message) { + b.mu.RLock() + fn := b.onPersisted + b.mu.RUnlock() + if fn != nil { + fn(msg) + } +} + +// persistToMain writes one message into the main web session with an explicit source +// platform and client deviceID, then notifies the transport hook so Web clients receive +// a `message_persisted` event without a refresh (design/16 §4.4). deviceID is the +// concrete client that sent the message (design/16 §4.5) — e.g. "weixin-ilink:conf0:wx_uid" +// for IM. It is a no-op when no history store is wired, so the bridge runs unchanged +// without storage. +func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, deviceID string) { + if b.history == nil { + return + } + full, err := b.history.AppendFull(ctx, history.Message{ + SessionID: history.MainSessionID, Role: role, Content: content, Platform: platform, DeviceID: deviceID, + }) + if err != nil { + b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID) + return + } + b.notifyPersisted(full) +} diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index 3eb3cf8..03a7e91 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -32,6 +32,8 @@ type historyMessage struct { Mood string `json:"mood,omitempty"` Image string `json:"image,omitempty"` Thinking string `json:"thinking,omitempty"` + Platform string `json:"platform,omitempty"` + DeviceID string `json:"deviceId,omitempty"` CreatedAt int64 `json:"createdAt"` } @@ -108,8 +110,9 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) { } before = v } + platform := r.URL.Query().Get("platform") - msgs, err := s.history.List(r.Context(), id, before, limit) + msgs, err := s.history.List(r.Context(), id, before, limit, platform) if err != nil { s.log.Error("history list messages", "err", err, "session", id) writeError(w, apperr.Wrap(apperr.ErrInternal, "messages_failed", "failed to list messages")) @@ -157,6 +160,8 @@ func toHistoryMessage(m history.Message) historyMessage { Mood: m.Mood, Image: m.Image, Thinking: m.Thinking, + Platform: m.Platform, + DeviceID: m.DeviceID, CreatedAt: m.CreatedAt, } } diff --git a/internal/services/gateway/history_handler_test.go b/internal/services/gateway/history_handler_test.go index 64a4531..677d3f5 100644 --- a/internal/services/gateway/history_handler_test.go +++ b/internal/services/gateway/history_handler_test.go @@ -133,6 +133,41 @@ func TestHistoryListMessagesPagination(t *testing.T) { } } +func TestHistoryListMessagesPlatformFilter(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "web", Platform: history.PlatformWeb, CreatedAt: 1000}); err != nil { + t.Fatalf("seed web: %v", err) + } + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "tg", Platform: "telegram", CreatedAt: 2000}); err != nil { + t.Fatalf("seed tg: %v", err) + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // No platform → the full main session (both lanes). + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?limit=10", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("all status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var all []historyMessage + decodeDataSlice(t, body, &all) + if len(all) != 2 { + t.Fatalf("all = %d, want 2", len(all)) + } + + // ?platform=telegram → only the telegram lane, and the DTO carries platform. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?limit=10&platform=telegram", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("platform status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var tg []historyMessage + decodeDataSlice(t, body, &tg) + if len(tg) != 1 || tg[0].Content != "tg" || tg[0].Platform != "telegram" { + t.Fatalf("telegram page = %+v, want one platform=telegram message", tg) + } +} + func TestHistoryDeleteSubSession(t *testing.T) { t.Parallel() _, ts, jwt, st := newHistoryEnv(t) diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index 05657a0..f3532c5 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -70,10 +70,13 @@ type loginResponse struct { // triggers streaming over the WebSocket channel and returns immediately. Image is // an optional base64 data: URI or http(s) URL attached for the vision path; the // loop routes it through the vision/OCR tool or the main model (design/06 §0). +// DeviceID is the client-produced identifier of the sending browser/device +// (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted as-is. type chatRequest struct { SessionID string `json:"sessionId"` Text string `json:"text"` Image string `json:"image,omitempty"` + DeviceID string `json:"deviceId,omitempty"` Stream bool `json:"stream"` } @@ -126,7 +129,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { return } if req.Stream { - s.publishChatStream(r.Context(), req.SessionID, req.Text, req.Image) + s.publishChatStream(r.Context(), req.SessionID, req.Text, req.Image, req.DeviceID) writeJSON(w, http.StatusOK, map[string]any{"stream": true}) return } @@ -141,7 +144,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { // it binds the full-access env: memory read+write, unchanged behavior. msgs := userChatMessages(req.Text, req.Image) if s.hub.orchestrator != nil { - sr, res := s.hub.orchestrator.SubmitSync(r.Context(), req.SessionID, msgs, loop.OwnerEnv()) + sr, res := s.hub.orchestrator.SubmitSync(withDeviceID(r.Context(), req.DeviceID), req.SessionID, msgs, loop.OwnerEnv()) if !sr.Accepted { if sr.PendingFull { writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙，待处理消息已满")) diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 5a2ba3d..046bb23 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -113,7 +113,7 @@ type HistoryStore interface { // (with the assigned id/timestamp) so the orchestrator can broadcast it for // cross-end realtime sync (design/16 §4.2). AppendFull(ctx context.Context, msg history.Message) (history.Message, error) - List(ctx context.Context, sessionID string, before int64, limit int) ([]history.Message, error) + List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]history.Message, error) Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) DeleteSession(ctx context.Context, sessionID string) error NewSubSession(ctx context.Context, ownerID string) (string, error) @@ -459,20 +459,25 @@ func newOrchestrator(s *Server) *Orchestrator { Emit: func(sessionID, typ string, data map[string]any) { s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, Data: data}) }, - OnPersisted: func(msg history.Message) { - // Broadcast the complete message so every browser of the same account - // appends it without a manual refresh (design/16 §4.2). The frontend - // dedups by the backend-generated id, and thinking is stripped to stay - // consistent with the WS history replay (design/16 §6). - wire := toHistoryMessage(msg) - wire.Thinking = "" - s.hub.broadcast(wsEvent{Type: "message_persisted", SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) - }, - Log: s.log, - History: s.history, + OnPersisted: s.BroadcastPersisted, + Log: s.log, + History: s.history, }) } +// BroadcastPersisted broadcasts a just-persisted chat message as a +// `message_persisted` WS event so every browser of the same account appends it +// without a manual refresh (design/16 §4.2). It is shared by the orchestrator +// (Web turns) and the channel bridge (IM owner-private messages routed into the +// main web session), so both paths emit the identical frame. Thinking is stripped +// to stay consistent with the WS history replay (design/16 §6). It is a no-op-safe +// call when the hub has broadcast wired. +func (s *Server) BroadcastPersisted(msg history.Message) { + wire := toHistoryMessage(msg) + wire.Thinking = "" + s.hub.broadcast(wsEvent{Type: "message_persisted", SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) +} + // sourcedLog returns a logger tagged with source "gateway", substituting a // discard logger for nil so Server can always log safely. func sourcedLog(l *slog.Logger) *slog.Logger { diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index e145832..7a73f53 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -128,10 +128,13 @@ type sessionState struct { } // pendingMsg is one queued user turn. It carries its own env so a resume runs -// under the same memory-isolation boundary the message arrived with. +// under the same memory-isolation boundary the message arrived with, plus the +// client deviceID of the originating turn (design/16 §4.5) so the resumed +// assistant reply is tagged with the same client that sent the message. type pendingMsg struct { messages []model.Message env loop.SessionEnv + deviceID string at time.Time } @@ -215,6 +218,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes st = &sessionState{mode: ModeIdle} o.states[sessionID] = st } + deviceID := deviceIDFrom(ctx) if st.mode != ModeIdle { if len(st.pending) >= o.maxPending { mode := st.mode @@ -222,11 +226,11 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.emitState(sessionID, EventPendingFull, map[string]any{"sessionId": sessionID, "mode": mode, "max": o.maxPending}) return SubmitResult{PendingFull: true}, nil } - st.pending = append(st.pending, pendingMsg{messages: messages, env: env, at: time.Now()}) + st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, at: time.Now()}) depth := len(st.pending) mode := st.mode o.mu.Unlock() - o.persistUserMessages(ctx, sessionID, messages) + o.persistUserMessages(ctx, sessionID, messages, deviceID) o.emitState(sessionID, EventPendingAck, map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth}) return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth}, nil } @@ -240,30 +244,32 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.mu.Unlock() o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking}) - o.persistUserMessages(ctx, sessionID, messages) + o.persistUserMessages(ctx, sessionID, messages, deviceID) if sync { res, err := o.run(runCtx, sessionID, messages, env) if err != nil { o.log.Error("orchestrator turn", "err", err, "session", sessionID) } - _ = o.completeTurn(sessionID, gen, res, err) + _ = o.completeTurn(sessionID, gen, res, err, deviceID) return SubmitResult{Accepted: true, Immediate: true, Err: err}, res } - go o.goTurn(sessionID, gen, messages, env, runCtx, cancel) + go o.goTurn(sessionID, gen, messages, env, runCtx, cancel, deviceID) return SubmitResult{Accepted: true, Immediate: true}, nil } // goTurn runs a single turn on a background goroutine and, when it finishes, // hands the outcome to completeTurn so the orchestrator can resume the queue. -func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc) { +// deviceID tags the persistent assistant reply with the client that sent the turn +// (design/16 §4.5). +func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID string) { defer cancel() res, err := o.run(ctx, sessionID, messages, env) if err != nil && ctx.Err() == nil { o.log.Error("orchestrator turn", "err", err, "session", sessionID) } - _ = o.completeTurn(sessionID, gen, res, err) + _ = o.completeTurn(sessionID, gen, res, err, deviceID) } // completeTurn runs after a turn settles (either it finished, errored, or was @@ -272,8 +278,9 @@ func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Mes // message "remembers" the prior context instead of starting from a blank slate // (design/15 §2.2: incremental, not full replay). A stale generation (an // interrupt or a newer top-level turn superseded this one) is a no-op. -func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error) error { - o.persistAssistantReply(sessionID, res) +// deviceID is the originating turn's client id, used to tag the persisted reply. +func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID string) error { + o.persistAssistantReply(sessionID, res, deviceID) o.mu.Lock() st := o.states[sessionID] if st == nil || st.gen != gen { @@ -298,7 +305,7 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu o.mu.Unlock() o.emitState(sessionID, EventPendingDrain, map[string]any{"sessionId": sessionID, "mode": ModeThinking, "queueDepth": depth}) - go o.goTurn(sessionID, gen, resume, env, runCtx, cancel) + go o.goTurn(sessionID, gen, resume, env, runCtx, cancel, next.deviceID) return nil } @@ -430,6 +437,23 @@ func (o *Orchestrator) notifyPersisted(msg history.Message) { } } +// deviceIDContextKey is the context key carrying a message's client deviceID +// (design/16 §4.5) from the HTTP/WS boundary into the orchestrator's persistence. +type deviceIDContextKey struct{} + +// withDeviceID returns a context carrying deviceID so the orchestrator can tag the +// persisted user message. An empty deviceID is a no-op (no tag is written). +func withDeviceID(ctx context.Context, deviceID string) context.Context { + return context.WithValue(ctx, deviceIDContextKey{}, deviceID) +} + +// deviceIDFrom returns the deviceID carried by ctx, or "" when none was set. It +// returns only well-formed non-empty values; anything else is treated as absent. +func deviceIDFrom(ctx context.Context) string { + v, _ := ctx.Value(deviceIDContextKey{}).(string) + return v +} + // boundedTail returns a fresh copy of the last maxTail messages, so continuation // stays incremental and bounded (design/15 §2.2, §4) and never aliases the input // slice or another state's tail. diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index e94d74f..5a197d8 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -175,7 +175,7 @@ func (f *fakeHistory) AppendFull(_ context.Context, msg history.Message) (histor return msg, nil } -func (f *fakeHistory) List(context.Context, string, int64, int) ([]history.Message, error) { +func (f *fakeHistory) List(context.Context, string, int64, int, string) ([]history.Message, error) { return nil, nil } @@ -230,6 +230,38 @@ func TestSubmitPersistsUserAndAssistant(t *testing.T) { } } +func TestSubmitPersistsDeviceID(t *testing.T) { + t.Parallel() + run := newGatedRunner(1) + hist := &fakeHistory{} + o := NewOrchestrator(OrchestratorOptions{ + Run: run.Run, + Log: slog.Default(), + History: hist, + }) + run.tickets <- struct{}{} + // The HTTP/WS boundary threads the browser deviceID through ctx (design/16 §4.5); + // both the user turn and the assistant reply must be tagged with it. + ctx := withDeviceID(context.Background(), "web:chrome:a1b2c3") + sr, res := o.SubmitSync(ctx, "web", []model.Message{userMsg("你好")}, loop.OwnerEnv()) + if !sr.Accepted || !sr.Immediate { + t.Fatalf("submit = %+v, want Accepted+Immediate", sr) + } + if res == nil { + t.Fatal("sync result is nil") + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + if recs[0].Role != history.RoleUser || recs[0].Content != "你好" || recs[0].DeviceID != "web:chrome:a1b2c3" { + t.Fatalf("user record = %+v, want DeviceID web:chrome:a1b2c3", recs[0]) + } + if recs[1].Role != history.RoleAssistant || recs[1].Content == "" || recs[1].DeviceID != "web:chrome:a1b2c3" { + t.Fatalf("assistant record = %+v, want DeviceID web:chrome:a1b2c3", recs[1]) + } +} + func TestSubmitPersistedBroadcast(t *testing.T) { t.Parallel() run := newGatedRunner(1) diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go index a451c66..df5879f 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -16,10 +16,11 @@ import ( // persistUserMessages writes each RoleUser message of a submitted turn to the // history store so the raw user stream is durable before the turn resolves, then // notifies the transport so every browser of the same account can append it -// without a manual refresh (design/16 §3.3, §4.2). It is a no-op when no history -// store is wired. Rejected turns (empty, pending-full) never reach it, so nothing -// vacuous is persisted. -func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message) { +// without a manual refresh (design/16 §3.3, §4.2). deviceID tags the message with +// the sending client (design/16 §4.5). It is a no-op when no history store is +// wired. Rejected turns (empty, pending-full) never reach it, so nothing vacuous +// is persisted. +func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID string) { if o.history == nil { return } @@ -31,9 +32,9 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string if text == "" && image == "" { continue } - full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image}) + full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID}) if err != nil { - o.log.Error("orchestrator persist user", "err", err, "session", sessionID) + o.log.Error("orchestrator persist user", "err", err, "session", sessionID, "device_id", deviceID) continue } o.notifyPersisted(full) @@ -41,17 +42,18 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string } // persistAssistantReply writes the final assistant text of a settled turn to the -// history store, then notifies the transport (design/16 §3.3, §4.2). It is a -// no-op when no history store is wired or the turn produced no text -// (interrupted/errored turns leave res nil). The write runs under a fresh -// background context so a canceled turn context never aborts it. -func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result) { +// history store, then notifies the transport (design/16 §3.3, §4.2). deviceID tags +// the reply with the client that sent the turn (design/16 §4.5). It is a no-op +// when no history store is wired or the turn produced no text (interrupted/errored +// turns leave res nil). The write runs under a fresh background context so a +// canceled turn context never aborts it. +func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID string) { if o.history == nil || res == nil || res.Text == "" { return } - full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text}) + full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, DeviceID: deviceID}) if err != nil { - o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID) + o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID, "device_id", deviceID) return } o.notifyPersisted(full) diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index f30c517..cc3c442 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -235,8 +235,11 @@ func (h *hub) streamChat(ctx context.Context, payload any) { // The Web main session is the owner's private conversation (JWT = owner), so // it binds the full-access env: memory read+write, unchanged behavior. msgs := userChatMessages(req.Text, req.Image) + // Thread the client's deviceID into the orchestrator via ctx so both the + // persisted user turn and the assistant reply are tagged (design/16 §4.5). + ctx = withDeviceID(ctx, req.DeviceID) if h.orchestrator != nil { - _ = h.orchestrator.Submit(context.Background(), req.SessionID, msgs, loop.OwnerEnv()) + _ = h.orchestrator.Submit(ctx, req.SessionID, msgs, loop.OwnerEnv()) return } _, _ = h.runTurn(ctx, req.SessionID, msgs, loop.OwnerEnv()) @@ -305,6 +308,10 @@ type chatStreamRequest struct { SessionID string `json:"sessionId"` Text string `json:"text"` Image string `json:"image,omitempty"` + // DeviceID is the client-produced identifier of the sending browser/device + // (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted + // alongside the message so the agent knows which end the owner is chatting from. + DeviceID string `json:"deviceId,omitempty"` } // handleWS upgrades /ws/chat after authenticating the JWT from the query token or @@ -432,11 +439,13 @@ func (c *wsConn) enqueueJSON(ev wsEvent) { } // publishChatStream publishes a ws.chat streaming trigger (REST stream mode). -func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image string) { +// deviceID is the sending browser/device id carried on the payload so the ws.chat +// consumer can tag the persisted turn (design/16 §4.5). +func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID string) { if s.bus == nil { return } - _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image}) + _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image, "deviceId": deviceID}) } // wsToken extracts the JWT from the ?token= query or the Authorization header.