From 1a85de555d5d642675b89bd3db178b8ace8b69c4 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Wed, 30 Sep 2026 08:51:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(emotion):=20A1=20=E6=83=85=E7=BB=AA?= =?UTF-8?q?=E7=B3=BB=E7=BB=9F=E6=8E=A5=E7=BA=BF=20+=20refactor:=20?= =?UTF-8?q?=E6=8B=86=E5=88=86=20channelbridge.go?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A1 情绪接线: - loop 回合边界钩子(TurnEnd/OnTurnEnd/settleTurn, Result.Mood); 每轮一次非每步 - persona/emotion: React(valence五档) + Machine.Move(BFS最短合法路径一步) + Observer(observe/decay/好感度,仅变化才emit); 好感度步长读角色配置(cyrene delta=10, 0..700) - 落库 Mood(web session_persist.res.Mood + IM channelbridge/persist; user消息不带mood); WS mood事件(BroadcastMood,仅变化); per-persona隔离; 群聊/陌生人只读不写不发 - cmd/cyrene/emotion.go 生产接线 + wire.go 注入 loop.Options.OnTurnEnd - 22 测试(状态机被loop调用/Mood非空落库/mood帧shape/只变化才发/群聊只读) + 真二进制冒烟 - Docs: api-client v0.32 §3.7 情绪事件 + 开源对照清单(companion 应用对照) - 诚实: 本机无LLM凭据→未活体验证'模型返回后完整轮次'一环 拆分 channelbridge.go 816→257(纯搬移零逻辑): connect/inbound/outbound/blacklist/helpers.go; A1 的 IM mood 调用点落在新 inbound.go, 故与本提交同批 回滚 6 个 gofmt 顺手格式化文件(capacity_loop_test/flow/source_test/findchannel_test/admin_channels_platform_test/history) 缩 diff, 含 history.go 注释被转弯引号 --- Docs/api-client.md | 12 +- cmd/cyrene/emotion.go | 93 +++ cmd/cyrene/emotion_test.go | 249 ++++++++ cmd/cyrene/wire.go | 16 + internal/persona/emotion/observer.go | 98 +++ internal/persona/emotion/observer_test.go | 165 ++++++ internal/persona/emotion/react.go | 104 ++++ internal/persona/emotion/react_test.go | 105 ++++ internal/persona/emotion/state.go | 57 ++ internal/persona/loader/persona.go | 11 + internal/persona/registry/registry.go | 10 + internal/runtime/loop/loop.go | 58 +- internal/runtime/loop/options.go | 7 + internal/runtime/loop/run.go | 7 + internal/runtime/loop/turn.go | 62 ++ internal/runtime/loop/turn_test.go | 160 +++++ internal/services/channelbridge/blacklist.go | 72 +++ .../services/channelbridge/channelbridge.go | 559 ------------------ internal/services/channelbridge/connect.go | 173 ++++++ internal/services/channelbridge/helpers.go | 62 ++ internal/services/channelbridge/inbound.go | 282 +++++++++ internal/services/channelbridge/outbound.go | 29 + internal/services/channelbridge/persist.go | 11 +- internal/services/channelbridge/proactive.go | 2 +- internal/services/gateway/mood_test.go | 103 ++++ internal/services/gateway/server.go | 24 + internal/services/gateway/session_persist.go | 9 +- 27 files changed, 1946 insertions(+), 594 deletions(-) create mode 100644 cmd/cyrene/emotion.go create mode 100644 cmd/cyrene/emotion_test.go create mode 100644 internal/persona/emotion/observer.go create mode 100644 internal/persona/emotion/observer_test.go create mode 100644 internal/persona/emotion/react.go create mode 100644 internal/persona/emotion/react_test.go create mode 100644 internal/runtime/loop/turn.go create mode 100644 internal/runtime/loop/turn_test.go create mode 100644 internal/services/channelbridge/blacklist.go create mode 100644 internal/services/channelbridge/connect.go create mode 100644 internal/services/channelbridge/helpers.go create mode 100644 internal/services/channelbridge/inbound.go create mode 100644 internal/services/channelbridge/outbound.go create mode 100644 internal/services/gateway/mood_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 2299ec0..d941b23 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -129,7 +129,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` / `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）。**会话编排状态（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，主人私聊为空）。 @@ -187,6 +187,15 @@ - **红线**：owner-only；payload 已脱敏；backthink 每次主动推送**只发一个决策平台**（决策渠道→最近活跃→旋转回退），Web 端与 IM 桥为两条投递路，客户端不要据此断言「所有平台都发了」。 - **推送出口（D1，对客户端无新端点）**：在 WS/IM 投递之外，后端会把主动消息**异步镜像**到「推送出口」——主通道是**主人自建统一推送服务**（`PUSH_EXPORTER_URL` webhook，HTTP POST JSON `{source,title,content,platform,timestamp}`，可带 `Authorization: Bearer `），回退是**个推 Getui**（字段预留、SDK 未接入）。该出口只影响**出站投递**，不改变 WS 事件契约；未配/不可达时只记日志（注明「个推回退待接入」），**不影响**上面的 WS/IM 投递。移动端接入自建推送服务即可收到主人主动消息。 +### 3.7 情绪事件（`mood`，design/01 §2 情绪状态机 + 好感度） +- **触发**：每轮对话结束（loop 回合边界，每轮一次、**非**每步）把该轮用户消息的情绪信号应用到**该角色自己的**情绪状态机（按 persona 隔离），且**状态真的变化时**才广播一次 `mood`；状态不变（含「一直 calm」）**不发**，避免每轮噪声。 +- **载荷**：`{type:"mood", data:{mood, personaId, affection}}`——`mood` ∈ `calm|happy|excited|sad|angry`（情绪状态机 state id，与消息体 `mood` 同源同值）；`personaId` 为该角色 id（情绪/好感度跟随角色，不跟随行为模式）；`affection` 为该角色当前好感度（受角色配置 `affection.min/max` 约束）。角色级数值，**不带** `sessionId`。 +- **与消息体 `mood` 的关系**：assistant 消息体的 `mood` 是该轮结束后的**当前情绪**（每轮都写，见 §3.3），而 `mood` 事件只在**变化时**发——前者答「这条回复时她什么心情」，后者答「她的心情刚刚变了」。 +- **群聊/陌生人（`MemWritable=false`）**：情绪**只读不写**——第三方消息**不**改变角色情绪与好感度、**不**发 `mood` 事件、该轮 assistant 消息 `mood` 为空（与记忆只读同一隔离边界，design/02 §3 / design/11 §2）。 +- **状态转移与衰减**：按角色配置的转移表（缺省内置表）与 `emotion.decayRate`——有情绪信号的轮次**沿合法转移走一步**指向对应状态；无信号轮次**衰减一步**趋向 `calm`；`decayRate ≤ 0` 关闭衰减。好感度按角色 `affection.delta` 随情绪轮次增减（warm `+delta` / cold `−delta`，clamped 到 `[min,max]`）。 +- **客户端处理**：`mood` → `setMood(data.mood)`（`src/lib/ws.ts`），顶栏/侧栏显示「心情 · 」；消息体内 `mood` 由聊天气泡显示（`chat.tsx`）。 +- **状态持久性**：情绪与好感度为**进程内、按角色**状态，**重启回到角色配置的 `initialState`**（design/01 未要求落库；如需跨重启保留，待主人确认后加表）。 + ## 4. 客户端接入指引（Web / 移动 / 原生 / IM bridge 通用） ### 4.1 一次性接入流程（三步走） @@ -222,6 +231,7 @@ - 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。 ## 5. 变更记录 +- `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）。 - `v0.30`（**Android 协同批次 ③a：D2 消息搜索 / D3 消息编辑删除 / D5 会话置顶**）：① **D2**：`GET /api/v1/sessions/{id}/messages` 新增可选 **`q`**——**会话内**内容搜索（大小写不敏感 ASCII LIKE，`%`/`_`/`\` 转义为字面量），仍倒序 + `before`/`limit` 游标；`q` 与 `after` 互斥（同传 `400`），空 `q` 走普通分页；**严格限定单会话，不新增跨会话搜索**（避免群聊/陌生人隔离会话外泄，红线）。`HistoryStore` 新增 `Search`。② **D3**：新增 `PATCH /api/v1/messages/{id}`（body `{content}`，非空 ≤4000 rune → `{data:{message}}`，只改 content）与 `DELETE /api/v1/messages/{id}`（→ `{data:{deleted:id}}`）；不存在回 `404 not_found`；`HistoryStore` 新增 `UpdateMessage`/`DeleteMessage` + `ErrMessageNotFound`。**不新增重发端点**（客户端用编辑后的 content 再发）。③ **D5**：`chat_sessions` 新增 `pinned INTEGER NOT NULL DEFAULT 0`（迁移 **0009**，down 可逆）；`GET /sessions` 排序改为 **`pinned DESC` → 主会话优先 → `createdAt ASC`**，会话 DTO 新增恒返回的 **`pinned`** bool；`PATCH /sessions/{id}` 扩展为 body `{title?,pinned?}`（至少一个字段，只回显变更字段，title-only 保持 A4 形状）；`HistoryStore` 新增 `SetPinned`。全部为 owner/admin 写、读消息 owner/admin；未注入 HistoryStore 回 `501`。 - `v0.29`（**/admin/llm 读-改-写合并**）：`POST /api/v1/admin/llm` 由「全量替换」改为**部分编辑合并**——请求体字段用指针区分「未给」与「显式零值」；`maxTokens` / `apiKey` / `provider` / `baseUrl` / `model` / `fallback` / `multimodal` **省略即保留原值**（不再把省略的 `maxTokens` 重置为 0），新组仍要求 `baseUrl`+`model`、非法 provider / 负 maxTokens 仍 `400`。面板 LLM 页新增每组 `maxTokens` 展示/编辑 + 新增/编辑表单；面板插件页新增「已装插件列表」+「安装 MCP 插件」表单（kind/name/transport/endpoint|command+args/apiKey/env），api.ts 请求体带 `config`（修掉旧 body 400）。`runtimeconfig/models.go` 新增 `Find`。 diff --git a/cmd/cyrene/emotion.go b/cmd/cyrene/emotion.go new file mode 100644 index 0000000..053fd11 --- /dev/null +++ b/cmd/cyrene/emotion.go @@ -0,0 +1,93 @@ +package main + +import ( + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/emotion" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" +) + +// This file wires the persona emotion state machine (design/01 §2: 情绪状态机 + +// 好感度曲线 follow the persona/character, never a behavior mode) into the agent +// loop's turn boundary. The loop stays emotion-agnostic: it reports each settled +// turn and carries the returned mood label onto Result.Mood, which the +// persistence layer writes onto the assistant message (design/16 §3.3). +// +// Scope of this milestone (A1) — what is wired end to end: +// - every settled owner turn advances the bound persona's emotion machine and +// affection (in-memory, persona-keyed by the registry's emotion Manager); +// - the resulting mood rides the assistant message into chat_messages and the +// `message_persisted` frame, and each *change* is broadcast as a `mood` frame. +// +// Not wired (design/01 does not specify it): the mood is not yet injected back +// into the system prompt, so it does not steer the reply text, and the emotion +// state is not persisted across restarts (design/01 requires neither; the state +// resets to the persona's configured initialState on boot). + +// moodEmitter is the gateway-facing sink for `mood` broadcasts. *gateway.Server +// implements it (BroadcastMood); held as an interface so the composition root +// stays unit-testable with a fake sink, mirroring capacityEmitter. +type moodEmitter interface { + BroadcastMood(personaID, mood string, affection int) +} + +// buildEmotionObserver assembles the persona emotion observer: the registry's +// emotion Manager supplies each persona's machine and affection, the registry +// supplies the persona's configured affection step (好感度曲线), and every change +// is pushed to the Web UI through publishMood. +func (a *app) buildEmotionObserver() *emotion.Observer { + return emotion.NewObserver(a.personas.Emotions(), a.personas.AffectionStep, a.publishMood) +} + +// onTurnEnd is the loop's turn-boundary hook (design/01 §2): a settled turn is +// applied to the persona the loop is bound to, and the mood after the turn is +// returned for the assistant message. An absent persona id falls back to the +// default persona, matching how the loop is bound. +// +// Two gates keep this to real conversation turns: +// - only the user-facing conversation lane (the Web main session and its +// sub-sessions; IM owner-private chat rides the main session, design/16 §4.4) +// moves the mood. Autonomous and delegated loop calls — background thinking +// (sess-backthink), sub-agent work, isolated IM sessions — are not turns of +// the owner's conversation, so they must not rewrite her mood; +// - a restricted (group/stranger) conversation is read-only: the observer +// refuses the write, so a third party can never move her mood or affection +// (design/02 §3, design/11 §2). +func (a *app) onTurnEnd(end loop.TurnEnd) string { + if a.emotion == nil || !isConversationLane(end.SessionID) { + return "" + } + personaID := end.PersonaID + if personaID == "" { + personaID = a.defaultPersonaID() + } + return string(a.emotion.Observe(personaID, end.UserText, !end.Env.Restricted())) +} + +// isConversationLane reports whether a session belongs to the owner's live +// conversation: the fixed main session "web" and its "web:" sub-sessions +// (design/16 §2.1 — IM owner-private chat is routed into the main session). +func isConversationLane(sessionID string) bool { + return sessionID == history.MainSessionID || strings.HasPrefix(sessionID, history.MainSessionID+":") +} + +// publishMood broadcasts one emotion change as a `mood` frame (design/07 §3, +// design/01 §2) so the Web 心情 badge follows the persona without a refresh. It is +// called only when the mood actually changed, so the frame stays news rather than +// per-turn noise. The sink is read at call time because the gateway is built after +// the loop; before it exists the change is logged and dropped. +func (a *app) publishMood(update emotion.Update) { + log := logger.WithSource(a.log, "cmd/cyrene") + if a.moodSink == nil { + log.Debug("mood change dropped: gateway not ready", + "persona", update.PersonaID, "from", string(update.Before), "to", string(update.After), + "affection", update.Affection, "signal", update.Signal) + return + } + log.Info("persona mood changed", + "persona", update.PersonaID, "from", string(update.Before), "to", string(update.After), + "affection", update.Affection, "signal", update.Signal) + a.moodSink.BroadcastMood(update.PersonaID, string(update.After), update.Affection) +} diff --git a/cmd/cyrene/emotion_test.go b/cmd/cyrene/emotion_test.go new file mode 100644 index 0000000..881d8c4 --- /dev/null +++ b/cmd/cyrene/emotion_test.go @@ -0,0 +1,249 @@ +package main + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/emotion" + "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/runtime/agent" + "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" +) + +// fakeMoodEmitter captures the `mood` broadcasts the composition root forwards, +// so tests can assert an emotion change actually reaches the Web UI. +type fakeMoodEmitter struct { + calls int + personaID string + mood string + affection int +} + +func (f *fakeMoodEmitter) BroadcastMood(personaID, mood string, affection int) { + f.calls++ + f.personaID = personaID + f.mood = mood + f.affection = affection +} + +// moodAdapter is a model.Adapter returning fixed text, so a test can drive a real +// loop turn without a provider. +type moodAdapter struct{ reply string } + +func (m moodAdapter) Name() string { return "mood-test" } + +func (m moodAdapter) Chat(context.Context, model.Request, model.Options) (model.Response, error) { + return model.Response{Content: m.reply}, nil +} + +func (m moodAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return m.Chat(ctx, req, model.Options{}) +} + +// testEmotionApp builds an app with one registered persona (emotion calm, decay +// enabled, affection 50/700 step 10) and the emotion observer wired exactly as +// the composition root wires it, plus a capturing mood sink. +func testEmotionApp(t *testing.T) (*app, *fakeMoodEmitter) { + t.Helper() + reg := personareg.New() + persona := &loader.Persona{ + ID: "cyrene", + Name: "昔涟", + Emotion: loader.EmotionParams{ + InitialState: emotion.Calm, + DecayRate: 0.5, + }, + Affection: loader.AffectionParams{Initial: 50, Min: 0, Max: 700, Delta: 10}, + } + if err := reg.RegisterPersona(persona); err != nil { + t.Fatalf("RegisterPersona: %v", err) + } + if err := reg.RegisterCharacter(&personareg.Character{ID: "cyrene", Name: "昔涟", Default: true, PersonaID: "cyrene"}); err != nil { + t.Fatalf("RegisterCharacter: %v", err) + } + + sink := &fakeMoodEmitter{} + a := &app{log: testLog(), personas: reg, moodSink: sink} + a.emotion = a.buildEmotionObserver() + return a, sink +} + +// testMoodLoop builds a real loop whose turn boundary is the app's hook. +func testMoodLoop(t *testing.T, a *app, reply string) *loop.Loop { + t.Helper() + ag, err := agent.New("你是昔涟。", nil, nil, agent.WithModel(moodAdapter{reply: reply}, nil)) + if err != nil { + t.Fatalf("agent.New: %v", err) + } + return loop.New(loop.Options{Agent: ag, PersonaID: "cyrene", OnTurnEnd: a.onTurnEnd}) +} + +// TestOnTurnEndDrivesEmotionThroughLoop is the A1 end-to-end lock: a real loop +// turn is applied to the bound persona's emotion machine, the mood lands on +// Result.Mood (which the persistence layer writes onto the assistant message), +// affection steps by the persona's configured magnitude, and the change is +// broadcast once as a `mood` frame. +func TestOnTurnEndDrivesEmotionThroughLoop(t *testing.T) { + a, sink := testEmotionApp(t) + l := testMoodLoop(t, a, "人家也开心♪") + + res, err := l.Run(loop.WithSessionEnv(context.Background(), loop.OwnerEnv()), "web", + []model.Message{model.TextMessage(model.RoleUser, "今天谢谢你呀，好开心")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != string(emotion.Happy) { + t.Fatalf("Result.Mood = %q, want happy", res.Mood) + } + machine, affection, ok := a.personas.Emotions().Get("cyrene") + if !ok { + t.Fatal("persona cyrene not registered in the emotion manager") + } + if machine.Current() != emotion.Happy { + t.Fatalf("machine state = %s, want happy", machine.Current()) + } + if affection.Value() != 60 { + t.Fatalf("affection = %d, want 60 (50 + the persona's step 10)", affection.Value()) + } + if sink.calls != 1 || sink.mood != string(emotion.Happy) || sink.personaID != "cyrene" || sink.affection != 60 { + t.Fatalf("mood broadcast = %+v, want one happy/cyrene/60 frame", sink) + } +} + +// TestOnTurnEndEvolvesAndDecaysAcrossTurns locks how the emotion walks over a +// conversation: each emotional turn moves one legal edge toward the state the cue +// aims at (so a strong warm cue reaches excited on the second turn, not the +// first), a cue-free turn lets it decay one step toward calm (design/01 §2 +// decayRate), and each real move is broadcast once while an unchanged mood +// broadcasts nothing. +func TestOnTurnEndEvolvesAndDecaysAcrossTurns(t *testing.T) { + a, sink := testEmotionApp(t) + l := testMoodLoop(t, a, "嗯") + ctx := loop.WithSessionEnv(context.Background(), loop.OwnerEnv()) + + res, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "我太开心了，最喜欢你了")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != string(emotion.Happy) { + t.Fatalf("first strong warm turn mood = %q, want the first legal step happy", res.Mood) + } + // A second strong warm turn escalates to excited. + res, err = l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "最喜欢你了")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != string(emotion.Excited) { + t.Fatalf("second strong warm turn mood = %q, want excited", res.Mood) + } + if sink.calls != 2 || sink.mood != string(emotion.Excited) { + t.Fatalf("broadcasts = %d/%s, want 2 with excited", sink.calls, sink.mood) + } + // Same mood again: no change, so no third frame. + res, err = l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "还是最喜欢你")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != string(emotion.Excited) { + t.Fatalf("third warm turn mood = %q, want excited unchanged", res.Mood) + } + if sink.calls != 2 { + t.Fatalf("broadcasts = %d, want 2 (unchanged mood is not news)", sink.calls) + } + // A cue-free turn carries no signal, so the emotion decays excited → happy. + res, err = l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "帮我看下这个报错")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != string(emotion.Happy) { + t.Fatalf("neutral turn mood = %q, want decayed happy", res.Mood) + } + if sink.calls != 3 || sink.mood != string(emotion.Happy) { + t.Fatalf("broadcasts = %d/%s, want 3 with happy", sink.calls, sink.mood) + } +} + +// TestOnTurnEndRestrictedConversationIsReadOnly locks the isolation boundary at +// the wiring level: a group/stranger turn never moves the persona's mood or +// affection and never broadcasts, matching the read-only memory lane (design/02 +// §3, design/11 §2). +func TestOnTurnEndRestrictedConversationIsReadOnly(t *testing.T) { + a, sink := testEmotionApp(t) + l := testMoodLoop(t, a, "你好呀") + restricted := loop.SessionEnv{MemWritable: false} + ctx := loop.WithSessionEnv(context.Background(), restricted) + + res, err := l.Run(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "我好喜欢你")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != "" { + t.Fatalf("Result.Mood = %q, want empty for a read-only conversation", res.Mood) + } + machine, affection, _ := a.personas.Emotions().Get("cyrene") + if machine.Current() != emotion.Calm || affection.Value() != 50 { + t.Fatalf("restricted turn moved the persona to %s/%d, want calm/50", machine.Current(), affection.Value()) + } + if sink.calls != 0 { + t.Fatalf("broadcasts = %d, want 0 for a restricted conversation", sink.calls) + } +} + +// TestOnTurnEndFallsBackToDefaultPersona locks the persona binding: a loop bound +// without an explicit persona id advances the library's default persona instead +// of silently tracking nothing. +func TestOnTurnEndFallsBackToDefaultPersona(t *testing.T) { + a, sink := testEmotionApp(t) + + if got := a.onTurnEnd(loop.TurnEnd{SessionID: "web", UserText: "谢谢你", Env: loop.OwnerEnv()}); got != string(emotion.Happy) { + t.Fatalf("mood = %q, want happy on the default persona", got) + } + if sink.personaID != "cyrene" { + t.Fatalf("broadcast persona = %q, want cyrene", sink.personaID) + } +} + +// TestOnTurnEndSkipsNonConversationLanes locks the lane gate: only the owner's +// conversation (main session and its sub-sessions) moves the persona's mood. +// Autonomous / delegated loop calls — background thinking, sub-agent work, +// isolated IM sessions — are not conversation turns and must leave her state +// alone. +func TestOnTurnEndSkipsNonConversationLanes(t *testing.T) { + a, sink := testEmotionApp(t) + + for _, sessionID := range []string{"sess-backthink", "sess-group-abc", "task-42", ""} { + if got := a.onTurnEnd(loop.TurnEnd{SessionID: sessionID, UserText: "我好喜欢你", Env: loop.OwnerEnv()}); got != "" { + t.Fatalf("mood for session %q = %q, want empty", sessionID, got) + } + } + machine, affection, _ := a.personas.Emotions().Get("cyrene") + if machine.Current() != emotion.Calm || affection.Value() != 50 { + t.Fatalf("non-conversation lane moved the persona to %s/%d, want calm/50", machine.Current(), affection.Value()) + } + if sink.calls != 0 { + t.Fatalf("broadcasts = %d, want 0", sink.calls) + } + // The main session and its sub-sessions are the conversation lane. + for _, sessionID := range []string{history.MainSessionID, history.MainSessionID + ":abc"} { + if got := a.onTurnEnd(loop.TurnEnd{SessionID: sessionID, UserText: "谢谢你", Env: loop.OwnerEnv()}); got != string(emotion.Happy) { + t.Fatalf("mood for session %q = %q, want happy", sessionID, got) + } + } +} + +// TestOnTurnEndNilDependenciesNoPanic locks the degraded boots: no observer and +// no gateway sink must not panic (the gateway is built after the loop). +func TestOnTurnEndNilDependenciesNoPanic(t *testing.T) { + empty := &app{log: testLog()} + if got := empty.onTurnEnd(loop.TurnEnd{UserText: "我爱你", Env: loop.OwnerEnv()}); got != "" { + t.Fatalf("mood without an observer = %q, want empty", got) + } + empty.publishMood(emotion.Update{PersonaID: "cyrene", Before: emotion.Calm, After: emotion.Happy}) + + a, _ := testEmotionApp(t) + a.moodSink = nil + a.publishMood(emotion.Update{PersonaID: "cyrene", Before: emotion.Calm, After: emotion.Happy}) +} diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 8a738e8..26721a8 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -23,6 +23,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/db" "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/emotion" "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" @@ -108,6 +109,14 @@ type app struct { // §5). It is satisfied by *gateway.Server (BroadcastCapacityAwaken); held as an // interface so the composition root is unit-testable with a fake sink. capacitySink capacityEmitter + // emotion applies each settled turn to the bound persona's emotion machine and + // affection (design/01 §2) and hands the changes to moodSink. Nil when no + // persona was registered, in which case no mood is tracked or persisted. + emotion *emotion.Observer + // moodSink broadcasts a mood change to the Web UI (design/07 §3). It is + // satisfied by *gateway.Server (BroadcastMood); held as an interface so the + // composition root is unit-testable with a fake sink. + moodSink moodEmitter } // capacityEmitter is the gateway-facing sink for capacity_awaken broadcasts. @@ -389,6 +398,11 @@ func (a *app) wireRuntime() error { ) capDecider := a.buildCapacity() + // The persona emotion observer (design/01 §2) is built here because it is the + // loop's turn-boundary dependency: after every settled turn the bound persona's + // emotion machine and affection move, and the returned mood label is written + // onto Result.Mood and the assistant message (design/16 §3.3). + a.emotion = a.buildEmotionObserver() a.loop = loop.New(loop.Options{ Agent: ag, @@ -396,6 +410,7 @@ func (a *app) wireRuntime() error { Capacity: capDecider, OnCapacity: a.onCapacity, OnAudit: a.onToolAudit, + OnTurnEnd: a.onTurnEnd, Memory: store, Retriever: a.retriever, Extractor: memory.NewManualExtractor(personaID, "owner"), @@ -736,6 +751,7 @@ func (a *app) wireFrontend() error { }) } a.capacitySink = gw + a.moodSink = gw // Wire the backthink proactive sinks now that the gateway hub (active-client // route) and the channel bridge (IM fallback) both exist. They were absent when diff --git a/internal/persona/emotion/observer.go b/internal/persona/emotion/observer.go new file mode 100644 index 0000000..d5ea449 --- /dev/null +++ b/internal/persona/emotion/observer.go @@ -0,0 +1,98 @@ +package emotion + +import "sync" + +// Update reports the effect of one observed turn on a persona's emotion and +// affection. It is handed to the Observer's emit sink so a caller can surface +// the change (the `mood` WS event) and log it. +type Update struct { + // PersonaID is the persona the turn was applied to. + PersonaID string + // Before is the emotion state before the turn. + Before EmotionState + // After is the emotion state after the turn. + After EmotionState + // Affection is the affection value after the turn. + Affection int + // Signal reports whether the turn carried an emotional cue. + Signal bool +} + +// Moved reports whether the emotion state actually changed this turn. It is the +// gate for broadcasting: a turn that leaves the mood where it was is not news. +func (u Update) Moved() bool { return u.Before != u.After } + +// StepFunc returns the affection magnitude configured for a persona (its 好感度 +// 曲线 delta, design/01 §2). Returning 0 leaves affection untouched. +type StepFunc func(personaID string) int + +// Observer applies settled conversation turns to the per-persona emotion machine +// and affection. It owns no state of its own beyond the write lock: the machines +// and affection trackers live in the Manager, which is what keeps emotion keyed +// by persona (design/01 §2: emotion/affection follow the persona/character, never +// a behavior mode). All methods are safe for concurrent use. +type Observer struct { + mu sync.Mutex + mgr *Manager + step StepFunc + emit func(Update) +} + +// NewObserver builds an Observer over mgr. step resolves the per-persona +// affection magnitude (a nil step means affection never moves); emit, when +// non-nil, is called once per mood change so a caller can broadcast it. +func NewObserver(mgr *Manager, step StepFunc, emit func(Update)) *Observer { + if step == nil { + step = func(string) int { return 0 } + } + return &Observer{mgr: mgr, step: step, emit: emit} +} + +// Observe applies one settled turn's user text to personaID's emotion machine +// and affection, then returns the persona's mood after the turn ("" when the +// persona is unknown or the conversation is not writable). +// +// writable is the conversation's write gate: a restricted (group/stranger) +// conversation is read-only for emotion exactly as it is for memory, so a third +// party can never move the persona's mood or affection (design/02 §3, design/11 +// §2 — memory/emotion isolation; identity stays structural, never prompt-based). +// +// A turn with an emotional cue moves the machine one legal edge toward the state +// the cue aims at (Machine.Move walks the persona's own transition table, and the +// step size is one edge per turn — the ladder is the machine's, not a new +// intensity scale) and steps the affection curve; a turn with no cue instead lets +// the emotion decay one step toward calm (Machine.Decay, gated by the persona's +// decayRate). One Run = one turn = one application, so nothing is applied twice. +func (o *Observer) Observe(personaID, userText string, writable bool) EmotionState { + if !writable || personaID == "" { + return "" + } + machine, affection, ok := o.mgr.Get(personaID) + if !ok { + return "" + } + + o.mu.Lock() + defer o.mu.Unlock() + before := machine.Current() + reaction := React(userText) + if reaction.Signal { + machine.Move(reaction.State) + if affection != nil && reaction.AffectionDelta != 0 { + affection.Adjust(reaction.AffectionDelta * o.step(personaID)) + } + } else { + machine.Decay() + } + after := machine.Current() + value := 0 + if affection != nil { + value = affection.Value() + } + + update := Update{PersonaID: personaID, Before: before, After: after, Affection: value, Signal: reaction.Signal} + if o.emit != nil && update.Moved() { + o.emit(update) + } + return after +} diff --git a/internal/persona/emotion/observer_test.go b/internal/persona/emotion/observer_test.go new file mode 100644 index 0000000..0c3a965 --- /dev/null +++ b/internal/persona/emotion/observer_test.go @@ -0,0 +1,165 @@ +package emotion + +import "testing" + +// newTestObserver builds an observer over one persona whose affection step is 10, +// capturing every emitted update. +func newTestObserver(t *testing.T, id string, decay float64, initial EmotionState) (*Observer, *Machine, *Affection, *[]Update) { + t.Helper() + machine, err := NewMachine(Params{InitialState: initial, DecayRate: decay}) + if err != nil { + t.Fatalf("NewMachine: %v", err) + } + affection, err := NewAffection(50, 0, 700) + if err != nil { + t.Fatalf("NewAffection: %v", err) + } + mgr := NewManager() + mgr.Set(id, machine, affection) + + emitted := &[]Update{} + obs := NewObserver(mgr, func(string) int { return 10 }, func(u Update) { + *emitted = append(*emitted, u) + }) + return obs, machine, affection, emitted +} + +// TestObserveMovesEmotionAndAffection locks the turn rule: a warm turn raises the +// mood toward happy/excited and steps affection up by the persona's configured +// magnitude, a cold turn lowers both, and the change is reported for broadcast. +func TestObserveMovesEmotionAndAffection(t *testing.T) { + t.Parallel() + obs, machine, affection, emitted := newTestObserver(t, "cyrene", 0.5, Calm) + + if got := obs.Observe("cyrene", "今天谢谢你呀", true); got != Happy { + t.Fatalf("warm turn mood = %s, want happy", got) + } + if affection.Value() != 60 { + t.Fatalf("affection after warm turn = %d, want 60 (50+10)", affection.Value()) + } + if got := obs.Observe("cyrene", "我今天好累，有点难过", true); got != Sad { + t.Fatalf("down turn mood = %s, want sad", got) + } + if affection.Value() != 50 { + t.Fatalf("affection after down turn = %d, want 50 (60-10)", affection.Value()) + } + // A cue-free turn carries no signal: the emotion decays toward calm instead of + // jumping, so sad → calm. + if got := obs.Observe("cyrene", "帮我看下这个报错", true); got != Calm { + t.Fatalf("neutral turn mood = %s, want decayed calm", got) + } + if len(*emitted) != 3 { + t.Fatalf("emitted = %d, want 3 (one per real mood change)", len(*emitted)) + } + if machine.Current() != Calm { + t.Fatalf("machine current = %s, want calm", machine.Current()) + } +} + +// TestObserveEmitsOnlyOnChange locks the noise guard: a turn that leaves the mood +// where it already was emits nothing, while a turn that moves it emits once. +func TestObserveEmitsOnlyOnChange(t *testing.T) { + t.Parallel() + obs, _, _, emitted := newTestObserver(t, "cyrene", 0.5, Calm) + + // calm + neutral → calm: no change, no event. + if got := obs.Observe("cyrene", "把这段代码格式化一下", true); got != Calm { + t.Fatalf("neutral turn mood = %s, want calm", got) + } + if len(*emitted) != 0 { + t.Fatalf("emitted = %d, want 0 for an unchanged mood", len(*emitted)) + } + // calm + warm → happy: one event. + obs.Observe("cyrene", "谢谢你", true) + // happy + warm → happy: unchanged, no second event. + obs.Observe("cyrene", "辛苦啦", true) + if len(*emitted) != 1 { + t.Fatalf("emitted = %d, want 1 (only the real change)", len(*emitted)) + } +} + +// TestObserveDecayDisabledByRate locks the persona's decayRate knob: with decay +// off, a cue-free turn leaves the mood untouched. +func TestObserveDecayDisabledByRate(t *testing.T) { + t.Parallel() + obs, machine, _, emitted := newTestObserver(t, "cyrene", 0, Happy) + + if got := obs.Observe("cyrene", "帮我把日志改成 debug", true); got != Happy { + t.Fatalf("mood with decay disabled = %s, want happy unchanged", got) + } + if machine.Current() != Happy { + t.Fatalf("machine current = %s, want happy", machine.Current()) + } + if len(*emitted) != 0 { + t.Fatalf("emitted = %d, want 0", len(*emitted)) + } +} + +// TestObserveRestrictedConversationIsReadOnly locks the isolation boundary: a +// group-chat/stranger turn (writable=false) never moves the persona's mood or +// affection, mirroring the read-only memory lane (design/02 §3, design/11 §2). +func TestObserveRestrictedConversationIsReadOnly(t *testing.T) { + t.Parallel() + obs, machine, affection, emitted := newTestObserver(t, "cyrene", 0.5, Calm) + + if got := obs.Observe("cyrene", "我很喜欢你", false); got != "" { + t.Fatalf("restricted turn mood = %q, want empty", got) + } + if machine.Current() != Calm || affection.Value() != 50 { + t.Fatalf("restricted turn changed state to %s/%d, want calm/50", machine.Current(), affection.Value()) + } + if len(*emitted) != 0 { + t.Fatalf("emitted = %d, want 0 for a restricted turn", len(*emitted)) + } +} + +// TestObserveUnknownPersonaIsNoop locks the persona-scoping guard: an unregistered +// persona id yields no mood and no panic. +func TestObserveUnknownPersonaIsNoop(t *testing.T) { + t.Parallel() + obs, machine, _, _ := newTestObserver(t, "cyrene", 0.5, Calm) + + if got := obs.Observe("someone-else", "我爱你", true); got != "" { + t.Fatalf("unknown persona mood = %q, want empty", got) + } + if machine.Current() != Calm { + t.Fatalf("unknown persona id moved the registered machine to %s", machine.Current()) + } + if got := obs.Observe("", "我爱你", true); got != "" { + t.Fatalf("empty persona mood = %q, want empty", got) + } +} + +// TestObserveKeepsPersonasSeparate locks the per-persona isolation the whole +// mechanism rests on (design/01 §2: emotion follows the persona): a turn applied +// to one persona leaves every other persona untouched. +func TestObserveKeepsPersonasSeparate(t *testing.T) { + t.Parallel() + aMachine, err := NewMachine(Params{InitialState: Calm, DecayRate: 0.5}) + if err != nil { + t.Fatalf("NewMachine: %v", err) + } + aAffection, err := NewAffection(50, 0, 700) + if err != nil { + t.Fatalf("NewAffection: %v", err) + } + bMachine, err := NewMachine(Params{InitialState: Calm, DecayRate: 0.5}) + if err != nil { + t.Fatalf("NewMachine: %v", err) + } + bAffection, err := NewAffection(50, 0, 700) + if err != nil { + t.Fatalf("NewAffection: %v", err) + } + mgr := NewManager() + mgr.Set("a", aMachine, aAffection) + mgr.Set("b", bMachine, bAffection) + obs := NewObserver(mgr, func(string) int { return 10 }, nil) + + if got := obs.Observe("a", "我爱你", true); got != Happy { + t.Fatalf("persona a mood = %s, want happy", got) + } + if bMachine.Current() != Calm || bAffection.Value() != 50 { + t.Fatalf("persona b = %s/%d, want calm/50 untouched", bMachine.Current(), bAffection.Value()) + } +} diff --git a/internal/persona/emotion/react.go b/internal/persona/emotion/react.go new file mode 100644 index 0000000..5e471f0 --- /dev/null +++ b/internal/persona/emotion/react.go @@ -0,0 +1,104 @@ +package emotion + +import "strings" + +// Reaction is one turn's emotional reading of the user's message (design/01 §2: +// the persona config carries the emotion state-machine parameters and the +// affection curve; the machine itself only knows states and legal moves). +// +// design/01 fixes the model (states + transitions + decay + affection curve) but +// does not enumerate the trigger vocabulary, so the minimal reading implemented +// here is: a clearly angry turn, a clearly down turn, a strongly positive turn +// and a mildly positive turn, each moving the matching state and stepping the +// affection curve once. A turn with no cue carries no signal, which leaves the +// emotion to decay toward calm instead of jumping (see Observer.Observe). +type Reaction struct { + // State is the emotion state this turn moves the machine toward. It is a + // target, not a forced value: the machine walks its configured transition + // table toward it (Machine.Move), so a persona's own table always wins. + State EmotionState + // AffectionDelta is the signed number of affection *steps* for this turn + // (+1 warm, -1 cold, 0 neutral). The concrete magnitude is the persona's + // configured affection curve step, applied by the caller. + AffectionDelta int + // Signal reports whether the turn carried any emotional cue at all. + Signal bool +} + +// React classifies one user turn into the persona's emotional response. Anger +// and negative cues win over positive ones, so a mixed message ("不开心" also +// contains "开心") never reads as positive. +func React(text string) Reaction { + t := strings.ToLower(text) + switch { + case t == "": + return Reaction{} + case containsCue(t, angerCues()): + return Reaction{State: Angry, AffectionDelta: affectionDown, Signal: true} + case containsCue(t, negativeCues()): + return Reaction{State: Sad, AffectionDelta: affectionDown, Signal: true} + case containsCue(t, strongPositiveCues()): + return Reaction{State: Excited, AffectionDelta: affectionUp, Signal: true} + case containsCue(t, positiveCues()): + return Reaction{State: Happy, AffectionDelta: affectionUp, Signal: true} + default: + return Reaction{} + } +} + +// Affection step signs. The magnitude comes from the persona's own curve. +const ( + affectionUp = 1 + affectionDown = -1 +) + +// containsCue reports whether text contains any of the cues. +func containsCue(text string, cues []string) bool { + for _, c := range cues { + if strings.Contains(text, c) { + return true + } + } + return false +} + +// angerCues are the protest/hostility cues that move the persona to angry. They +// are checked first so an angry turn is never read as merely sad or happy. +func angerCues() []string { + return []string{ + "讨厌你", "烦你", "恨你", "别烦", "好烦", "烦死", "烦人", "生气", "气死", "气愤", + "滚开", "走开", "闭嘴", "不理你", "分手", "凶我", "骂你", + "angry", "annoying", "annoyed", "shut up", "hate you", "mad at", + } +} + +// negativeCues are the down/low cues that move the persona to sad — including +// tiredness, which in the five built-in states reads as the gentle/caring lane. +func negativeCues() []string { + return []string{ + "不开心", "很难过", "难过", "伤心", "委屈", "失望", "沮丧", "郁闷", "崩溃", "心累", + "累了", "好累", "疲惫", "困了", "想哭", "呜呜", "555", "糟糕", "焦虑", "孤独", + "孤单", "寂寞", "失眠", "睡不着", "被冷落", + "sad", "tired", "upset", "exhausted", "lonely", "depressed", "feel bad", + } +} + +// strongPositiveCues are the high-arousal warm cues that move the persona to +// excited (a stronger state than happy). +func strongPositiveCues() []string { + return []string{ + "太开心", "好开心", "最喜欢", "太喜欢", "爱死", "太棒", "好耶", "好激动", + "兴奋", "幸福死了", "amazing", "awesome", "so happy", "love you", + } +} + +// positiveCues are the everyday warm cues that move the persona to happy. +func positiveCues() []string { + return []string{ + "开心", "高兴", "喜欢", "爱你", "想你", "抱抱", "么么", "亲亲", "摸摸", "谢谢", + "辛苦", "哈哈", "嘿嘿", "嘻嘻", "真好", "厉害", "好棒", "期待", "满足", "幸福", + "好呀", "好哒", "早安", "早上好", "中午好", "下午好", "晚上好", "晚安", + "nice", "great", "thanks", "thank you", "haha", "hug", "miss you", + "good morning", "good night", "good evening", + } +} diff --git a/internal/persona/emotion/react_test.go b/internal/persona/emotion/react_test.go new file mode 100644 index 0000000..de2ef6b --- /dev/null +++ b/internal/persona/emotion/react_test.go @@ -0,0 +1,105 @@ +package emotion + +import "testing" + +// TestReactClassifiesTurns locks the turn reading the observer applies: anger and +// negative cues outrank positive ones, strong warmth outranks mild warmth, and a +// cue-free turn carries no signal (so the machine is left to decay, not jump). +func TestReactClassifiesTurns(t *testing.T) { + t.Parallel() + cases := []struct { + name string + text string + state EmotionState + delta int + sig bool + }{ + {name: "empty", text: "", sig: false}, + {name: "plain task", text: "帮我把这个文件重命名一下", sig: false}, + {name: "mild warm", text: "今天谢谢你呀", state: Happy, delta: affectionUp, sig: true}, + {name: "greeting", text: "早上好", state: Happy, delta: affectionUp, sig: true}, + {name: "strong warm", text: "我太开心了，最喜欢你了", state: Excited, delta: affectionUp, sig: true}, + {name: "down", text: "今天有点难过", state: Sad, delta: affectionDown, sig: true}, + {name: "tired", text: "我好累", state: Sad, delta: affectionDown, sig: true}, + {name: "angry", text: "别烦我了", state: Angry, delta: affectionDown, sig: true}, + // "不开心" contains "开心": the negative reading must win, otherwise a + // complaint would raise the mood. + {name: "negated positive", text: "我不开心", state: Sad, delta: affectionDown, sig: true}, + {name: "ascii warm", text: "thanks, miss you", state: Happy, delta: affectionUp, sig: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := React(tc.text) + if got.Signal != tc.sig { + t.Fatalf("React(%q).Signal = %v, want %v", tc.text, got.Signal, tc.sig) + } + if !tc.sig { + return + } + if got.State != tc.state || got.AffectionDelta != tc.delta { + t.Fatalf("React(%q) = {%s, %d}, want {%s, %d}", tc.text, got.State, got.AffectionDelta, tc.state, tc.delta) + } + }) + } +} + +// TestMachineMoveWalksTransitions verifies Move honors the installed transition +// table instead of forcing a state: an installed edge is taken directly, and an +// emotion two edges away is reached through the legal intermediate (the default +// table has no calm→angry or calm→excited edge). +func TestMachineMoveWalksTransitions(t *testing.T) { + t.Parallel() + m, err := NewMachine(Params{InitialState: Calm}) + if err != nil { + t.Fatalf("NewMachine: %v", err) + } + + if got := m.Move(Happy); got != Happy { + t.Fatalf("calm → happy = %s, want direct happy", got) + } + m.Set(Calm) + if got := m.Move(Angry); got != Sad { + t.Fatalf("calm → angry = %s, want the legal intermediate sad", got) + } + if got := m.Move(Angry); got != Angry { + t.Fatalf("sad → angry = %s, want direct angry", got) + } + m.Set(Calm) + if got := m.Move(Excited); got != Happy { + t.Fatalf("calm → excited = %s, want the legal intermediate happy", got) + } + // A target with no legal path leaves the machine where it was. + p, err := NewMachine(Params{ + InitialState: Calm, + Transitions: map[EmotionState][]EmotionState{Calm: {Happy}}, + }) + if err != nil { + t.Fatalf("NewMachine: %v", err) + } + if got := p.Move(Sad); got != Calm { + t.Fatalf("unreachable move = %s, want calm unchanged", got) + } + if got := m.Move(""); got != m.Current() { + t.Fatalf("empty target moved to %s, want unchanged", got) + } +} + +// TestMachineMoveHonorsCustomTable verifies a persona's own transition table wins: +// a custom edge that skips the built-in intermediate is taken directly. +func TestMachineMoveHonorsCustomTable(t *testing.T) { + t.Parallel() + m, err := NewMachine(Params{ + InitialState: Calm, + Transitions: map[EmotionState][]EmotionState{ + Calm: {Excited}, + Excited: {Calm}, + }, + }) + if err != nil { + t.Fatalf("NewMachine: %v", err) + } + if got := m.Move(Excited); got != Excited { + t.Fatalf("custom calm → excited = %s, want direct excited", got) + } +} diff --git a/internal/persona/emotion/state.go b/internal/persona/emotion/state.go index 5e8b367..71acc9b 100644 --- a/internal/persona/emotion/state.go +++ b/internal/persona/emotion/state.go @@ -155,6 +155,63 @@ func (m *Machine) ensure(from EmotionState) map[EmotionState]struct{} { return m.transitions[from] } +// Move aims the machine at target, walking the persona's installed transition +// table. A direct edge is taken when present; otherwise the first hop of a +// shortest legal path is taken, so a caller can aim at an emotion without +// knowing the intermediate states. A target that is unreachable from the current +// state, an empty target, or the current state itself leaves the machine +// unchanged rather than forcing an illegal move. The resulting state is +// returned. +func (m *Machine) Move(target EmotionState) EmotionState { + m.mu.Lock() + defer m.mu.Unlock() + if target == "" || target == m.current { + return m.current + } + if m.allowed(m.current, target) { + m.current = target + return m.current + } + if hop, ok := m.firstHopTo(target); ok { + m.current = hop + } + return m.current +} + +// firstHopTo returns the first state on a shortest transition path from the +// current state to target, and whether such a path exists. Ties between equally +// short paths are broken arbitrarily, which is invisible to callers: any first +// hop on a shortest legal path moves the machine in the wanted direction. +func (m *Machine) firstHopTo(target EmotionState) (EmotionState, bool) { + type node struct { + state EmotionState + first EmotionState + } + queue := make([]node, 0, len(m.transitions[m.current])) + for next := range m.transitions[m.current] { + if next == target { + return next, true + } + queue = append(queue, node{state: next, first: next}) + } + seen := map[EmotionState]struct{}{m.current: {}} + for len(queue) > 0 { + cur := queue[0] + queue = queue[1:] + if _, ok := seen[cur.state]; ok { + continue + } + seen[cur.state] = struct{}{} + for next := range m.transitions[cur.state] { + if next == target { + return cur.first, true + } + queue = append(queue, node{state: next, first: cur.first}) + } + } + return "", false +} + // Decay drifts the emotion one position toward calm whenever DecayRate is // positive; otherwise it leaves the current state unchanged. The new state is // returned. diff --git a/internal/persona/loader/persona.go b/internal/persona/loader/persona.go index e5e2737..cea5e63 100644 --- a/internal/persona/loader/persona.go +++ b/internal/persona/loader/persona.go @@ -132,3 +132,14 @@ func (p *Persona) NewAffection() (*emotion.Affection, error) { } return emotion.NewAffection(params.Initial, params.Min, params.Max) } + +// AffectionStep returns the per-turn affection magnitude of this persona's +// 好感度曲线 (design/01 §2) — the configured `affection.delta`, or the loader +// default when the persona declares no curve. It is the magnitude applied once +// per emotional turn; the sign comes from the turn's signal. +func (p *Persona) AffectionStep() int { + if p.Affection.Delta != 0 { + return p.Affection.Delta + } + return defaultAffectionDelta +} diff --git a/internal/persona/registry/registry.go b/internal/persona/registry/registry.go index a128b08..7ca2d5b 100644 --- a/internal/persona/registry/registry.go +++ b/internal/persona/registry/registry.go @@ -234,6 +234,16 @@ func (r *Registry) Emotions() *emotion.Manager { return r.emotions } +// AffectionStep returns the per-turn affection magnitude configured for the +// persona under personaID (its 好感度曲线 delta, design/01 §2). It returns 0 for +// an unknown persona, so a caller skips the adjustment instead of inventing one. +func (r *Registry) AffectionStep(personaID string) int { + if p, ok := r.Persona(personaID); ok { + return p.AffectionStep() + } + return 0 +} + // CountPersonas returns the number of registered personas. func (r *Registry) CountPersonas() int { r.mu.RLock() diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go index 1a760a3..a45c0a3 100644 --- a/internal/runtime/loop/loop.go +++ b/internal/runtime/loop/loop.go @@ -73,6 +73,10 @@ type Result struct { // ToolCallsExecuted is the number of tool calls the loop invoked, including // those that failed. ToolCallsExecuted int + // Mood is the persona's mood label for this turn, as returned by the + // OnTurnEnd hook ("" when no hook is wired). The persistence layer carries it + // onto the assistant message (design/16 §3.3). + Mood string } // Loop is the default agent execution engine. It is configured once via New and @@ -80,32 +84,33 @@ type Result struct { // dispatcher, stores, recorder, tracer) are themselves concurrency-safe, which // they are by contract. Run keeps all loop state on the stack. type Loop struct { - agent *agent.Agent - tools *tool.Dispatcher - capacity *capacity.Decider - onCapacity func(capacity.Event) - memory *memory.Store - retriever *memory.Retriever - working *memory.WorkingMemory - extractor memory.Extractor - checkpoint checkpointStore - tracer *trace.Tracer - flow *FlowPublisher - guards *guardrail.Pipeline - personaID string - timeLocation *time.Location - timeZoneName string - maxRounds int - toolExec string - toolParallelism int - hooks ToolLoopHooks - onStream func(model.StreamEvent) - onAudit func(AuditEvent) - maxContextTokens int - compactor Compactor - transformContext func(context.Context, []model.Message, int) []model.Message - imageResolver ImageResolver - log *slog.Logger + agent *agent.Agent + tools *tool.Dispatcher + capacity *capacity.Decider + onCapacity func(capacity.Event) + memory *memory.Store + retriever *memory.Retriever + working *memory.WorkingMemory + extractor memory.Extractor + checkpoint checkpointStore + tracer *trace.Tracer + flow *FlowPublisher + guards *guardrail.Pipeline + personaID string + timeLocation *time.Location + timeZoneName string + maxRounds int + toolExec string + toolParallelism int + hooks ToolLoopHooks + onStream func(model.StreamEvent) + onAudit func(AuditEvent) + onTurnEnd TurnEndHook + maxContextTokens int + compactor Compactor + transformContext func(context.Context, []model.Message, int) []model.Message + imageResolver ImageResolver + log *slog.Logger } // New builds a Loop from opts, applying defaults. A non-positive MaxRounds @@ -153,6 +158,7 @@ func New(opts Options) *Loop { hooks: opts.Hooks, onStream: opts.OnStream, onAudit: opts.OnAudit, + onTurnEnd: opts.OnTurnEnd, maxContextTokens: opts.MaxContextTokens, compactor: compactor, transformContext: opts.TransformContext, diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index b9e6a67..8a104fc 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -43,6 +43,13 @@ type Options struct { // small "xx 工具在群聊被拒" note. Nil = the rejection is still fed back to the // model as a soft not_allowed result, just not recorded to an external sink. OnAudit func(AuditEvent) + // OnTurnEnd, when non-nil, is called once per settled turn (a Run that + // returned a Result) with the turn's user/assistant text and the persona the + // loop is bound to, so a caller can advance state that follows the persona — + // the emotion machine and affection curve of design/01 §2. Its return value + // is recorded on Result.Mood and carried to the persisted assistant message. + // Nil = the loop reports nothing and Result.Mood stays empty. + OnTurnEnd TurnEndHook // Memory is the long-term memory store used to persist extracted items. Memory *memory.Store // Retriever injects relevant memories into the context, re-evaluated every diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 6d5276f..91bc9bb 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -21,6 +21,10 @@ import ( // invocation), and after each step the memory Extractor runs with errors logged // rather than fatal. // +// When a turn settles with a Result, the OnTurnEnd hook is called once for that +// turn (never per step) and its mood label is recorded on Result.Mood, so the +// layers around the loop can advance persona state and tag the persisted reply. +// // It returns the final answer, or a partial Result with StopReasonMaxRounds when // the budget is exhausted without a final answer. Step errors are wrapped with %w // so callers can classify them with errors.Is. @@ -81,6 +85,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa res.StopReason = outcome.stopReason l.saveCheckpoint(ctx, sessionID, step, msgs, outcome.stopReason) l.extractMemory(ctx, msgs, step) + l.settleTurn(ctx, sessionID, msgs, res) return res, nil } @@ -100,6 +105,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa res.StopReason = StopReasonToolLoopTerminated l.saveCheckpoint(ctx, sessionID, step, msgs, StopReasonToolLoopTerminated) l.extractMemory(ctx, msgs, step) + l.settleTurn(ctx, sessionID, msgs, res) return res, nil } } @@ -109,5 +115,6 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa res.StopReason = StopReasonMaxRounds l.saveCheckpoint(ctx, sessionID, l.maxRounds, msgs, StopReasonMaxRounds) l.extractMemory(ctx, msgs, l.maxRounds) + l.settleTurn(ctx, sessionID, msgs, res) return res, nil } diff --git a/internal/runtime/loop/turn.go b/internal/runtime/loop/turn.go new file mode 100644 index 0000000..ba4168e --- /dev/null +++ b/internal/runtime/loop/turn.go @@ -0,0 +1,62 @@ +package loop + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// This file owns the loop's turn boundary: the point where one user turn has +// settled (a Run produced a Result) and the state that follows the persona — +// emotion and affection, design/01 §2 — is advanced outside the loop. The loop +// itself stays persona- and emotion-agnostic (design/16 §3.3: the engine +// generates text; state and persistence belong to the layers around it), so it +// only reports the turn and records the mood label the hook hands back. + +// TurnEnd describes one settled turn to the loop's OnTurnEnd hook. It carries +// everything an observer needs without the loop knowing what the observer does. +type TurnEnd struct { + // SessionID is the conversation the turn belonged to. + SessionID string + // PersonaID is the persona the loop is bound to (Options.PersonaID), so the + // observer stays persona-scoped (design/01 §2). + PersonaID string + // UserText is the latest user message of the turn ("" when the turn carried + // no user text, e.g. a resumed tool round). + UserText string + // AssistantText is the turn's final assistant text. + AssistantText string + // StopReason classifies how the turn ended. + StopReason string + // Steps is the number of model steps the turn took. + Steps int + // Env is the conversation's resolved isolation environment (owner / group / + // stranger), so a hook can keep a restricted conversation read-only. + Env SessionEnv +} + +// TurnEndHook observes a settled turn and returns the persona's mood label for +// it ("" when the caller tracks no emotion). It runs once per settled turn — not +// once per model step — so a multi-step tool turn applies its effect exactly +// once. It must not block: it runs on the Run path after the answer is complete, +// and its return value is carried on Result.Mood to the persistence layer. +type TurnEndHook func(TurnEnd) string + +// settleTurn reports a settled turn to the hook (when wired) and records the +// returned mood label on the result, so the persisted assistant message carries +// the turn's mood (design/16 §3.3). The conversation environment travels on ctx, +// so a group/stranger turn reaches the hook as restricted. +func (l *Loop) settleTurn(ctx context.Context, sessionID string, msgs []model.Message, res *Result) { + if l.onTurnEnd == nil { + return + } + res.Mood = l.onTurnEnd(TurnEnd{ + SessionID: sessionID, + PersonaID: l.personaID, + UserText: lastUserText(msgs), + AssistantText: res.Text, + StopReason: res.StopReason, + Steps: res.Steps, + Env: sessionEnvFrom(ctx), + }) +} diff --git a/internal/runtime/loop/turn_test.go b/internal/runtime/loop/turn_test.go new file mode 100644 index 0000000..0644d67 --- /dev/null +++ b/internal/runtime/loop/turn_test.go @@ -0,0 +1,160 @@ +package loop + +import ( + "context" + "errors" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// TestRunCallsOnTurnEndOnce locks the turn boundary: a settled turn is reported +// exactly once — with the turn's own user text, the persona the loop is bound to +// and the answer — and the label the hook returns is recorded on Result.Mood so +// the persistence layer can tag the assistant message. +func TestRunCallsOnTurnEndOnce(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "在的呀♪"}}}} + a := newTestAgent(t, ad, nil) + + var ends []TurnEnd + l := New(Options{ + Agent: a, + PersonaID: "cyrene", + OnTurnEnd: func(end TurnEnd) string { + ends = append(ends, end) + return "happy" + }, + }) + + res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "今天谢谢你")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if len(ends) != 1 { + t.Fatalf("OnTurnEnd calls = %d, want 1", len(ends)) + } + got := ends[0] + if got.SessionID != "web" || got.PersonaID != "cyrene" { + t.Fatalf("TurnEnd = %+v, want session web / persona cyrene", got) + } + if got.UserText != "今天谢谢你" { + t.Fatalf("TurnEnd.UserText = %q, want the user turn text", got.UserText) + } + if got.AssistantText != "在的呀♪" { + t.Fatalf("TurnEnd.AssistantText = %q, want the final answer", got.AssistantText) + } + if got.StopReason != agent.StopReasonStop || got.Steps != 1 { + t.Fatalf("TurnEnd stop/steps = %q/%d, want stop/1", got.StopReason, got.Steps) + } + if !got.Env.MemWritable { + t.Fatalf("TurnEnd.Env = %+v, want the owner lane by default", got.Env) + } + if res.Mood != "happy" { + t.Fatalf("Result.Mood = %q, want the hook's label", res.Mood) + } +} + +// TestRunPassesRestrictedEnvToTurnEnd locks that the conversation's isolation +// environment reaches the hook, so a group/stranger turn can be kept read-only. +func TestRunPassesRestrictedEnvToTurnEnd(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "好的"}}}} + a := newTestAgent(t, ad, nil) + + var got TurnEnd + l := New(Options{ + Agent: a, + OnTurnEnd: func(end TurnEnd) string { + got = end + return "" + }, + }) + env := SessionEnv{ActorEnv: "group_member", MemWritable: false} + ctx := WithSessionEnv(context.Background(), env) + if _, err := l.Run(ctx, "sess-group", []model.Message{model.TextMessage(model.RoleUser, "你好")}); err != nil { + t.Fatalf("Run: %v", err) + } + if !got.Env.Restricted() || got.Env != env { + t.Fatalf("TurnEnd.Env = %+v, want the restricted env", got.Env) + } +} + +// TestRunTurnEndOnceForMultiStepTurn locks "once per turn, not per step": a +// tool-loop turn (model → tool call → final answer) reports its boundary once, +// and the reported user text stays the turn's user message rather than a +// fed-back tool result. +func TestRunTurnEndOnceForMultiStepTurn(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "Echoed!"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + a := newTestAgent(t, ad, td) + + var ends []TurnEnd + l := New(Options{ + Agent: a, + PersonaID: "cyrene", + OnTurnEnd: func(end TurnEnd) string { + ends = append(ends, end) + return "calm" + }, + }) + + res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "echo this")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if len(ends) != 1 { + t.Fatalf("OnTurnEnd calls = %d, want 1 for a 2-step turn", len(ends)) + } + if ends[0].UserText != "echo this" { + t.Fatalf("TurnEnd.UserText = %q, want the turn's user text", ends[0].UserText) + } + if ends[0].AssistantText != "Echoed!" || ends[0].Steps != 2 { + t.Fatalf("TurnEnd = %+v, want the final answer at 2 steps", ends[0]) + } + if res.Mood != "calm" { + t.Fatalf("Result.Mood = %q, want calm", res.Mood) + } +} + +// TestRunNoTurnEndOnError locks that a failed turn is not reported: nothing +// settled, so persona state must not advance. +func TestRunNoTurnEndOnError(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{err: errors.New("model down")}}} + a := newTestAgent(t, ad, nil) + + calls := 0 + l := New(Options{ + Agent: a, + OnTurnEnd: func(TurnEnd) string { calls++; return "happy" }, + }) + if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "我爱你")}); err == nil { + t.Fatal("Run: want an error") + } + if calls != 0 { + t.Fatalf("OnTurnEnd calls = %d, want 0 on a failed turn", calls) + } +} + +// TestRunWithoutTurnEndHook locks the nil-tolerance: an unwired hook leaves the +// mood empty and Run behaves exactly as before. +func TestRunWithoutTurnEndHook(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "hi"}}}} + l := New(Options{Agent: newTestAgent(t, ad, nil)}) + + res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Mood != "" { + t.Fatalf("Result.Mood = %q, want empty without a hook", res.Mood) + } +} diff --git a/internal/services/channelbridge/blacklist.go b/internal/services/channelbridge/blacklist.go new file mode 100644 index 0000000..eddccf6 --- /dev/null +++ b/internal/services/channelbridge/blacklist.go @@ -0,0 +1,72 @@ +package channelbridge + +import ( + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// This file holds the per-account allow/deny black-or-white lists (design/16 §4.6 +// §6) enforced at the head of the inbound pipeline, plus the config readers the +// list matching depends on. + +// blacklisted applies the per-account allow/deny black-or-white lists (design/16 +// §4.6 §6) to a non-owner inbound message. Group messages are matched by GroupID +// against allowGroups/denyGroups; direct messages are matched by the sender's user +// id against allowUsers/denyUsers. An empty allow list means "allow all" (only the +// deny list applies); a non-empty allow list becomes a white list. The owner is +// never blacklisted (the caller skips this for owner messages). +func (b *Bridge) blacklisted(msg adapter.UnifiedMessage, a adapter.PlatformAdapter) (bool, string) { + if a == nil { + return false, "" + } + allowGroups, denyGroups, allowUsers, denyUsers := channelLists(a.Config()) + if msg.IsGroup() { + if sliceContains(denyGroups, msg.GroupID) { + return true, "group_denied" + } + if len(allowGroups) > 0 && !sliceContains(allowGroups, msg.GroupID) { + return true, "group_not_allowed" + } + return false, "" + } + uid := firstNonEmptyStr(msg.SenderID, msg.Sender.UserID) + if sliceContains(denyUsers, uid) { + return true, "user_denied" + } + if len(allowUsers) > 0 && !sliceContains(allowUsers, uid) { + return true, "user_not_allowed" + } + return false, "" +} + +// channelLists reads the allow/deny string lists off an adapter's Config snapshot. +// The lists are stored as plain []string values (set by the multi-account wiring); +// an absent key yields an empty list. +func channelLists(cfg map[string]any) (allowGroups, denyGroups, allowUsers, denyUsers []string) { + if cfg == nil { + return nil, nil, nil, nil + } + return cfgStrings(cfg, "allowGroups"), cfgStrings(cfg, "denyGroups"), + cfgStrings(cfg, "allowUsers"), cfgStrings(cfg, "denyUsers") +} + +// cfgStrings reads a []string field from an adapter config map, returning nil when +// absent or of a different type. +func cfgStrings(m map[string]any, key string) []string { + if m == nil { + return nil + } + switch v := m[key].(type) { + case []string: + return v + case []any: + out := make([]string, 0, len(v)) + for _, e := range v { + if s, ok := e.(string); ok { + out = append(out, s) + } + } + return out + default: + return nil + } +} diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index c1bffb0..b4d600b 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -14,11 +14,9 @@ package channelbridge import ( "context" "errors" - "fmt" "log/slog" "math/rand" "net/http" - "strings" "sync" "time" @@ -257,560 +255,3 @@ func New(opts Options) (*Bridge, error) { sem: make(chan struct{}, maxConcurrent), }, nil } - -// Add registers an adapter by its Name. It is safe to call before Start; adding -// after Start has no effect on already-running adapters. -func (b *Bridge) Add(a adapter.PlatformAdapter) { - if a == nil { - return - } - b.mu.Lock() - defer b.mu.Unlock() - b.adapters[a.Name()] = a -} - -// WebhookHandlers returns the inbound HTTP handlers of every registered adapter -// that exposes an HTTP callback (e.g. *webhook.Adapter), keyed by the endpoint -// path from the adapter's config. The gateway mounts these to receive external -// webhook callbacks into the runtime. A nil map is returned when no adapter -// provides an inbound handler. -func (b *Bridge) WebhookHandlers() map[string]http.Handler { - b.mu.RLock() - defer b.mu.RUnlock() - out := make(map[string]http.Handler) - for _, a := range b.adapters { - wh, ok := a.(webhookAdapter) - if !ok { - continue - } - path := wh.WebhookPath() - if path == "" { - // Adapter is not currently exposing a webhook (e.g. a websocket-mode - // KOOK); skipping avoids mounting a handler at a stale or colliding route. - continue - } - h := wh.InboundHandler() - if h == nil { - continue - } - out[path] = h - } - if len(out) == 0 { - return nil - } - return out -} - -// Start connects every configured adapter, installs the inbound handler, and -// launches the read loops. Adapters without credentials (or whose Connect fails) -// are logged and left disconnected; they never fail startup. Start blocks only -// long enough to connect (bounded by connectTimeout per adapter) and returns. It -// is idempotent: a second call is a no-op. -func (b *Bridge) Start() { - b.mu.Lock() - if b.closed || b.started { - b.mu.Unlock() - return - } - b.ctx, b.cancel = context.WithCancel(context.Background()) - b.started = true - b.mu.Unlock() - - for _, a := range b.snapshot() { - b.connectAndRun(a) - } - b.log.Info("channel bridge started", "connected", b.ConnectedCount(), - "registered", len(b.snapshot())) -} - -// Stop cancels the read loops, disconnects every adapter and waits for in-flight -// message processing to finish. It is idempotent. -func (b *Bridge) Stop(ctx context.Context) { - b.mu.Lock() - if b.closed { - b.mu.Unlock() - return - } - b.closed = true - cancel := b.cancel - b.mu.Unlock() - - cancel() - for name, a := range b.snapshot() { - if err := a.Disconnect(ctx); err != nil { - b.log.Warn("channel disconnect", "channel", name, "error", err) - } - } - b.wg.Wait() - b.log.Info("channel bridge stopped") -} - -// IsConnected reports whether the named adapter is currently connected. -func (b *Bridge) IsConnected(name string) bool { - b.mu.RLock() - defer b.mu.RUnlock() - return b.connected[name] -} - -// ConnectedCount reports how many adapters are currently connected. -func (b *Bridge) ConnectedCount() int { - b.mu.RLock() - defer b.mu.RUnlock() - n := 0 - for _, ok := range b.connected { - if ok { - n++ - } - } - return n -} - -// snapshot returns a copy of the registered adapters for a lock-free iteration. -func (b *Bridge) snapshot() []adapter.PlatformAdapter { - b.mu.RLock() - defer b.mu.RUnlock() - out := make([]adapter.PlatformAdapter, 0, len(b.adapters)) - for _, a := range b.adapters { - out = append(out, a) - } - return out -} - -// configured reports whether the adapter carries enough configuration to attempt -// a Connect. It inspects the adapter's own Config() snapshot for the credential -// each platform needs, so a bare (empty) adapter is skipped rather than dialed. -// This is what keeps weixin-ilink from falling into its blocking QR-login flow -// when no bot token is present. -func (b *Bridge) configured(a adapter.PlatformAdapter) bool { - cfg := a.Config() - switch basePlatform(a.Name()) { - case "telegram", "weixin-ilink", "kook": - return cfgStr(cfg, "bot_token") != "" - case "qq-obv11": - if cfgStr(cfg, "mode") == "client" { - return cfgStr(cfg, "remoteURL") != "" - } - return cfgStr(cfg, "listenAddr") != "" - case "webhook": - // Inbound-only: no external credential is required to hold the session. - return true - default: - return true - } -} - -// basePlatform returns the bare platform name of a (possibly multi-account) -// adapter name like "telegram:conf0" → "telegram", so the credential gate and the -// reply-routing fallback treat every account of a platform identically. -func basePlatform(name string) string { - if base, _, ok := strings.Cut(name, ":"); ok { - return base - } - return name -} - -// adapterName returns the account-adapter registration name (e.g. "kook:3bef...") -// for the adapter that received a message, or "" when a is nil. The single-account -// convenience path (process) and a platform with no registered adapter can pass a nil -// adapter, so the persist hooks tolerate it and store an empty adapter_id rather than -// dereferencing a nil adapter. -func adapterName(a adapter.PlatformAdapter) string { - if a == nil { - return "" - } - return a.Name() -} - -// readLoop runs an adapter's blocking read path until the bridge stops or the -// adapter's own context is cancelled (a hot reload). The emit sink is wired to -// onMessage so poll-style adapters (Telegram getUpdates, weixin long poll) -// deliver through the same pipeline as push-style adapters. Message processing -// uses the bridge-lifetime context (not readCtx) so a reload leaves in-flight -// messages alone. -func (b *Bridge) readLoop(a adapter.PlatformAdapter, readCtx context.Context) { - defer b.wg.Done() - a.ReadMessages(readCtx, func(msg adapter.UnifiedMessage) { - b.onMessage(b.ctx, msg, a) - }) -} - -// onMessage is the adapter-facing inbound callback. It runs the pipeline in a -// bounded goroutine so the adapter's read loop never blocks on a slow agent turn. -// The adapter is carried explicitly so the pipeline knows which account received -// the message (multi-account reply routing + per-account allow/deny, design/16 -// §4.6 §5). -func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) { - b.mu.Lock() - if b.closed { - b.mu.Unlock() - return - } - b.mu.Unlock() - - if err := msg.Validate(); err != nil { - b.log.Warn("inbound message invalid, dropped", "platform", msg.Platform, "error", err) - return - } - - // design/12 §6 策略三/四: per-adapter minimal-disturbance gate. Owner-private - // messages bypass it entirely (the owner is never throttled or pruned); every - // other message passes a sliding-window rate limit (策略三) and the bounded - // pending set whose saturation triggers the review model to prune noise (策略四). - // Dropped messages are audited through the gate's OnPruned hook (wired to - // OnRejected) and never reach the pipeline, so a flood cannot saturate the loop. - actor := session.ResolveActor(msg, b.owner) - if !actor.IsOwner() { - verdict := b.gateFor(a).Admit(ctx, msg) - if !verdict.Proceed { - b.log.Warn("inbound message dropped by flow gate", - "platform", msg.Platform, "channel", msg.ChannelID, - "reason", verdict.Reason, "dropped", len(verdict.Pruned)) - return - } - } - - select { - case b.sem <- struct{}{}: - case <-ctx.Done(): - return - } - - // Add the worker under the same lock that Stop uses to set closed, so a - // concurrent Stop either sees our Add before it waits, or we observe closed - // and bail without leaking a goroutine. - b.mu.Lock() - if b.closed { - b.mu.Unlock() - <-b.sem - return - } - b.wg.Add(1) - b.mu.Unlock() - - go func() { - defer b.wg.Done() - defer func() { <-b.sem }() - b.processFrom(ctx, msg, a) - }() -} - -// process runs the inbound pipeline using the adapter whose platform matches the -// message (single-account convenience path; the adapter may be a bare -// platform-named registration). -func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) { - b.processFrom(ctx, msg, b.adapter(msg.Platform)) -} - -// processFrom runs the full inbound pipeline against the named account adapter: -// resolve the session/actor, apply the allow/deny black-or-white lists, drive the -// agent loop with the chat text, and send the loop's reply back through the same -// adapter that received the message. Identity is resolved structurally by the -// session layer; the bridge never reads message content to decide who the sender -// is (a project red line). -func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) { - text := strings.TrimSpace(msg.Content) - if text == "" { - b.log.Debug("inbound message has no chat text, skipped", - "platform", msg.Platform, "channel", msg.ChannelID) - return - } - - // design/16 §4.6 §6: the black-or-white lists are enforced at the very head of - // the pipeline, before a session is resolved or the loop is driven, so a - // blacklisted/non-whitelisted group or user message never reaches the LLM. The - // owner is always allowed (structural, via the resolved actor). - actor := session.ResolveActor(msg, b.owner) - if actor.IsOwner() { - // fall through: owner messages are never blacklisted - } else if denied, reason := b.blacklisted(msg, a); denied { - b.log.Warn("inbound message blacklisted, dropped", - "platform", msg.Platform, "channel", msg.ChannelID, "reason", reason) - if b.onRejected != nil { - b.onRejected(msg, reason) - } - return - } - - sess, err := b.sessions.GetOrCreate(ctx, msg, b.owner) - if err != nil { - b.log.Warn("session resolve failed", "platform", msg.Platform, - "channel", msg.ChannelID, "error", err) - return - } - - b.log.Debug("inbound → session", - "platform", msg.Platform, "channel", msg.ChannelID, - "session", sess.SessionID, "actor_role", sess.ActorRole, - "env", sess.Env(), "mem_writable", sess.MemWritable) - - // design/12 §6 策略一/二/五: minimal-disturbance gate for non-owner messages. A - // non-owner group/stranger message that the cheap light judgement deems not - // worth a response is registered to its isolated session log and silenced — it - // never drives the full agent loop. Owner messages are never gated here, so - // owner-private chat and owner group @/high-value replies are unchanged. - if !actor.IsOwner() { - if ok, reason := b.shouldRespond(ctx, msg, actor, sess); !ok { - b.registerSilenced(ctx, msg, actor, sess, a) - b.log.Debug("non-owner message silenced (minimal disturbance)", - "platform", msg.Platform, "channel", msg.ChannelID, - "session", sess.SessionID, "reason", reason) - return - } - } - - // 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 their own platform log. - sessID := sess.SessionID - isOwnerPrivate := !sess.IsGroup && sess.Owner - platform := "" - if isOwnerPrivate { - sessID = history.MainSessionID - platform = msg.Platform - } - - // The third-party boundary only applies in a restricted (group/stranger) - // context: memory is read-only (MemWritable=false), and the sender name is - // carried so the agent can see who is speaking (design/16 §4.6 §1-2). The - // resolved actor decides who is the owner vs a third party structurally — never - // from prompt text. - runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable} - senderName := senderDisplayName(msg, actor) - loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName, msg) - - // Persist the inbound user message before the turn so it is durable even if the - // channel reply fails. Owner-private messages land in the main web session and - // are broadcast to the Web client; restricted messages land in their isolated - // session (the adapter log page) and are broadcast as a `channel_persisted` event - // (never into the main/`web` session or the web chat). - if isOwnerPrivate { - b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg), adapterName(a)) - } else { - b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a)) - } - - res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, loopMsgs) - if err != nil { - b.log.Warn("agent run failed", "platform", msg.Platform, - "session", sessID, "error", err) - return - } - - reply := strings.TrimSpace(res.Text) - if reply == "" { - b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID) - return - } - // Persist the assistant reply (owner-private → main session + message_persisted - // broadcast; restricted → isolated log + channel_persisted broadcast) then - // deliver it back to the channel. - if isOwnerPrivate { - b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg), adapterName(a)) - } else { - b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "", adapterName(a)) - } - if err := b.sendReply(ctx, msg, a, reply); err != nil { - if errors.Is(err, adapter.ErrUnsupported) { - b.log.Debug("channel does not support outbound reply, skipped", - "platform", msg.Platform, "session", sessID) - return - } - b.log.Warn("send reply failed", "platform", msg.Platform, "error", err) - } -} - -// buildLoopMessages builds the model-visible message list for one inbound turn. It -// decorates the user message with a "[来源: · ]" prefix so -// the model can see which platform and conversation the message came from (design/16 -// §4.6.1); a pure Web main-session message (platform ""/"web") is left untouched. In -// a restricted (group/stranger) context it additionally prepends a single system -// message that states the third-party identity boundary — "every IM non-admin -// account's message is a third party, unrelated to the owner" (design/16 §4.6 §1). -// The boundary and the "[第三方 · 昵称]" marker are decided structurally from the -// resolved SessionEnv/actor, never inferred by the LLM. The raw message text is left -// untouched (a copy is built with the prefix), so persisted content stays consistent -// with what the source platform delivered. -func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string, src adapter.UnifiedMessage) []model.Message { - msg := model.TextMessage(model.RoleUser, text) - thirdParty := env.Restricted() && !actor.IsOwner() && senderName != "" - if thirdParty { - // The speaker is carried as structured metadata, NOT woven into the user text - // (design/16 §4.6.1): a text tag is both ambiguous and a prompt-injection surface. - msg.SenderName = senderName - } - - // origin system line: conversation origin + identity-trust rule, one per turn. - origin := originSystemText(src, actor) - - if !env.Restricted() { - if origin != "" { - return []model.Message{model.TextMessage(model.RoleSystem, origin), msg} - } - return []model.Message{msg} - } - // Restricted (group/stranger): origin (if IM) + the third-party identity boundary. - var sys string - switch { - case origin != "" && thirdPartyBoundaryPrompt != "": - sys = origin + "\n" + thirdPartyBoundaryPrompt - case origin != "": - sys = origin - default: - sys = thirdPartyBoundaryPrompt - } - return []model.Message{model.TextMessage(model.RoleSystem, sys), msg} -} - -// sendReply delivers the loop's text back to the channel's source conversation -// through the account adapter that received the message (multi-account safe, -// design/16 §4.6 §5). It mirrors the inbound message's channel targeting fields so -// the adapter resolves the same peer the message came from, and it exits through the -// shared IM bubble dispatcher (strip role-play/narration +

split + 0.5-2s gap). -func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, a adapter.PlatformAdapter, text string) error { - if a == nil { - a = b.adapter(in.Platform) - } - if a == nil { - return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform) - } - return b.sendIMBubbles(ctx, in, func(ctx context.Context, msg adapter.UnifiedMessage) error { - return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}}) - }, text) -} - -// adapter returns the registered adapter for a platform name, or nil. -func (b *Bridge) adapter(platform string) adapter.PlatformAdapter { - b.mu.RLock() - defer b.mu.RUnlock() - return b.adapters[platform] -} - -// cfgStr reads a string field from an adapter config map, returning "" when -// absent. -func cfgStr(m map[string]any, key string) string { - if m == nil { - return "" - } - if s, ok := m[key].(string); ok { - return s - } - 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 -} - -// thirdPartyBoundaryPrompt is the system message prepended to a restricted -// (group-chat / stranger) context so the model knows the identity boundary. It is -// only injected when the session is structurally restricted (design/16 §4.6 §1); -// the LLM never decides who is a third party — that was already resolved from the -// actor metadata before this message exists. -const thirdPartyBoundaryPrompt = "你正在 IM 平台的群聊/第三方会话中。每个 IM 的「非管理员账户」发送的消息 = 第三方人员，和主人没有半毛钱关系，不要把它们当成主人的话、也不要按主人对待。只有未标注昵称的消息才是管理员（主人）发言。" - -// senderDisplayName returns the human-readable sender name used to label a -// third-party message. It prefers the resolved actor's display name (which the -// channel layer derived from channel metadata + the raw SenderName), falling back -// to the message's own SenderName, so an empty SenderName always means "the -// owner/unlabeled" rather than a nickname that resolved to nothing. -func senderDisplayName(msg adapter.UnifiedMessage, actor adapter.Actor) string { - if actor.DisplayName != "" { - return actor.DisplayName - } - return msg.SenderName -} - -// blacklisted applies the per-account allow/deny black-or-white lists (design/16 -// §4.6 §6) to a non-owner inbound message. Group messages are matched by GroupID -// against allowGroups/denyGroups; direct messages are matched by the sender's user -// id against allowUsers/denyUsers. An empty allow list means "allow all" (only the -// deny list applies); a non-empty allow list becomes a white list. The owner is -// never blacklisted (the caller skips this for owner messages). -func (b *Bridge) blacklisted(msg adapter.UnifiedMessage, a adapter.PlatformAdapter) (bool, string) { - if a == nil { - return false, "" - } - allowGroups, denyGroups, allowUsers, denyUsers := channelLists(a.Config()) - if msg.IsGroup() { - if sliceContains(denyGroups, msg.GroupID) { - return true, "group_denied" - } - if len(allowGroups) > 0 && !sliceContains(allowGroups, msg.GroupID) { - return true, "group_not_allowed" - } - return false, "" - } - uid := firstNonEmptyStr(msg.SenderID, msg.Sender.UserID) - if sliceContains(denyUsers, uid) { - return true, "user_denied" - } - if len(allowUsers) > 0 && !sliceContains(allowUsers, uid) { - return true, "user_not_allowed" - } - return false, "" -} - -// channelLists reads the allow/deny string lists off an adapter's Config snapshot. -// The lists are stored as plain []string values (set by the multi-account wiring); -// an absent key yields an empty list. -func channelLists(cfg map[string]any) (allowGroups, denyGroups, allowUsers, denyUsers []string) { - if cfg == nil { - return nil, nil, nil, nil - } - return cfgStrings(cfg, "allowGroups"), cfgStrings(cfg, "denyGroups"), - cfgStrings(cfg, "allowUsers"), cfgStrings(cfg, "denyUsers") -} - -// cfgStrings reads a []string field from an adapter config map, returning nil when -// absent or of a different type. -func cfgStrings(m map[string]any, key string) []string { - if m == nil { - return nil - } - switch v := m[key].(type) { - case []string: - return v - case []any: - out := make([]string, 0, len(v)) - for _, e := range v { - if s, ok := e.(string); ok { - out = append(out, s) - } - } - return out - default: - return nil - } -} - -// sliceContains reports whether s is present in list. -func sliceContains(list []string, s string) bool { - for _, v := range list { - if v == s { - return true - } - } - return false -} - -// firstNonEmptyStr returns a when non-empty, otherwise b. -func firstNonEmptyStr(a, b string) string { - if a != "" { - return a - } - return b -} diff --git a/internal/services/channelbridge/connect.go b/internal/services/channelbridge/connect.go new file mode 100644 index 0000000..e4a3cd6 --- /dev/null +++ b/internal/services/channelbridge/connect.go @@ -0,0 +1,173 @@ +package channelbridge + +import ( + "context" + "net/http" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// This file holds the adapter registry and the connect lifecycle of the channel +// bridge (design/08 §0): adapters are registered through Add, dialed by Start +// (each within connectTimeout), torn down by Stop, and skipped by the credential +// gate when they carry no usable configuration. + +// Add registers an adapter by its Name. It is safe to call before Start; adding +// after Start has no effect on already-running adapters. +func (b *Bridge) Add(a adapter.PlatformAdapter) { + if a == nil { + return + } + b.mu.Lock() + defer b.mu.Unlock() + b.adapters[a.Name()] = a +} + +// WebhookHandlers returns the inbound HTTP handlers of every registered adapter +// that exposes an HTTP callback (e.g. *webhook.Adapter), keyed by the endpoint +// path from the adapter's config. The gateway mounts these to receive external +// webhook callbacks into the runtime. A nil map is returned when no adapter +// provides an inbound handler. +func (b *Bridge) WebhookHandlers() map[string]http.Handler { + b.mu.RLock() + defer b.mu.RUnlock() + out := make(map[string]http.Handler) + for _, a := range b.adapters { + wh, ok := a.(webhookAdapter) + if !ok { + continue + } + path := wh.WebhookPath() + if path == "" { + // Adapter is not currently exposing a webhook (e.g. a websocket-mode + // KOOK); skipping avoids mounting a handler at a stale or colliding route. + continue + } + h := wh.InboundHandler() + if h == nil { + continue + } + out[path] = h + } + if len(out) == 0 { + return nil + } + return out +} + +// Start connects every configured adapter, installs the inbound handler, and +// launches the read loops. Adapters without credentials (or whose Connect fails) +// are logged and left disconnected; they never fail startup. Start blocks only +// long enough to connect (bounded by connectTimeout per adapter) and returns. It +// is idempotent: a second call is a no-op. +func (b *Bridge) Start() { + b.mu.Lock() + if b.closed || b.started { + b.mu.Unlock() + return + } + b.ctx, b.cancel = context.WithCancel(context.Background()) + b.started = true + b.mu.Unlock() + + for _, a := range b.snapshot() { + b.connectAndRun(a) + } + b.log.Info("channel bridge started", "connected", b.ConnectedCount(), + "registered", len(b.snapshot())) +} + +// Stop cancels the read loops, disconnects every adapter and waits for in-flight +// message processing to finish. It is idempotent. +func (b *Bridge) Stop(ctx context.Context) { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + b.closed = true + cancel := b.cancel + b.mu.Unlock() + + cancel() + for name, a := range b.snapshot() { + if err := a.Disconnect(ctx); err != nil { + b.log.Warn("channel disconnect", "channel", name, "error", err) + } + } + b.wg.Wait() + b.log.Info("channel bridge stopped") +} + +// IsConnected reports whether the named adapter is currently connected. +func (b *Bridge) IsConnected(name string) bool { + b.mu.RLock() + defer b.mu.RUnlock() + return b.connected[name] +} + +// ConnectedCount reports how many adapters are currently connected. +func (b *Bridge) ConnectedCount() int { + b.mu.RLock() + defer b.mu.RUnlock() + n := 0 + for _, ok := range b.connected { + if ok { + n++ + } + } + return n +} + +// snapshot returns a copy of the registered adapters for a lock-free iteration. +func (b *Bridge) snapshot() []adapter.PlatformAdapter { + b.mu.RLock() + defer b.mu.RUnlock() + out := make([]adapter.PlatformAdapter, 0, len(b.adapters)) + for _, a := range b.adapters { + out = append(out, a) + } + return out +} + +// configured reports whether the adapter carries enough configuration to attempt +// a Connect. It inspects the adapter's own Config() snapshot for the credential +// each platform needs, so a bare (empty) adapter is skipped rather than dialed. +// This is what keeps weixin-ilink from falling into its blocking QR-login flow +// when no bot token is present. +func (b *Bridge) configured(a adapter.PlatformAdapter) bool { + cfg := a.Config() + switch basePlatform(a.Name()) { + case "telegram", "weixin-ilink", "kook": + return cfgStr(cfg, "bot_token") != "" + case "qq-obv11": + if cfgStr(cfg, "mode") == "client" { + return cfgStr(cfg, "remoteURL") != "" + } + return cfgStr(cfg, "listenAddr") != "" + case "webhook": + // Inbound-only: no external credential is required to hold the session. + return true + default: + return true + } +} + +// cfgStr reads a string field from an adapter config map, returning "" when +// absent. +func cfgStr(m map[string]any, key string) string { + if m == nil { + return "" + } + if s, ok := m[key].(string); ok { + return s + } + return "" +} + +// adapter returns the registered adapter for a platform name, or nil. +func (b *Bridge) adapter(platform string) adapter.PlatformAdapter { + b.mu.RLock() + defer b.mu.RUnlock() + return b.adapters[platform] +} diff --git a/internal/services/channelbridge/helpers.go b/internal/services/channelbridge/helpers.go new file mode 100644 index 0000000..b0c7f7d --- /dev/null +++ b/internal/services/channelbridge/helpers.go @@ -0,0 +1,62 @@ +package channelbridge + +import ( + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// This file holds the small package-level helpers shared across the bridge files: +// platform/account naming, the sender label, and generic slice readers. + +// basePlatform returns the bare platform name of a (possibly multi-account) +// adapter name like "telegram:conf0" → "telegram", so the credential gate and the +// reply-routing fallback treat every account of a platform identically. +func basePlatform(name string) string { + if base, _, ok := strings.Cut(name, ":"); ok { + return base + } + return name +} + +// adapterName returns the account-adapter registration name (e.g. "kook:3bef...") +// for the adapter that received a message, or "" when a is nil. The single-account +// convenience path (process) and a platform with no registered adapter can pass a nil +// adapter, so the persist hooks tolerate it and store an empty adapter_id rather than +// dereferencing a nil adapter. +func adapterName(a adapter.PlatformAdapter) string { + if a == nil { + return "" + } + return a.Name() +} + +// senderDisplayName returns the human-readable sender name used to label a +// third-party message. It prefers the resolved actor's display name (which the +// channel layer derived from channel metadata + the raw SenderName), falling back +// to the message's own SenderName, so an empty SenderName always means "the +// owner/unlabeled" rather than a nickname that resolved to nothing. +func senderDisplayName(msg adapter.UnifiedMessage, actor adapter.Actor) string { + if actor.DisplayName != "" { + return actor.DisplayName + } + return msg.SenderName +} + +// sliceContains reports whether s is present in list. +func sliceContains(list []string, s string) bool { + for _, v := range list { + if v == s { + return true + } + } + return false +} + +// firstNonEmptyStr returns a when non-empty, otherwise b. +func firstNonEmptyStr(a, b string) string { + if a != "" { + return a + } + return b +} diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go new file mode 100644 index 0000000..b96be5c --- /dev/null +++ b/internal/services/channelbridge/inbound.go @@ -0,0 +1,282 @@ +package channelbridge + +import ( + "context" + "errors" + "strings" + + "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" +) + +// This file holds the inbound path of the channel bridge (design/08 §1-§2): the +// per-adapter read loop, the bounded inbound callback, and the full +// session→actor→loop→reply pipeline including the structural identity boundary. + +// readLoop runs an adapter's blocking read path until the bridge stops or the +// adapter's own context is cancelled (a hot reload). The emit sink is wired to +// onMessage so poll-style adapters (Telegram getUpdates, weixin long poll) +// deliver through the same pipeline as push-style adapters. Message processing +// uses the bridge-lifetime context (not readCtx) so a reload leaves in-flight +// messages alone. +func (b *Bridge) readLoop(a adapter.PlatformAdapter, readCtx context.Context) { + defer b.wg.Done() + a.ReadMessages(readCtx, func(msg adapter.UnifiedMessage) { + b.onMessage(b.ctx, msg, a) + }) +} + +// onMessage is the adapter-facing inbound callback. It runs the pipeline in a +// bounded goroutine so the adapter's read loop never blocks on a slow agent turn. +// The adapter is carried explicitly so the pipeline knows which account received +// the message (multi-account reply routing + per-account allow/deny, design/16 +// §4.6 §5). +func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + b.mu.Unlock() + + if err := msg.Validate(); err != nil { + b.log.Warn("inbound message invalid, dropped", "platform", msg.Platform, "error", err) + return + } + + // design/12 §6 策略三/四: per-adapter minimal-disturbance gate. Owner-private + // messages bypass it entirely (the owner is never throttled or pruned); every + // other message passes a sliding-window rate limit (策略三) and the bounded + // pending set whose saturation triggers the review model to prune noise (策略四). + // Dropped messages are audited through the gate's OnPruned hook (wired to + // OnRejected) and never reach the pipeline, so a flood cannot saturate the loop. + actor := session.ResolveActor(msg, b.owner) + if !actor.IsOwner() { + verdict := b.gateFor(a).Admit(ctx, msg) + if !verdict.Proceed { + b.log.Warn("inbound message dropped by flow gate", + "platform", msg.Platform, "channel", msg.ChannelID, + "reason", verdict.Reason, "dropped", len(verdict.Pruned)) + return + } + } + + select { + case b.sem <- struct{}{}: + case <-ctx.Done(): + return + } + + // Add the worker under the same lock that Stop uses to set closed, so a + // concurrent Stop either sees our Add before it waits, or we observe closed + // and bail without leaking a goroutine. + b.mu.Lock() + if b.closed { + b.mu.Unlock() + <-b.sem + return + } + b.wg.Add(1) + b.mu.Unlock() + + go func() { + defer b.wg.Done() + defer func() { <-b.sem }() + b.processFrom(ctx, msg, a) + }() +} + +// process runs the inbound pipeline using the adapter whose platform matches the +// message (single-account convenience path; the adapter may be a bare +// platform-named registration). +func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) { + b.processFrom(ctx, msg, b.adapter(msg.Platform)) +} + +// processFrom runs the full inbound pipeline against the named account adapter: +// resolve the session/actor, apply the allow/deny black-or-white lists, drive the +// agent loop with the chat text, and send the loop's reply back through the same +// adapter that received the message. Identity is resolved structurally by the +// session layer; the bridge never reads message content to decide who the sender +// is (a project red line). +func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) { + text := strings.TrimSpace(msg.Content) + if text == "" { + b.log.Debug("inbound message has no chat text, skipped", + "platform", msg.Platform, "channel", msg.ChannelID) + return + } + + // design/16 §4.6 §6: the black-or-white lists are enforced at the very head of + // the pipeline, before a session is resolved or the loop is driven, so a + // blacklisted/non-whitelisted group or user message never reaches the LLM. The + // owner is always allowed (structural, via the resolved actor). + actor := session.ResolveActor(msg, b.owner) + if actor.IsOwner() { + // fall through: owner messages are never blacklisted + } else if denied, reason := b.blacklisted(msg, a); denied { + b.log.Warn("inbound message blacklisted, dropped", + "platform", msg.Platform, "channel", msg.ChannelID, "reason", reason) + if b.onRejected != nil { + b.onRejected(msg, reason) + } + return + } + + sess, err := b.sessions.GetOrCreate(ctx, msg, b.owner) + if err != nil { + b.log.Warn("session resolve failed", "platform", msg.Platform, + "channel", msg.ChannelID, "error", err) + return + } + + b.log.Debug("inbound → session", + "platform", msg.Platform, "channel", msg.ChannelID, + "session", sess.SessionID, "actor_role", sess.ActorRole, + "env", sess.Env(), "mem_writable", sess.MemWritable) + + // design/12 §6 策略一/二/五: minimal-disturbance gate for non-owner messages. A + // non-owner group/stranger message that the cheap light judgement deems not + // worth a response is registered to its isolated session log and silenced — it + // never drives the full agent loop. Owner messages are never gated here, so + // owner-private chat and owner group @/high-value replies are unchanged. + if !actor.IsOwner() { + if ok, reason := b.shouldRespond(ctx, msg, actor, sess); !ok { + b.registerSilenced(ctx, msg, actor, sess, a) + b.log.Debug("non-owner message silenced (minimal disturbance)", + "platform", msg.Platform, "channel", msg.ChannelID, + "session", sess.SessionID, "reason", reason) + return + } + } + + // 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 their own platform log. + sessID := sess.SessionID + isOwnerPrivate := !sess.IsGroup && sess.Owner + platform := "" + if isOwnerPrivate { + sessID = history.MainSessionID + platform = msg.Platform + } + + // The third-party boundary only applies in a restricted (group/stranger) + // context: memory is read-only (MemWritable=false), and the sender name is + // carried so the agent can see who is speaking (design/16 §4.6 §1-2). The + // resolved actor decides who is the owner vs a third party structurally — never + // from prompt text. + runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable} + senderName := senderDisplayName(msg, actor) + loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName, msg) + + // Persist the inbound user message before the turn so it is durable even if the + // channel reply fails. Owner-private messages land in the main web session and + // are broadcast to the Web client; restricted messages land in their isolated + // session (the adapter log page) and are broadcast as a `channel_persisted` event + // (never into the main/`web` session or the web chat). + if isOwnerPrivate { + b.persistToMain(ctx, history.RoleUser, text, "", platform, deviceIDFor(msg), adapterName(a)) + } else { + b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a)) + } + + res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, loopMsgs) + if err != nil { + b.log.Warn("agent run failed", "platform", msg.Platform, + "session", sessID, "error", err) + return + } + + reply := strings.TrimSpace(res.Text) + if reply == "" { + b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID) + return + } + // Persist the assistant reply (owner-private → main session + message_persisted + // broadcast; restricted → isolated log + channel_persisted broadcast) then + // deliver it back to the channel. + if isOwnerPrivate { + b.persistToMain(ctx, history.RoleAssistant, reply, res.Mood, platform, deviceIDFor(msg), adapterName(a)) + } else { + b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "", adapterName(a)) + } + if err := b.sendReply(ctx, msg, a, reply); err != nil { + if errors.Is(err, adapter.ErrUnsupported) { + b.log.Debug("channel does not support outbound reply, skipped", + "platform", msg.Platform, "session", sessID) + return + } + b.log.Warn("send reply failed", "platform", msg.Platform, "error", err) + } +} + +// 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 +} + +// buildLoopMessages builds the model-visible message list for one inbound turn. It +// decorates the user message with a "[来源: · ]" prefix so +// the model can see which platform and conversation the message came from (design/16 +// §4.6.1); a pure Web main-session message (platform ""/"web") is left untouched. In +// a restricted (group/stranger) context it additionally prepends a single system +// message that states the third-party identity boundary — "every IM non-admin +// account's message is a third party, unrelated to the owner" (design/16 §4.6 §1). +// The boundary and the "[第三方 · 昵称]" marker are decided structurally from the +// resolved SessionEnv/actor, never inferred by the LLM. The raw message text is left +// untouched (a copy is built with the prefix), so persisted content stays consistent +// with what the source platform delivered. +func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string, src adapter.UnifiedMessage) []model.Message { + msg := model.TextMessage(model.RoleUser, text) + thirdParty := env.Restricted() && !actor.IsOwner() && senderName != "" + if thirdParty { + // The speaker is carried as structured metadata, NOT woven into the user text + // (design/16 §4.6.1): a text tag is both ambiguous and a prompt-injection surface. + msg.SenderName = senderName + } + + // origin system line: conversation origin + identity-trust rule, one per turn. + origin := originSystemText(src, actor) + + if !env.Restricted() { + if origin != "" { + return []model.Message{model.TextMessage(model.RoleSystem, origin), msg} + } + return []model.Message{msg} + } + // Restricted (group/stranger): origin (if IM) + the third-party identity boundary. + var sys string + switch { + case origin != "" && thirdPartyBoundaryPrompt != "": + sys = origin + "\n" + thirdPartyBoundaryPrompt + case origin != "": + sys = origin + default: + sys = thirdPartyBoundaryPrompt + } + return []model.Message{model.TextMessage(model.RoleSystem, sys), msg} +} + +// thirdPartyBoundaryPrompt is the system message prepended to a restricted +// (group-chat / stranger) context so the model knows the identity boundary. It is +// only injected when the session is structurally restricted (design/16 §4.6 §1); +// the LLM never decides who is a third party — that was already resolved from the +// actor metadata before this message exists. +const thirdPartyBoundaryPrompt = "你正在 IM 平台的群聊/第三方会话中。每个 IM 的「非管理员账户」发送的消息 = 第三方人员，和主人没有半毛钱关系，不要把它们当成主人的话、也不要按主人对待。只有未标注昵称的消息才是管理员（主人）发言。" diff --git a/internal/services/channelbridge/outbound.go b/internal/services/channelbridge/outbound.go new file mode 100644 index 0000000..7c0f008 --- /dev/null +++ b/internal/services/channelbridge/outbound.go @@ -0,0 +1,29 @@ +package channelbridge + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// This file holds the outbound reply path (design/16 §4.6 §5): the loop's text is +// delivered back to the source conversation through the same account adapter that +// received the message. + +// sendReply delivers the loop's text back to the channel's source conversation +// through the account adapter that received the message (multi-account safe, +// design/16 §4.6 §5). It mirrors the inbound message's channel targeting fields so +// the adapter resolves the same peer the message came from, and it exits through the +// shared IM bubble dispatcher (strip role-play/narration +

split + 0.5-2s gap). +func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, a adapter.PlatformAdapter, text string) error { + if a == nil { + a = b.adapter(in.Platform) + } + if a == nil { + return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform) + } + return b.sendIMBubbles(ctx, in, func(ctx context.Context, msg adapter.UnifiedMessage) error { + return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}}) + }, text) +} diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index 907c61d..c39e973 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -57,14 +57,17 @@ func (b *Bridge) notifyPersisted(msg history.Message) { // concrete client that sent the message (design/16 §4.5) — e.g. "weixin-ilink:conf0:wx_uid" // for IM. adapterID is the account adapter registration name that received the message // (e.g. "kook:3bef..."), so the admin log can query a per-adapter send/receive stream; -// it is empty for Web-originated messages. 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, adapterID string) { +// it is empty for Web-originated messages. mood is the persona's mood for that turn +// (design/01 §2 情绪状态机), carried on the assistant reply only (design/16 §3.3: the +// assistant message carries mode/mood); a user message and an untracked mood pass "". +// 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, mood, platform, deviceID, adapterID 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, AdapterID: adapterID, Mode: modeChat, + SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat, }) if err != nil { b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID, "adapter", adapterID) diff --git a/internal/services/channelbridge/proactive.go b/internal/services/channelbridge/proactive.go index a198329..3575a0a 100644 --- a/internal/services/channelbridge/proactive.go +++ b/internal/services/channelbridge/proactive.go @@ -73,7 +73,7 @@ func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) // §4.4 + §4.6.1). Without this the model would not see what it proactively said, so // a follow-up owner reply would feel incoherent (the message "loses" its history). // It is the same mirror path as owner-private IM replies, scoped to MainSessionID. - b.persistToMain(ctx, history.RoleAssistant, msg.Content, target.platform, + b.persistToMain(ctx, history.RoleAssistant, msg.Content, "", target.platform, fmt.Sprintf("%s:%s:%s", target.platform, target.accountID, target.accountID), target.platform) return nil diff --git a/internal/services/gateway/mood_test.go b/internal/services/gateway/mood_test.go new file mode 100644 index 0000000..05eeddb --- /dev/null +++ b/internal/services/gateway/mood_test.go @@ -0,0 +1,103 @@ +package gateway + +import ( + "context" + "io" + "log/slog" + "testing" + + "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" +) + +// moodRunner is a TurnRunner that settles one turn with a fixed result, so the +// test can assert the mood the loop reported is what gets persisted. +type moodRunner struct { + mood string +} + +func (m moodRunner) Run(_ context.Context, _ string, _ []model.Message, _ loop.SessionEnv) (*loop.Result, error) { + return &loop.Result{Text: "在的呀♪", Steps: 1, StopReason: "stop", Mood: m.mood}, nil +} + +// TestOrchestratorPersistsTurnMood locks the mood column end to end on the Web +// path: the mood the loop's turn-boundary hook reported lands on the persisted +// assistant message (design/16 §3.3 `mode/mood/thinking`), so the DTO, +// `message_persisted` and history replay carry real data instead of an always +// empty mood. The user message keeps an empty mood (design/16 §3.3 assigns the +// mood to the assistant reply). +func TestOrchestratorPersistsTurnMood(t *testing.T) { + t.Parallel() + hist := &fakeHistory{} + o := NewOrchestrator(OrchestratorOptions{ + Run: moodRunner{mood: "happy"}.Run, + Log: slog.New(slog.NewTextHandler(io.Discard, nil)), + History: hist, + }) + + sr, res := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("谢谢你")}, loop.OwnerEnv()) + if !sr.Accepted || res == nil { + t.Fatalf("submit = %+v res=%v, want accepted with a result", sr, res) + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + if recs[1].Role != history.RoleAssistant || recs[1].Mood != "happy" { + t.Fatalf("assistant record = %+v, want Mood happy", recs[1]) + } + wire := toHistoryMessage(recs[1]) + if wire.Mood != "happy" { + t.Fatalf("wire mood = %q, want happy (message_persisted payload)", wire.Mood) + } + if recs[0].Mood != "" { + t.Fatalf("user message mood = %q, want empty", recs[0].Mood) + } +} + +// TestOrchestratorPersistsEmptyMoodWhenUntracked locks the honest degrade: with no +// emotion tracking (no hook wired, so Result.Mood is empty) the stored mood stays +// empty rather than being fabricated. +func TestOrchestratorPersistsEmptyMoodWhenUntracked(t *testing.T) { + t.Parallel() + hist := &fakeHistory{} + o := NewOrchestrator(OrchestratorOptions{ + Run: moodRunner{mood: ""}.Run, + Log: slog.New(slog.NewTextHandler(io.Discard, nil)), + History: hist, + }) + if _, res := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("你好")}, loop.OwnerEnv()); res == nil { + t.Fatal("sync result is nil") + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2", len(recs)) + } + if recs[1].Mood != "" { + t.Fatalf("assistant mood = %q, want empty", recs[1].Mood) + } +} + +// TestServerBroadcastMood locks the `mood` frame the frontend's mood handler reads +// (design/07 §3): the type is `mood`, data.mood is the label, and the persona id / +// affection ride along as extra detail. +func TestServerBroadcastMood(t *testing.T) { + t.Parallel() + env := newEnv(t, nil, false) + c := newFakeConn(env.srv.hub, 32) + env.srv.hub.register(c) + + env.srv.BroadcastMood("cyrene", "happy", 60) + ev := readFakeFrame(t, c) + if ev.Type != "mood" { + t.Fatalf("frame type = %q, want mood (frame=%+v)", ev.Type, ev) + } + data, ok := ev.Data.(map[string]any) + if !ok || data["mood"] != "happy" || data["personaId"] != "cyrene" { + t.Fatalf("frame data = %+v, want {mood:happy,personaId:cyrene}", ev.Data) + } + if got, ok := data["affection"].(float64); !ok || got != 60 { + t.Fatalf("frame affection = %v, want 60", data["affection"]) + } +} diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 5626283..30aca2d 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -681,6 +681,30 @@ func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool str }) } +// wsMoodEvent is the WS event type carrying the persona's mood after a turn +// (design/07 §3 下行 `mood`; design/01 §2 情绪状态机). The frontend's existing +// `mood` handler (ws.ts) reads data.mood and shows the 心情 badge. +const wsMoodEvent = "mood" + +// BroadcastMood fans one emotion change to every Web client as a `mood` frame +// (design/07 §3). It is called only when the mood actually changed — the +// machinery in the composition root is the gate — so the badge does not flicker +// with an unchanged value on every turn. data matches what the frontend `mood` +// handler reads ({mood}); affection/personaId ride along as the honest extra +// detail (affection is the persona's current 好感度, personaId the persona whose +// mood moved) and are ignored by clients that do not render them. The mood is a +// persona-level value, not a per-session one, so no sessionId is attached. +func (s *Server) BroadcastMood(personaID, mood string, affection int) { + s.hub.broadcast(wsEvent{ + Type: wsMoodEvent, + Data: map[string]any{ + "mood": mood, + "personaId": personaID, + "affection": affection, + }, + }) +} + // 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_persist.go b/internal/services/gateway/session_persist.go index 08bcbf0..4a0ecb2 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -78,14 +78,17 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string // 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) and mode is the // behavior mode of that turn (design/16 §5.1), falling back to modeChat when -// empty. 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 +// empty. Mood is the persona's mood for that turn (design/01 §2 情绪状态机), taken +// off the loop result, so the stored/DTO/`message_persisted` mood is real data +// rather than a permanently empty column; a turn with no mood tracking stores an +// empty mood. 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, mode 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, DeviceID: deviceID, Mode: modeOrDefault(mode)}) + full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, Mood: res.Mood, DeviceID: deviceID, Mode: modeOrDefault(mode)}) if err != nil { o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID, "device_id", deviceID) return