From 5fbe9a6d1a88ae39e5edc25c4255d451edbf04bd Mon Sep 17 00:00:00 2001 From: yibo Date: Mon, 25 May 2026 01:28:43 +0800 Subject: [PATCH] feat(stream + docs): /v1/runs SSE 流式 + interim_assistant_callback + 文档更新 --- server/services/wechat/HANDOFF.md | 118 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------- server/services/wechat/auto-reply.js | 127 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------- 2 files changed, 209 insertions(+), 36 deletions(-) diff --git a/server/services/wechat/HANDOFF.md b/server/services/wechat/HANDOFF.md index 3c1f1d8..53e56a4 100644 --- a/server/services/wechat/HANDOFF.md +++ b/server/services/wechat/HANDOFF.md @@ -3,7 +3,7 @@ > **谁会读这份文档**：刚接手这台 Pixel 部署的 claude-run / 开发者，没有上下文。 > **目标**：5 分钟搞清楚正在跑什么、文件在哪、怎么改、怎么验证。 > -> **生成时间**：2026-05-24，由 Mac 端 Claude 部署后写入。请定期核对实际状态， +> **生成时间**：2026-05-25（v2，加 FSM + 流式 + hermes 补丁章节）。请定期核对实际状态， > 因为本文件不会随代码自动更新。 --- @@ -261,17 +261,21 @@ pkill -f auto-reply.js ## 七、已知待办（任选） +**今日已完成**（5/24）：群昵称查询 / A11y 长按引用 + @ 回复 / wake-lock + keepalive 守护 / 清理老凭据 / Frida 残留清理 / A11y FSM 重构 / 流式 SSE / hermes 补丁。 + 按优先级： -1. **群聊昵称查询**：`android-wechat-tracker-sqlcipher.js` 的 `getContactNickname` 对 `@chatroom` 后缀直接 return wxid，导致 A11y 搜不到群名。 - - 改法：删掉 `if (wxid.endsWith('@chatroom')) return wxid;` 分支，让它走通用 rcontact 查询 -2. **群聊引用 + @ 回复**：扩 `HermesAccessibilityService.java`，加长按消息 → 选"引用" → 输入 `@` → 选成员的 UI 自动化 -3. **`termux-wake-lock` 启用**：保证息屏深 Doze 时 `termux-notification-list` 仍响应。 - - `am startservice -a com.termux.service_wake_lock com.termux/.app.TermuxService` - - 写进 `keepalive.sh` -4. **清理 auth.json 老凭据**：删 `credential_pool["kimi-coding"]` 里 last_status=exhausted 的项 -5. **杀 Frida 残留**：`pkill frida-server` + `rm -rf ~/.frida_queue/` -6. **重命名/移除已废弃文件**：`android-wechat-sender-frida.js` 等已加废弃注释，可改名为 `*.deprecated.js` 或直接删 +1. **A11y service 重命名**：`android-wechat-sender-frida.js` 等已加废弃注释，可改名为 `*.deprecated.js` 或删除 +2. **Kimi content 沉默问题**：模型在 hermes agentic mode 下 `content` 永远空，所有内容在 `reasoning_content`，导致 `interim_assistant_callback` 无法 fire。当前两条路： + - 短期：开 `STREAM_REASONING=true`，把 reasoning 加 `🧠` 前缀作为思考过程发给用户 + - 长期：改 hermes system prompt 或 SOUL.md，让 Kimi 在 content 字段写"对用户的话"而不是只塞 reasoning_content +3. **群里长消息分段**：单条 utterance > 500 字会被截断（`maxReplyLength`），应该改成自动分段成多条 +4. **照片 / 文件消息**：tracker 只识别 type=1 文本，type=3 (图片) 等忽略 +5. **群聊消息去重 / 合流**：短时间内多条新消息 → bot 可能每条都触发回复，体验差。需要 debounce + 合并 +6. **流式重试 / 部分输出落地**：当前 run 失败丢全部进度，应该让已 emit 的 utterance 保留为状态 +7. **多模型 fallback**：当前只 Moonshot CN。Anthropic OAuth 还在 auth.json 但已 401。补一个有效的 fallback provider +8. **Hermes SOUL.md 优化**：当前是默认描述，可针对微信场景定制 prompt 让回复更"小火子" +9. **A11y service 故障自愈**：发送失败时（如 chat 找不到）能不能更智能恢复（如点 BACK / 切微信 tab） --- @@ -292,6 +296,9 @@ pkill -f auto-reply.js | `[HERMES] API timeout` / `socket hang up` 在 turns=9 多上下文场景下偶发 | Moonshot kimi-k2.6 长 context LLM call 可达 60s+，旧的 `HERMES_TIMEOUT=120000` 偶发不够 | 启动 env 设 `HERMES_TIMEOUT=180000`（3 分钟），keepalive.sh 已内联 | | A11y 状态机 `CHAT_LIST → click → WRONG_CHAT → BACK` 循环死磕直到 cap=6 给出 | 群聊页 `obn` (id=`/obn`) 标题带成员数后缀（如 `"宜创AI全员群(60)"`），而 broadcast 传的 `toUserDisplayName` 是 `"宜创AI全员群"`（chatroom.displayname 不带后缀），`equals` 永远 false | `classifyState` 拿 `obn.text` 后用正则 `\\(\\d+\\)$` 剥成员数后缀再比对（带/不带都接受） | | WeChat `LauncherUI` 不在主"会话" tab，`FLAG_ACTIVITY_CLEAR_TASK` 也无法回去 | LauncherUI 是 fragment 容器，CLEAR_TASK 重置 task 但不重置内部 fragment 状态。从聊天页或其它主 tab 调 startActivity 不会跳回会话列表 | 状态机的 `NO_CHAT_VIEW` 处理优先 `findBottomNavTab("微信")` 点击（按 Y 坐标取最下方那个，区分标题栏同名 text），fallback `GLOBAL_ACTION_BACK` | +| Hermes `/v1/runs` 长时间一片寂静、无任何 `[UTTER]` 事件输出，最后 `run.completed` 一次性给一大段总结 | Kimi-k2.6 在 hermes agentic mode 下 `assistant_message.content` 永远空，所有内容在 `reasoning_content`；`message.delta`/`interim_assistant_callback`/`message.complete` 都不 fire；只有 `reasoning.delta` 大量喷涌 | 短期开 `STREAM_REASONING=true` 让 reasoning 加 `🧠` 前缀作 utterance；长期改 hermes system prompt 让 Kimi 写 content 字段。详见 §十二.4 | +| auto-reply node 进程被 keepalive 反复重启，CPU 占用高、log 文件被洪水般写入 | 流式场景下高频 `message.delta`/`reasoning.delta` 事件触发同步 `fs.appendFileSync` + 每条都 console.log，sync I/O 卡死 node；keepalive 检测心跳超时 → kill → restart | `log()` 去掉 `appendFileSync`（stdout 已由 start 脚本重定向到 log file）；诊断事件加节流（高频事件 5 秒打一次） | +| Hermes `/v1/runs` 只输出"最终一条回复"，没多步进度 | hermes 默认 `/v1/runs` 只挂 `stream_delta_callback` + `tool_progress_callback`，缺 `reasoning_callback` + `interim_assistant_callback`（gateway 平台用 7+ 个回调） | 补丁 `gateway/platforms/api_server.py`，在 `_create_agent` 签名 + AIAgent 构造 + run handler 三处都加上缺的两个回调，emit `reasoning.delta` / `message.complete` SSE 事件。详见 §十二.3 | --- @@ -341,4 +348,95 @@ State reset --- +## 十一、A11y 状态机（5/24 重构） + +旧代码用嵌套 if + ad-hoc relaunch 处理"目标聊天 / 错误聊天 / 主界面 / 微信不在前台"等多种状态，边角情况覆盖不全（特别是 LauncherUI fragment 容器 + CLEAR_TASK 不重置内部 fragment）。 + +新实现：**显式状态机**（`HermesAccessibilityService.java::driveStateMachine`） + +| 状态 | classifier 特征 | 动作 | +|------|-----------------|------| +| `TARGET_CHAT` | `bkk` + `obn.text` 等于 displayName（剥末尾 `(数字)` 后） | `performTypeAndSend` 直走 | +| `WRONG_CHAT` | `bkk` + `obn.text` 不匹配 | `GLOBAL_ACTION_BACK` → 再 classify | +| `CHAT_LIST` | 无 `bkk`，`kbq` 多个 | `findChatListRow` 点 cj1 → 再 classify | +| `NO_CHAT_VIEW` | 无 `bkk` 无 `kbq` 但在 com.tencent.mm | `findBottomNavTab("微信")` 点底部 tab，找不到就 BACK | +| `OUTSIDE_WECHAT` | `root.pkg != com.tencent.mm` | `startActivity(LauncherUI)` → 再 classify | + +- 入口：`onReceive` 拉 LauncherUI 后 3.5s 调 `driveStateMachine(0)` +- `MAX_STATE_TRANSITIONS = 6` 防死循环 +- 关键修复：**chat list 优先于搜索**（避开"丁鹏、🍋火星人、火星人"被搜索解读成"创建包含 3 人的新群"歧义） +- 关键修复：**obn 标题剥末尾 `(\d+)`**（群聊页 obn 是"宜创AI全员群(60)" 含成员数后缀） + +--- + +## 十二、流式多步消息（/v1/runs SSE，5/24 实现 + hermes 补丁） + +### 12.1 架构 + +``` +auto-reply.js hermes (/v1/runs) Moonshot CN + │ │ │ + ├─POST /v1/runs─────────────────→│ 返回 run_id │ + │ { input, conversation_history,│ │ + │ session_id, instructions } │ │ + │←───────────── run_id ──────────┤ │ + │ │ │ + ├─GET /v1/runs/{id}/events ─────→│ 长连 SSE │ + │ │←──API call #1───────────┤ + │←─message.delta / reasoning.delta / │ + │ tool.started / message.complete │ + │ │ ...iterating tools... │ + │ │←──API call #N───────────┤ + │←──run.completed ───────────────┤ │ +``` + +每个事件触发一段独立 utterance（微信单独发送一条消息），实现"hermes 多步思考 → 微信多条回复"。 + +### 12.2 SSE 事件类型（按消费优先级） + +| 事件 | payload | auto-reply 处理 | +|------|---------|----------------| +| `message.complete` | `{ text, already_streamed }` | **主路径**：每 turn 结束 / tool 前，emit text 作为一条 utterance（已 strip `` 块，是用户可见内容） | +| `message.delta` | `{ delta }` | 累积到 textBuf，tool.started 时兜底 flush | +| `reasoning.delta` | `{ text }` | 默认丢弃（模型内心独白）；`STREAM_REASONING=true` 时累积 + 加 `🧠` 前缀 emit | +| `reasoning.available` | `{ text }` | 同上，仅在 reasoning.delta 没累积时用 | +| `tool.started` | `{ tool, preview }` | flush 任何累积；`STREAM_TOOL_PROGRESS=true` 时额外 emit `🔧 工具: 参数` | +| `tool.completed` | `{ tool, duration, error }` | 忽略 | +| `run.completed` | `{ output, usage }` | emit output 作为最终 utterance + resolve Promise | +| `run.failed` | `{ error }` | reject Promise | + +### 12.3 Hermes API server 补丁（已应用到 phone） + +`/v1/runs` **upstream** 默认只挂 `stream_delta_callback`（content tokens）+ `tool_progress_callback`，**缺**： +- `reasoning_callback` → Kimi/Moonshot 的 chain-of-thought 走不出来 +- `interim_assistant_callback` → TUI/Telegram/Discord 用的多步消息机制 + +我们在 `~/hermes-agent/gateway/platforms/api_server.py` 加了两个回调 + 在 `_create_agent` 签名 / AIAgent 构造调用都补了参数。改动以注释 `# Forward reasoning_content...` 和 `# Forward interim assistant messages...` 标识。 + +**改动文件**：phone 上 `~/hermes-agent/gateway/platforms/api_server.py`（备份 `.bak`）。Mac repo 上没同步——hermes-agent 不在我们的 git repo 里。如果 hermes 升级要重新补丁。 + +### 12.4 Kimi-k2.6 在 Hermes agentic mode 的限制 ⚠️ + +实测发现：**Kimi-k2.6 在 hermes 系统提示下，`assistant_message.content` 永远是空字符串**，所有可见内容塞 `reasoning_content`。导致： + +- `message.delta` 计数 = 0（content 字段没 token） +- `interim_assistant_callback` 内部检查 `visible = strip_think(content).strip()` 为空就 return → 不 fire +- `message.complete` 事件也不出现 +- 用户只能在 `run.completed.output` 时看到最终一段总结 + +短期 workaround：**`STREAM_REASONING=true`**（默认 OFF）让 chain-of-thought 加 `🧠` 前缀作为 utterance 发出去。 + +长期方案：改 hermes system prompt / SOUL.md 让 Kimi 把"对用户说的话"写到 content 字段。 + +### 12.5 流式相关 ENV flag + +| 变量 | 默认 | 作用 | +|------|------|------| +| `STREAM_TOOL_PROGRESS` | false | 每个 `tool.started` 发一条 `🔧 工具: 参数` 进度消息 | +| `STREAM_REASONING` | false | 把 Kimi 的 reasoning_content（内心独白）加 `🧠` 前缀作为 utterance 发 | +| `HERMES_TIMEOUT` | `180000` (旧) → `600000` (新) | SSE 长连 timeout，长 agentic run 经常 3-5min | +| `HISTORY_CONTEXT_SIZE` | 8 | 喂给 hermes 的微信历史消息条数 | + +--- + **END OF HANDOFF** — 改动这份文档时记得更新顶部生成时间。 diff --git a/server/services/wechat/auto-reply.js b/server/services/wechat/auto-reply.js index adba284..a29b904 100644 --- a/server/services/wechat/auto-reply.js +++ b/server/services/wechat/auto-reply.js @@ -54,6 +54,15 @@ const CONFIG = { // 上下文长度：每次调 hermes 时回看多少条最近消息（含本次触发的那条） historyContextSize: parseInt(process.env.HISTORY_CONTEXT_SIZE, 10) || 8, + // 流式行为：模型沉默 agentic 调用时，是否发"🔧 调用 X"进度消息 + // 默认 OFF（避免群里刷屏）；开了之后每个 tool.started 都会发一条进度消息 + streamToolProgress: process.env.STREAM_TOOL_PROGRESS === 'true', + + // 是否把 reasoning_content（模型内心独白）作为消息发给用户 + // 默认 OFF — reasoning 是模型自言自语（"用户要求..." 第三人称）， + // 直接发给用户极其诡异。开了之后会带 "🧠 " 前缀作为思考进度发。 + streamReasoning: process.env.STREAM_REASONING === 'true', + // 通知监听（息屏兜底） notificationPollIntervalMs: 3000, notificationEnabled: process.env.NOTIFICATION_ENABLED !== 'false', @@ -125,10 +134,9 @@ class ReplyState { function log(...args) { const line = `[${new Date().toISOString()}] ${args.join(' ')}`; + // start-ar.sh 已经把 stdout 重定向到 auto-reply.log，console.log 即写入文件。 + // 不再额外 fs.appendFileSync — 同步 I/O 在高频流式场景下会卡死 node。 console.log(line); - try { - fs.appendFileSync(CONFIG.logFile, line + '\n'); - } catch {} } // ═══════════════════════════════════════════════════════════════ @@ -207,34 +215,57 @@ async function callHermesRunStream(messages, sessionId, onUtterance) { timeout: CONFIG.hermesTimeoutMs, }, res => { let buf = ''; - let textBuf = ''; + let textBuf = ''; // 累积 message.delta（assistant content tokens） + let reasoningBuf = ''; // 累积 reasoning.delta（Kimi 的 chain-of-thought tokens） let utteranceIdx = 0; - let sendChain = Promise.resolve(); // 串行队列：每个 flush 在前一个完成后执行 + let sendChain = Promise.resolve(); // 串行队列：每个 utterance 在前一个完成后执行 let finalResult = {}; - const flush = (isFinal) => { - const text = textBuf.trim(); - textBuf = ''; - if (!text) return; - const idx = utteranceIdx++; - // 关键：链式 append，不等待 - sendChain = sendChain - .then(() => onUtterance(text, idx, !!isFinal)) - .catch(e => log('[onUtterance error]', e.message)); - }; - - // diagnostic counters - const evCount = { 'message.delta': 0 }; - let lastDiagLogAt = Date.now(); + // diagnostic counters — message.delta / reasoning.delta 都是高频 token 流， + // 必须节流否则 600+ events × sync 文件 IO 会卡死 node + const evCount = {}; + let lastDiagLogAt = 0; + const HIGH_FREQ_EVENTS = new Set(['message.delta', 'reasoning.delta']); const diagLog = (evt) => { evCount[evt] = (evCount[evt] || 0) + 1; const now = Date.now(); - if (evt !== 'message.delta' || now - lastDiagLogAt > 10000) { - log('[STREAM EV]', evt, 'counts=', JSON.stringify(evCount), 'textBuf.len=' + textBuf.length); + const isHighFreq = HIGH_FREQ_EVENTS.has(evt); + if (!isHighFreq || now - lastDiagLogAt > 5000) { + log('[STREAM EV]', evt, 'counts=', JSON.stringify(evCount), + 'textBuf.len=' + textBuf.length, 'reasoningBuf.len=' + reasoningBuf.length); lastDiagLogAt = now; } }; + // 跟踪上一次实际发出去的文本，避免 reasoning.available 和 run.completed 重复发同样内容 + let lastEmitted = ''; + const emit = (text, isFinal) => { + const trimmed = (text || '').trim(); + if (!trimmed) return; + // 简单 dedupe：完全相同 / 完全包含 / 被包含都跳过 + if (trimmed === lastEmitted) return; + if (lastEmitted && (trimmed.includes(lastEmitted) || lastEmitted.includes(trimmed))) return; + lastEmitted = trimmed; + const idx = utteranceIdx++; + sendChain = sendChain + .then(() => onUtterance(trimmed, idx, !!isFinal)) + .catch(e => log('[onUtterance error]', e.message)); + }; + + // 重新定义 flush 让它走 emit 通道 + const flushTextBuf = (isFinal) => { + const t = textBuf.trim(); + textBuf = ''; + if (t) emit(t, isFinal); + }; + + // 把累积的 reasoning（Kimi chain-of-thought）作为一段独立 utterance flush + const flushReasoningBuf = (isFinal) => { + const t = reasoningBuf.trim(); + reasoningBuf = ''; + if (t) emit(t, isFinal); + }; + res.setEncoding('utf-8'); res.on('data', chunk => { buf += chunk; @@ -248,17 +279,61 @@ async function callHermesRunStream(messages, sessionId, onUtterance) { try { ev = JSON.parse(line.substring(6)); } catch (_) { continue; } diagLog(ev.event); switch (ev.event) { + case 'message.complete': + // 🎯 主路径：interim_assistant_callback（gateway 平台用的事件） + // 在每个 turn 末尾、tool 之前 fire，携带模型本轮**给用户看的** + // 可见 content（已 strip 块）。这是真正的"多步消息"信号。 + // already_streamed=true 表示 content 已通过 message.delta 流过，避免重复 + if (!ev.already_streamed) { + emit(ev.text, false); + // 同 turn 的 message.delta 已被 interim 覆盖，重置 textBuf + textBuf = ''; + } + // reasoning 同 turn 的 chain-of-thought 已落地，清掉避免泄露给用户 + reasoningBuf = ''; + break; case 'message.delta': + // 模型 content tokens — 主要靠 message.complete 拿干净文本， + // message.delta 累积仅做兜底（如果 interim 没 fire） textBuf += (ev.delta || ''); break; + case 'reasoning.delta': + // Kimi/Moonshot 的 chain-of-thought tokens — 模型内心独白， + // 默认不发给用户（会出现第三人称"用户要求..."的诡异内容）。 + // STREAM_REASONING=true 时累积，flush 时加 "🧠 " 前缀 + if (CONFIG.streamReasoning) { + reasoningBuf += (ev.text || ''); + } + break; + case 'reasoning.available': + // 整段 turn 末尾的 thinking，同样默认不发 + if (CONFIG.streamReasoning && !reasoningBuf.trim() && ev.text) { + emit('🧠 ' + ev.text, false); + } + break; case 'tool.started': - // turn 边界：上一段 assistant 文本完成 - flush(false); + // turn 边界兜底：interim_assistant_callback 已 emit 过本轮 content， + // textBuf 仅在 interim 未触发时用 + if (CONFIG.streamReasoning) { + const r = reasoningBuf.trim(); + reasoningBuf = ''; + if (r) emit('🧠 ' + r, false); + } + flushTextBuf(false); + if (CONFIG.streamToolProgress) { + const toolName = ev.tool || '?'; + const preview = (ev.preview || '').toString().split('\n')[0].substring(0, 60); + emit(preview ? `🔧 ${toolName}: ${preview}` : `🔧 ${toolName}`, false); + } break; case 'run.completed': - flush(true); + // 收尾：emit 最终 output（如果跟 last interim 重复会被 dedupe 吃掉） + if (ev.output) { + emit(ev.output, true); + } else { + flushTextBuf(true); + } finalResult = { output: ev.output, usage: ev.usage }; - // 等串行队列 drain 完再 resolve sendChain.then(() => resolve(finalResult)); break; case 'run.failed': @@ -267,7 +342,7 @@ async function callHermesRunStream(messages, sessionId, onUtterance) { case 'run.cancelled': reject(new Error('run cancelled')); break; - // tool.completed / reasoning.available / approval.* 忽略 + // tool.completed / approval.* 忽略 } } } -- libgit2 0.21.4