wehub/wechat-gggithub--wechat-claude-code

Files

T

wehub-resource-sync 30b27fe8fb chore: import upstream snapshot with attribution

2026-07-13 12:33:50 +08:00

43 KiB

Raw Permalink Blame History

微信消息分块优化 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 把 Claude CLI 最终答案从"按段落碎片推送"改为"按回合整段推送"，仅在超 4000 字时按段落硬切；agent loop 期间的 interstitial 保持实时段落推送。

Architecture: 在 provider.ts 暴露 onTurnEnd(stopReason) 回调（基于 message_delta 事件的 stop_reason 字段）；新增 src/claude/turn-router.ts 的 TurnRouter 类把 text_delta 按回合累积，根据 stop_reason 分流为 interstitial（立即发）或 final（流结束发）；main.ts 替换原 textBuffer 逻辑，接入 TurnRouter，删除段落边界 flush 相关死代码。

Tech Stack: TypeScript（strict）、Node.js ESM（"type": "module"、Node16 module resolution）、Node 内置 test runner（node --test）。

Global Constraints

不改 src/wechat/api.ts 的限流逻辑（2.5s 间隔、60s 冷却、指数退避保持原样）。
不改 MAX_MESSAGE_LENGTH = 4000、splitMessage、parseBlocks、findSafeSplitPoint、splitByNewline。
不改 typing 指示器、silence warning 5min 兜底（flushTimer）、文件自动推送。
TypeScript strict 模式，编译命令 npm run build（tsc）。
测试入口 npm test = node --test dist/tests/*.test.js，必须先 npm run build。
ESM 导入必须带 .js 后缀（即便源是 .ts）。
提交信息遵循现有风格：type: 中文描述（参考 git log）。

Spec 参考：docs/superpowers/specs/2026-06-20-message-batching-design.md

Task 1: 提取 `handleStreamLine` 到 provider.ts（纯重构，为可测性铺路）

Files:

Modify: src/claude/provider.ts（提取 rl.on('line', ...) 内的 switch 到导出函数）
Create: src/tests/provider.test.ts

Interfaces:

Consumes: 无（首个任务）
Produces:
- export interface StreamParserState { sessionId: string; textParts: string[]; errorMessage?: string; trackingSkill: boolean; skillInputAccum: string; }
- export interface StreamParserCallbacks { onText?: (text: string) => void; onBlockEnd?: () => void; }
- export function handleStreamLine(line: string, state: StreamParserState, callbacks: StreamParserCallbacks): void
Step 1: 写覆盖现有行为的失败测试

Create src/tests/provider.test.ts:

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { handleStreamLine, type StreamParserState } from '../claude/provider.js';

function freshState(): StreamParserState {
  return { sessionId: '', textParts: [], trackingSkill: false, skillInputAccum: '' };
}

test('handleStreamLine: system init 设置 sessionId', () => {
  const state = freshState();
  handleStreamLine(
    JSON.stringify({ type: 'system', subtype: 'init', session_id: 'sess-123' }),
    state,
    {},
  );
  assert.equal(state.sessionId, 'sess-123');
});

test('handleStreamLine: text_delta 触发 onText', () => {
  const calls: string[] = [];
  handleStreamLine(
    JSON.stringify({
      type: 'stream_event',
      event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'hello' } },
    }),
    freshState(),
    { onText: (t) => calls.push(t) },
  );
  assert.deepEqual(calls, ['hello']);
});

test('handleStreamLine: content_block_stop 触发 onBlockEnd', () => {
  let called = 0;
  handleStreamLine(
    JSON.stringify({ type: 'stream_event', event: { type: 'content_block_stop', index: 0 } }),
    freshState(),
    { onBlockEnd: () => called++ },
  );
  assert.equal(called, 1);
});

test('handleStreamLine: assistant 消息文本累积到 textParts', () => {
  const state = freshState();
  handleStreamLine(
    JSON.stringify({
      type: 'assistant',
      message: { content: [{ type: 'text', text: '回复内容' }] },
    }),
    state,
    {},
  );
  assert.deepEqual(state.textParts, ['回复内容']);
});

test('handleStreamLine: 空行和非法 JSON 静默跳过', () => {
  const state = freshState();
  handleStreamLine('', state, {});
  handleStreamLine('not json', state, {});
  handleStreamLine('   ', state, {});
  assert.deepEqual(state.textParts, []);
});

Step 2: 跑测试确认失败（函数不存在）

Run:

npm run build && node --test dist/tests/provider.test.js

Expected: 编译错误 handleStreamLine is not exported 或测试运行报错 cannot find module。

Step 3: 实现提取

In src/claude/provider.ts:

3a. 在文件顶部 imports 之后、claudeQuery 之前，加入类型定义和提取的函数：

// ---------------------------------------------------------------------------
// Stream parser (extracted for testability)
// ---------------------------------------------------------------------------

export interface StreamParserState {
  sessionId: string;
  textParts: string[];
  errorMessage?: string;
  trackingSkill: boolean;
  skillInputAccum: string;
}

export interface StreamParserCallbacks {
  onText?: (text: string) => void;
  onBlockEnd?: () => void;
}

export function handleStreamLine(
  line: string,
  state: StreamParserState,
  callbacks: StreamParserCallbacks,
): void {
  if (!line.trim()) return;
  let obj: any;
  try {
    obj = JSON.parse(line);
  } catch {
    return;
  }

  switch (obj.type) {
    case 'system': {
      if (obj.subtype === 'init' && obj.session_id) {
        state.sessionId = obj.session_id;
      }
      break;
    }
    case 'assistant': {
      const content = obj.message?.content;
      if (Array.isArray(content)) {
        const text = content
          .filter((b: any) => b.type === 'text')
          .map((b: any) => b.text ?? '')
          .join('');
        if (text) state.textParts.push(text);
      }
      break;
    }
    case 'stream_event': {
      const evt = obj.event;
      if (evt?.type === 'content_block_start' && evt.content_block?.type === 'tool_use') {
        if (evt.content_block.name === 'Skill') {
          state.trackingSkill = true;
          state.skillInputAccum = '';
        }
      } else if (evt?.type === 'content_block_delta' && evt.delta?.type === 'text_delta') {
        const delta: string = evt.delta.text;
        if (delta && callbacks.onText) {
          callbacks.onText(delta);
        }
      } else if (evt?.type === 'content_block_delta' && evt.delta?.type === 'input_json_delta' && state.trackingSkill) {
        state.skillInputAccum += evt.delta.partial_json ?? '';
        try {
          const parsed = JSON.parse(state.skillInputAccum);
          if (parsed.skill) {
            const msg = `\n正在调用 ${parsed.skill} 技能\n\n`;
            if (callbacks.onText) callbacks.onText(msg);
            state.trackingSkill = false;
          }
        } catch {
          // JSON not complete yet
        }
      } else if (evt?.type === 'content_block_stop') {
        state.trackingSkill = false;
        if (callbacks.onBlockEnd) callbacks.onBlockEnd();
      }
      break;
    }
    case 'result': {
      if (obj.result && typeof obj.result === 'string') {
        const combined = state.textParts.join('');
        if (!combined.includes(obj.result)) {
          state.textParts.push(obj.result);
        }
      }
      if (obj.subtype === 'error' || (obj.errors && obj.errors.length > 0)) {
        const errors = obj.errors ?? [obj.error_message ?? 'Unknown error'];
        state.errorMessage = Array.isArray(errors) ? errors.join('; ') : String(errors);
        logger.error('CLI returned error result', { errors });
      }
      break;
    }
    default:
      break;
  }
}

3b. 在 claudeQuery 内替换原 rl.on('line', ...) 块。 找到现有的：

    // Parse NDJSON from stdout
    let skillInputAccum = '';
    let trackingSkill = false;

    const rl = createInterface({ input: child.stdout! });
    rl.on('line', (line: string) => {
      if (!line.trim()) return;
      let obj: any;
      try {
        obj = JSON.parse(line);
      } catch {
        // Skip unparseable lines
        return;
      }

      switch (obj.type) {
        case 'system': {
          if (obj.subtype === 'init' && obj.session_id) {
            sessionId = obj.session_id;
          }
          break;
        }
        case 'assistant': {
          const content = obj.message?.content;
          if (Array.isArray(content)) {
            const text = content
              .filter((b: any) => b.type === 'text')
              .map((b: any) => b.text ?? '')
              .join('');
            if (text) textParts.push(text);
          }
          break;
        }
        case 'stream_event': {
          const evt = obj.event;
          if (evt?.type === 'content_block_start' && evt.content_block?.type === 'tool_use') {
            if (evt.content_block.name === 'Skill') {
              trackingSkill = true;
              skillInputAccum = '';
            }
          } else if (evt?.type === 'content_block_delta' && evt.delta?.type === 'text_delta') {
            const delta: string = evt.delta.text;
            if (delta && onText) {
              Promise.resolve(onText(delta)).catch(() => {});
            }
          } else if (evt?.type === 'content_block_delta' && evt.delta?.type === 'input_json_delta' && trackingSkill) {
            skillInputAccum += evt.delta.partial_json ?? '';
            try {
              const parsed = JSON.parse(skillInputAccum);
              if (parsed.skill) {
                const msg = `\n正在调用 ${parsed.skill} 技能\n\n`;
                if (onText) Promise.resolve(onText(msg)).catch(() => {});
                trackingSkill = false;
              }
            } catch {
              // JSON not complete yet, keep accumulating
            }
          } else if (evt?.type === 'content_block_stop') {
            trackingSkill = false;
            if (onBlockEnd) Promise.resolve(onBlockEnd()).catch(() => {});
          }
          break;
        }
        case 'result': {
          if (obj.result && typeof obj.result === 'string') {
            const combined = textParts.join('');
            if (!combined.includes(obj.result)) {
              textParts.push(obj.result);
            }
          }
          if (obj.subtype === 'error' || (obj.errors && obj.errors.length > 0)) {
            const errors = obj.errors ?? [obj.error_message ?? 'Unknown error'];
            errorMessage = Array.isArray(errors) ? errors.join('; ') : String(errors);
            logger.error('CLI returned error result', { errors });
          }
          break;
        }
        default:
          break;
      }
    });

替换为：

    // Parse NDJSON from stdout (logic in handleStreamLine for testability)
    const parserState: StreamParserState = {
      sessionId: '',
      textParts: [],
      trackingSkill: false,
      skillInputAccum: '',
    };
    const parserCallbacks: StreamParserCallbacks = { onText, onBlockEnd };

    const rl = createInterface({ input: child.stdout! });
    rl.on('line', (line: string) => {
      handleStreamLine(line, parserState, parserCallbacks);
    });

3c. 把所有读写 sessionId / textParts / errorMessage 的地方改成操作 parserState.*。 在 claudeQuery 内：

函数开头删除三个局部声明：let sessionId = '';、const textParts: string[] = [];、let errorMessage: string | undefined;（状态现在在 parserState 里）。
timeout handler 内：const partialText = textParts.join('\n').trim(); → parserState.textParts.join('\n').trim();；finish({ ..., sessionId, ... }) → finish({ ..., sessionId: parserState.sessionId, ... })。
onAbort 内：同上两处替换。
child.on('close', ...) 内（5 处替换，含读和写）：
- !textParts.length && !errorMessage → !parserState.textParts.length && !parserState.errorMessage
- errorMessage = stderr || \claude exited with code ${code}`;→parserState.errorMessage = stderr || `claude exited with code ${code}`;`
- const fullText = textParts.join('\n').trim(); → parserState.textParts.join('\n').trim();
- if (!fullText && !errorMessage) → if (!fullText && !parserState.errorMessage)
- errorMessage = 'Claude returned an empty response.'; → parserState.errorMessage = 'Claude returned an empty response.';
- finish({ text: fullText, sessionId, error: errorMessage }) → finish({ text: fullText, sessionId: parserState.sessionId, error: parserState.errorMessage })
- 日志里 textLength: fullText.length 等读取 fullText 的不动（它是局部变量）。
child.on('error', ...) 内：finish({ text: '', sessionId, error: ... }) → finish({ text: '', sessionId: parserState.sessionId, error: ... })。
Step 4: 编译并跑测试确认通过

Run:

npm run build && node --test dist/tests/provider.test.js

Expected: 5 个测试全 PASS，无 TypeScript 编译错误。

Step 5: 端到端冒烟（确认重构没破坏 claudeQuery）

Run:

echo "你好" | node dist/main.js 2>&1 | head -5 || true

Expected: 进程能启动（即使因为没有配置账号/凭证而退出，也不应在 provider.ts 上报 TypeError）。如果输出 未找到账号 之类的运行期错误，说明导入和类型正常。

Step 6: 提交

git add src/claude/provider.ts src/tests/provider.test.ts
git commit -m "$(cat <<'EOF'
refactor: 提取 handleStreamLine 为可测的纯函数

把 claudeQuery 里 rl.on('line') 的 NDJSON 解析 switch 体抽成独立导出函数
handleStreamLine(line, state, callbacks)，状态外置到 StreamParserState。
行为完全不变，仅是为后续 onTurnEnd 接入和单元测试铺路。

附首批单元测试覆盖 system init / text_delta / content_block_stop /
assistant 文本累积 / 非法行跳过 5 条路径。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
EOF
)"

Task 2: 在 provider.ts 加 `onTurnEnd` 回调

Files:

Modify: src/claude/provider.ts（QueryOptions 加字段、handleStreamLine 加分支、StreamParserCallbacks 加字段）
Modify: src/tests/provider.test.ts（新增测试）

Interfaces:

Consumes: Task 1 的 handleStreamLine / StreamParserState / StreamParserCallbacks
Produces:
- QueryOptions.onTurnEnd?: (stopReason: string) => void
- StreamParserCallbacks.onTurnEnd?: (stopReason: string) => void
- handleStreamLine 在收到 message_delta 事件且 delta.stop_reason 存在时触发 onTurnEnd
Step 1: 写失败测试

Append to src/tests/provider.test.ts:

test('handleStreamLine: message_delta 带 stop_reason 触发 onTurnEnd', () => {
  const calls: string[] = [];
  handleStreamLine(
    JSON.stringify({
      type: 'stream_event',
      event: { type: 'message_delta', delta: { stop_reason: 'end_turn' } },
    }),
    freshState(),
    { onTurnEnd: (r) => calls.push(r) },
  );
  assert.deepEqual(calls, ['end_turn']);
});

test('handleStreamLine: message_delta 无 stop_reason 不触发 onTurnEnd', () => {
  const calls: string[] = [];
  handleStreamLine(
    JSON.stringify({
      type: 'stream_event',
      event: { type: 'message_delta', delta: {} },
    }),
    freshState(),
    { onTurnEnd: (r) => calls.push(r) },
  );
  assert.deepEqual(calls, []);
});

test('handleStreamLine: tool_use stop_reason 也正常透传', () => {
  const calls: string[] = [];
  handleStreamLine(
    JSON.stringify({
      type: 'stream_event',
      event: { type: 'message_delta', delta: { stop_reason: 'tool_use' } },
    }),
    freshState(),
    { onTurnEnd: (r) => calls.push(r) },
  );
  assert.deepEqual(calls, ['tool_use']);
});

Step 2: 跑测试确认失败

Run:

npm run build && node --test dist/tests/provider.test.js

Expected: 3 个新测试 FAIL（onTurnEnd 类型不存在或回调不触发），原 5 个 PASS。

Step 3: 实现

3a. 在 QueryOptions 接口加字段（src/claude/provider.ts）：

找到现有的：

  /** Called when a content block ends — use to flush buffered text. */
  onBlockEnd?: () => Promise<void> | void;

在其之后加：

  /** Called when an assistant turn ends, with its stop_reason
   *  ('tool_use' | 'end_turn' | 'max_tokens' | 'stop_sequence' | 'pause_turn' | ...).
   *  Use to decide whether the turn's text is interstitial or final answer. */
  onTurnEnd?: (stopReason: string) => Promise<void> | void;

3b. 在 StreamParserCallbacks 接口加字段（同文件，Task 1 新增的部分）：

找到：

export interface StreamParserCallbacks {
  onText?: (text: string) => void;
  onBlockEnd?: () => void;
}

改为：

export interface StreamParserCallbacks {
  onText?: (text: string) => void;
  onBlockEnd?: () => void;
  onTurnEnd?: (stopReason: string) => void;
}

3c. 在 handleStreamLine 的 stream_event case 加分支。 找到 content_block_stop 分支：

      } else if (evt?.type === 'content_block_stop') {
        state.trackingSkill = false;
        if (callbacks.onBlockEnd) callbacks.onBlockEnd();
      }
      break;

在其之后（仍在 stream_event case 内、break; 之前）插入：

      } else if (evt?.type === 'message_delta' && evt.delta?.stop_reason) {
        if (callbacks.onTurnEnd) callbacks.onTurnEnd(evt.delta.stop_reason);
      }

注意：因为这是 else if 链，要把上面那个 } 闭合改一下。完整片段应该是：

      } else if (evt?.type === 'content_block_stop') {
        state.trackingSkill = false;
        if (callbacks.onBlockEnd) callbacks.onBlockEnd();
      } else if (evt?.type === 'message_delta' && evt.delta?.stop_reason) {
        if (callbacks.onTurnEnd) callbacks.onTurnEnd(evt.delta.stop_reason);
      }
      break;

3d. 把 onTurnEnd 透传到 parserCallbacks。 在 claudeQuery 内找到：

    const parserCallbacks: StreamParserCallbacks = { onText, onBlockEnd };

改为：

    const parserCallbacks: StreamParserCallbacks = { onText, onBlockEnd, onTurnEnd };

Step 4: 跑测试确认通过

Run:

npm run build && node --test dist/tests/provider.test.js

Expected: 8 个测试全 PASS。

Step 5: 提交

git add src/claude/provider.ts src/tests/provider.test.ts
git commit -m "$(cat <<'EOF'
feat: provider 暴露 onTurnEnd 回调，按 stop_reason 标记回合类型

handleStreamLine 在收到 message_delta 事件且带 stop_reason 时触发
onTurnEnd(stopReason)。下游可据此区分 'tool_use' 回合（interstitial）
和 'end_turn' / 'max_tokens' 等终态回合（final answer）。

本任务仅暴露信号，不改变任何现有 flush 行为——main.ts 下一任务接入。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
EOF
)"

Task 3: 创建 `TurnRouter` 状态机

Files:

Create: src/claude/turn-router.ts
Create: src/tests/turn-router.test.ts

Interfaces:

Consumes: 无（独立模块）
Produces:
- export type MessageRole = 'interstitial' | 'final';
- export interface RoutedMessage { text: string; role: MessageRole; }
- export class TurnRouter { constructor(emit: (msg: RoutedMessage) => void); onText(delta: string): void; onTurnEnd(stopReason: string): void; drain(): void; }

行为契约：

onText 累积到内部 turnBuffer，不立即 emit。
onTurnEnd('tool_use')：把 turnBuffer 作为 interstitial emit（trim 后非空才发），清空 turnBuffer。
onTurnEnd(其他)：把 turnBuffer 追加到 pendingFinal（用 \n\n 连接非空两端），清空 turnBuffer，不立即 emit。
drain()：先 emit pendingFinal 作为 final（非空才发），再 emit 残留 turnBuffer 作为 interstitial（非空才发），清空两者。
Step 1: 写失败测试

Create src/tests/turn-router.test.ts:

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { TurnRouter, type RoutedMessage } from '../claude/turn-router.js';

function newRouter() {
  const emitted: RoutedMessage[] = [];
  const router = new TurnRouter((m) => emitted.push(m));
  return { router, emitted };
}

test('onText 累积不立即 emit', () => {
  const { router, emitted } = newRouter();
  router.onText('hello ');
  router.onText('world');
  assert.deepEqual(emitted, []);
});

test('onTurnEnd(tool_use) 把 turnBuffer 作为 interstitial emit', () => {
  const { router, emitted } = newRouter();
  router.onText('让我看一下');
  router.onTurnEnd('tool_use');
  assert.deepEqual(emitted, [{ text: '让我看一下', role: 'interstitial' }]);
});

test('onTurnEnd(end_turn) 不立即 emit，攒到 drain', () => {
  const { router, emitted } = newRouter();
  router.onText('最终答案第一段');
  router.onTurnEnd('end_turn');
  assert.deepEqual(emitted, []);
  router.drain();
  assert.deepEqual(emitted, [{ text: '最终答案第一段', role: 'final' }]);
});

test('多个 end_turn 回合用 \\n\\n 连接成一个 final', () => {
  const { router, emitted } = newRouter();
  router.onText('段一');
  router.onTurnEnd('pause_turn');
  router.onText('段二');
  router.onTurnEnd('end_turn');
  router.drain();
  assert.deepEqual(emitted, [{ text: '段一\n\n段二', role: 'final' }]);
});

test('tool_use 和 end_turn 混合：interstitial 立即发，final 攒到 drain', () => {
  const { router, emitted } = newRouter();
  router.onText('让我查一下');
  router.onTurnEnd('tool_use');          // → interstitial 立即
  router.onText('找到了。');
  router.onText('详细说明...');
  router.onTurnEnd('end_turn');          // → final 攒着
  router.drain();                         // → final 发出
  assert.deepEqual(emitted, [
    { text: '让我查一下', role: 'interstitial' },
    { text: '找到了。详细说明...', role: 'final' },
  ]);
});

test('空文本回合不产生空消息', () => {
  const { router, emitted } = newRouter();
  router.onTurnEnd('tool_use');           // turnBuffer 空
  router.onTurnEnd('end_turn');           // turnBuffer 空
  router.drain();
  assert.deepEqual(emitted, []);
});

test('onTurnEnd 未触发时 drain 也能把残留 turnBuffer 当 interstitial 发出', () => {
  const { router, emitted } = newRouter();
  router.onText('未结束的残留');
  router.drain();
  assert.deepEqual(emitted, [
    { text: '未结束的残留', role: 'interstitial' },
  ]);
});

test('纯文本 Q&A（无 tool_use，单 end_turn）整段作为 final', () => {
  const { router, emitted } = newRouter();
  const chunks = ['闭包是...', '举个例子...', '总结...'];
  for (const c of chunks) router.onText(c);
  router.onTurnEnd('end_turn');
  router.drain();
  assert.deepEqual(emitted, [
    { text: chunks.join(''), role: 'final' },
  ]);
});

Step 2: 跑测试确认失败

Run:

npm run build && node --test dist/tests/turn-router.test.js

Expected: 导入失败 Cannot find module ../claude/turn-router.js。

Step 3: 实现 TurnRouter

Create src/claude/turn-router.ts:

/**
 * TurnRouter 把 Claude CLI 的流式输出按"回合"分流：
 *
 * - tool_use 回合的文本 → 立即作为 interstitial emit（agent loop 进度）
 * - 其他 stop_reason（end_turn / max_tokens / stop_sequence / pause_turn / ...）
 *   的文本 → 攒到 pendingFinal，drain 时一次性作为 final emit
 *
 * 设计参考 docs/superpowers/specs/2026-06-20-message-batching-design.md。
 *
 * 本类不做任何 I/O，只决定"何时把哪段文本以什么 role emit"。
 * 调用方（main.ts）负责把 RoutedMessage 切分（splitMessage）并发到微信。
 */

export type MessageRole = 'interstitial' | 'final';

export interface RoutedMessage {
  text: string;
  role: MessageRole;
}

export class TurnRouter {
  private turnBuffer = '';
  private pendingFinal = '';

  constructor(private readonly emit: (msg: RoutedMessage) => void) {}

  onText(delta: string): void {
    this.turnBuffer += delta;
  }

  onTurnEnd(stopReason: string): void {
    const text = this.turnBuffer;
    this.turnBuffer = '';
    if (!text.trim()) return;
    if (stopReason === 'tool_use') {
      this.emit({ text, role: 'interstitial' });
    } else {
      // end_turn / max_tokens / stop_sequence / pause_turn / 未知值
      // 一律当最终答案处理（宁可合并也不丢）
      this.pendingFinal += this.pendingFinal ? '\n\n' + text : text;
    }
  }

  /** 流结束时调用。先发 final，再 drain 残留 interstitial。 */
  drain(): void {
    if (this.pendingFinal.trim()) {
      this.emit({ text: this.pendingFinal, role: 'final' });
      this.pendingFinal = '';
    }
    if (this.turnBuffer.trim()) {
      this.emit({ text: this.turnBuffer, role: 'interstitial' });
      this.turnBuffer = '';
    }
  }
}

Step 4: 跑测试确认通过

Run:

npm run build && node --test dist/tests/turn-router.test.js

Expected: 8 个测试全 PASS。

Step 5: 提交

git add src/claude/turn-router.ts src/tests/turn-router.test.ts
git commit -m "$(cat <<'EOF'
feat: 新增 TurnRouter 状态机，按 stop_reason 分流 interstitial/final

纯逻辑模块，不持 I/O。onText 累积到 turnBuffer；onTurnEnd(tool_use)
立即 emit 为 interstitial，其他 stop_reason 攒到 pendingFinal；
drain 时一次性 emit 为 final。

覆盖 8 条路径：累积不 emit / tool_use 立即发 / end_turn 攒到 drain /
多 end_turn 用 \\n\\n 连接 / 混合 / 空回合不产生空消息 / 残留 drain /
纯文本 Q&A 整段 final。

下一个任务把 main.ts 接进来。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
EOF
)"

Task 4: 把 TurnRouter 接入 `main.ts`，删除死代码

Files:

Modify: src/main.ts（sendToClaude 函数：替换 textBuffer 逻辑、接入 TurnRouter、改 flush 顺序、删除死代码）
Modify: src/claude/provider.ts（删除 QueryOptions.onBlockEnd 字段、StreamParserCallbacks.onBlockEnd 字段、handleStreamLine 的 content_block_stop 分支调用）
Modify: src/tests/provider.test.ts（删除 onBlockEnd 相关测试）

Interfaces:

Consumes:
- Task 1/2 的 handleStreamLine / StreamParserCallbacks（无 onBlockEnd）
- Task 2 的 QueryOptions.onTurnEnd
- Task 3 的 TurnRouter / RoutedMessage
Produces: 无新对外接口（sendToClaude 仍是内部函数）
Step 1: 删除 provider.ts 里的 onBlockEnd（先消提供方，让消费方编译报错暴露出来）

1a. 修改 src/claude/provider.ts 的 QueryOptions 接口，删除：

  /** Called when a content block ends — use to flush buffered text. */
  onBlockEnd?: () => Promise<void> | void;

1b. 修改 StreamParserCallbacks，删除 onBlockEnd 字段：

export interface StreamParserCallbacks {
  onText?: (text: string) => void;
  onBlockEnd?: () => void;        // ← 删这一行
  onTurnEnd?: (stopReason: string) => void;
}

变成：

export interface StreamParserCallbacks {
  onText?: (text: string) => void;
  onTurnEnd?: (stopReason: string) => void;
}

1c. 修改 handleStreamLine 的 content_block_stop 分支，不再调用 callbacks：

找到：

      } else if (evt?.type === 'content_block_stop') {
        state.trackingSkill = false;
        if (callbacks.onBlockEnd) callbacks.onBlockEnd();
      } else if (evt?.type === 'message_delta' && evt.delta?.stop_reason) {

改为：

      } else if (evt?.type === 'content_block_stop') {
        state.trackingSkill = false;
      } else if (evt?.type === 'message_delta' && evt.delta?.stop_reason) {

1d. 修改 claudeQuery 内的 parserCallbacks 构造，删除 onBlockEnd：

找到：

    const parserCallbacks: StreamParserCallbacks = { onText, onBlockEnd, onTurnEnd };

改为：

    const parserCallbacks: StreamParserCallbacks = { onText, onTurnEnd };

1e. 在 claudeQuery 函数签名解构里删除 onBlockEnd。 找到：

  const {
    prompt,
    cwd,
    resume,
    model,
    systemPrompt,
    images,
    onText,
    onBlockEnd,
    abortController,
  } = options;

改为（同时加 onTurnEnd）：

  const {
    prompt,
    cwd,
    resume,
    model,
    systemPrompt,
    images,
    onText,
    onTurnEnd,
    abortController,
  } = options;

Step 2: 改 provider.test.ts，删 onBlockEnd 测试

在 src/tests/provider.test.ts 删除：

test('handleStreamLine: content_block_stop 触发 onBlockEnd', () => {
  let called = 0;
  handleStreamLine(
    JSON.stringify({ type: 'stream_event', event: { type: 'content_block_stop', index: 0 } }),
    freshState(),
    { onBlockEnd: () => called++ },
  );
  assert.equal(called, 1);
});

替换为（验证 content_block_stop 仍重置 trackingSkill，但不发回调）：

test('handleStreamLine: content_block_stop 重置 trackingSkill，无回调', () => {
  const state = freshState();
  state.trackingSkill = true;
  let textCalls = 0;
  handleStreamLine(
    JSON.stringify({ type: 'stream_event', event: { type: 'content_block_stop', index: 0 } }),
    state,
    { onText: () => textCalls++ },
  );
  assert.equal(state.trackingSkill, false);
  assert.equal(textCalls, 0);
});

Step 3: 编译确认 provider 侧改动不报错（此时 main.ts 还在传 onBlockEnd，预期会报错）

Run:

npm run build 2>&1 | head -20

Expected: 在 src/main.ts 报 TypeScript 错误，类似：

error TS2322: Type '{ onText: ...; onBlockEnd: ...; onTurnEnd: ...; }' is not assignable to type 'QueryOptions' ...
  Object literal may only specify known properties, and 'onBlockEnd' does not exist in type 'QueryOptions'.

这是预期的——下一个 step 修 main.ts。

Step 4: 重写 main.ts 的 sendToClaude，接入 TurnRouter

4a. 加 import。 在 src/main.ts 顶部 imports 区，找到：

import { claudeQuery, type QueryOptions } from './claude/provider.js';

在其之后加：

import { TurnRouter } from './claude/turn-router.js';

4b. 删除死代码 endsWithStructuralBoundary。 在 src/main.ts 的 sendToClaude 函数内（约 501-503 行）找到：

    /** Check if buffer ends at a structural boundary (double newline or horizontal rule). */
    function endsWithStructuralBoundary(text: string): boolean {
      return /\n\n\s*$/.test(text) || /\n[-*_]{3,}\s*$/.test(text);
    }

整个函数删除。

注意：MIN_BATCH_FLUSH_LEN 和 SOFT_FLUSH_LIMIT 两个常量在 sendToClaude 内部声明，紧跟在 endsWithStructuralBoundary 上方。它们会在 step 4c 的整段替换里一并消失（OLD 块包含它们，NEW 块不包含），无需单独处理。

4c. 替换 sendToClaude 内的流式处理段。 找到现有的（约 493-591 行）：

    let textBuffer = '';
    let anySent = false;
    let lastSentTime = Date.now();

    const MIN_BATCH_FLUSH_LEN = 30;
    const SOFT_FLUSH_LIMIT = 3800;

    /** Check if buffer ends at a structural boundary (double newline or horizontal rule). */
    function endsWithStructuralBoundary(text: string): boolean {
      return /\n\n\s*$/.test(text) || /\n[-*_]{3,}\s*$/.test(text);
    }

    // Serial promise chain — each flushText() appends to the chain, no flags needed
    let flushChain: Promise<void> = Promise.resolve();

    function flushText(): Promise<void> {
      // Capture and clear synchronously to prevent race condition:
      // new deltas can arrive while the chain awaits sendText,
      // causing the async callback to clear content it never captured.
      const captured = textBuffer.trim();
      textBuffer = '';
      if (!captured) return flushChain;

      flushChain = flushChain.then(async () => {
        const chunks = splitMessage(captured);
        for (let i = 0; i < chunks.length; i++) {
          try {
            await sender.sendText(fromUserId, contextToken, chunks[i]);
          } catch (err) {
            // Rate-limit exhaustion etc.: put the unsent chunks back at the
            // front of the buffer so the next flush retries them. Content is
            // never silently dropped (previously the for-loop aborted here and
            // the already-cleared buffer lost everything from this chunk on).
            const remaining = chunks.slice(i).join('\n\n');
            textBuffer = remaining + (textBuffer ? '\n\n' + textBuffer : '');
            logger.warn('flushText send failed, content retained for retry', {
              error: err instanceof Error ? err.message : String(err),
              retainedChunks: chunks.length - i,
            });
            return;
          }
        }
        anySent = true;
        lastSentTime = Date.now();
      });
      return flushChain;
    }

    // Safety net: send keepalive if nothing was sent for 5 minutes
    const SILENCE_WARNING_MS = 5 * 60 * 1000;
    const SILENCE_MESSAGES = [
      '我还在处理中，这个问题有点复杂，请再稍等一下',
      '正在努力干活中，马上就有结果了，请稍等片刻',
      '有点复杂正在处理，再给我一点时间，很快就好',
      '快好了别着急，正在收尾阶段，马上给你回复',
      '还在跑呢，任务量比较大，不过马上就能出结果了',
      '任务比想象的复杂一些，再等等我，正在全力处理',
      '正在处理中，进展顺利，再等一会儿就好',
      '还没完不过已经快了，再给我一分钟就能搞定',
      '我在认真思考这个问题，请再稍等一会儿',
      '稍微有点棘手，不过已经快解决了，再等我一下',
    ];
    flushTimer = setInterval(() => {
      if (Date.now() - lastSentTime > SILENCE_WARNING_MS) {
        const msg = SILENCE_MESSAGES[Math.floor(Math.random() * SILENCE_MESSAGES.length)];
        sender.sendText(fromUserId, contextToken, msg).catch(() => {});
        lastSentTime = Date.now();
      }
    }, 2000);

    const queryOptions: QueryOptions = {
      prompt,
      cwd: (session.workingDirectory || config.workingDirectory).replace(/^~/, homedir()),
      resume: session.sdkSessionId,
      model: session.model,
      systemPrompt: [
        '你正在通过微信与用户对话，不是在终端里。不要让用户去终端操作。如果用户需要文件，直接输出文件地址就行，会自动识别解析推送文件到用户的微信中。',
        config.systemPrompt,
      ].filter(Boolean).join('\n'),
      abortController,
      images,
      onText: async (delta: string) => {
        textBuffer += delta;

        // Flush at structural boundaries (only if buffer is substantial) or when approaching size limit
        const shouldFlush =
          (endsWithStructuralBoundary(textBuffer) && textBuffer.trim().length >= MIN_BATCH_FLUSH_LEN)
          || textBuffer.length > SOFT_FLUSH_LIMIT;

        if (shouldFlush) {
          await flushText();
        }
      },
      onBlockEnd: () => {
        if (textBuffer.trim().length >= MIN_BATCH_FLUSH_LEN || textBuffer.length > SOFT_FLUSH_LIMIT) {
          flushText();
        }
      },
    };

整段替换为：

    let anySent = false;
    let lastSentTime = Date.now();
    let pendingRetry = '';   // sendText 失败时未发出的 chunks，下一次 flush 优先重试

    // Serial promise chain — each emit appends to the chain, no flags needed
    let flushChain: Promise<void> = Promise.resolve();

    /** 把一段文本切分后串行发到微信。失败时把未发的 chunks 攒到 pendingRetry，下次重试。 */
    function emitText(text: string, role: 'interstitial' | 'final'): void {
      if (!text.trim()) return;
      flushChain = flushChain.then(async () => {
        const combined = pendingRetry ? pendingRetry + '\n\n' + text : text;
        pendingRetry = '';
        if (!combined.trim()) return;
        const chunks = splitMessage(combined);
        for (let i = 0; i < chunks.length; i++) {
          try {
            await sender.sendText(fromUserId, contextToken, chunks[i]);
          } catch (err) {
            // Rate-limit exhaustion etc.: put the unsent chunks back so the
            // next emit retries them. Content is never silently dropped.
            pendingRetry = chunks.slice(i).join('\n\n');
            logger.warn('emitText send failed, content retained for retry', {
              role,
              error: err instanceof Error ? err.message : String(err),
              retainedChunks: chunks.length - i,
            });
            return;
          }
        }
        anySent = true;
        lastSentTime = Date.now();
      });
    }

    const router = new TurnRouter((msg) => emitText(msg.text, msg.role));

    // Safety net: send keepalive if nothing was sent for 5 minutes
    const SILENCE_WARNING_MS = 5 * 60 * 1000;
    const SILENCE_MESSAGES = [
      '我还在处理中，这个问题有点复杂，请再稍等一下',
      '正在努力干活中，马上就有结果了，请稍等片刻',
      '有点复杂正在处理，再给我一点时间，很快就好',
      '快好了别着急，正在收尾阶段，马上给你回复',
      '还在跑呢，任务量比较大，不过马上就能出结果了',
      '任务比想象的复杂一些，再等等我，正在全力处理',
      '正在处理中，进展顺利，再等一会儿就好',
      '还没完不过已经快了，再给我一分钟就能搞定',
      '我在认真思考这个问题，请再稍等一会儿',
      '稍微有点棘手，不过已经快解决了，再等我一下',
    ];
    flushTimer = setInterval(() => {
      if (Date.now() - lastSentTime > SILENCE_WARNING_MS) {
        const msg = SILENCE_MESSAGES[Math.floor(Math.random() * SILENCE_MESSAGES.length)];
        sender.sendText(fromUserId, contextToken, msg).catch(() => {});
        lastSentTime = Date.now();
      }
    }, 2000);

    const queryOptions: QueryOptions = {
      prompt,
      cwd: (session.workingDirectory || config.workingDirectory).replace(/^~/, homedir()),
      resume: session.sdkSessionId,
      model: session.model,
      systemPrompt: [
        '你正在通过微信与用户对话，不是在终端里。不要让用户去终端操作。如果用户需要文件，直接输出文件地址就行，会自动识别解析推送文件到用户的微信中。',
        config.systemPrompt,
      ].filter(Boolean).join('\n'),
      abortController,
      images,
      onText: (delta: string) => {
        router.onText(delta);
      },
      onTurnEnd: (stopReason: string) => {
        router.onTurnEnd(stopReason);
      },
    };

4d. 修改流结束后的 flush 顺序。 找到（约 605-627 行）：

    // Stop periodic flush and send any remaining buffered content
    clearInterval(flushTimer);
    await flushText();

    // Send result back to WeChat
    if (result.text) {
      if (result.error) {
        logger.warn('Claude query had error but returned text, using text', { error: result.error });
      }
      sessionStore.addChatMessage(session, 'assistant', result.text);
      // If nothing was streamed at all (e.g. streaming not supported), send full text now
      if (!anySent) {
        const chunks = splitMessage(result.text);
        for (const chunk of chunks) {
          await sender.sendText(fromUserId, contextToken, chunk);
        }
      }
    } else if (result.error) {

替换 clearInterval(flushTimer); 和 await flushText(); 这两行（保留后面所有内容不变）：

    // Stop periodic flush, drain router (final 先于 interstitial), wait for queued sends
    clearInterval(flushTimer);
    router.drain();
    await flushChain;

    // Send result back to WeChat
    if (result.text) {
      if (result.error) {
        logger.warn('Claude query had error but returned text, using text', { error: result.error });
      }
      sessionStore.addChatMessage(session, 'assistant', result.text);
      // If nothing was streamed at all (e.g. streaming not supported), send full text now
      if (!anySent) {
        const chunks = splitMessage(result.text);
        for (const chunk of chunks) {
          await sender.sendText(fromUserId, contextToken, chunk);
        }
      }
    } else if (result.error) {

Step 5: 编译并跑所有测试

Run:

npm run build && npm test

Expected:

TypeScript 编译零错误（确认 onBlockEnd 已彻底从 main.ts 移除）。
所有测试 PASS：provider.test.ts（8 个）+ turn-router.test.ts（8 个）。
Step 6: 端到端冒烟

Run:

echo "" | node dist/main.js 2>&1 | head -5 || true

Expected: 进程启动、读到 未找到账号 或类似的运行期错误（因为没配置），但不应报 TypeScript / 模块解析错误。这验证编译产物导入正常。

Step 7: 真实 trace 验证新逻辑依赖的信号确实存在

重新跑一份 trace，确认 Claude CLI 真实输出里包含新逻辑依赖的 message_delta 事件：

mkdir -p /tmp/verify-batching && cd /tmp/verify-batching && \
echo "用三段话解释 JavaScript 闭包" | claude -p - --output-format stream-json --verbose --include-partial-messages --dangerously-skip-permissions 2>/dev/null > trace.jsonl && \
echo "trace 行数: $(wc -l < trace.jsonl)" && \
echo "--- message_delta 事件数（应 >= 1）---" && \
grep -c '"type":"message_delta"' trace.jsonl && \
echo "--- 各 stop_reason 分布 ---" && \
grep -o '"stop_reason":"[^"]*"' trace.jsonl | sort | uniq -c

Expected:

trace 行数 > 100。
至少 1 个 message_delta 事件。
stop_reason 分布里至少有 1 个 end_turn（纯 Q&A 场景）。

这一步只验证「我们依赖的信号真实存在」。完整端到端效果验证（消息条数从 N 降到 1）需要装上 daemon 在微信里实测，见下方「验收清单」。

Step 8: 提交

git add src/main.ts src/claude/provider.ts src/tests/provider.test.ts
git commit -m "$(cat <<'EOF'
feat: main.ts 接入 TurnRouter，最终答案改为按回合整段推送

把 sendToClaude 原来的单 textBuffer + 段落边界 flush 逻辑替换为
TurnRouter 状态机：onText 只累积不 flush，onTurnEnd(tool_use) 立即
emit 为 interstitial，其他 stop_reason 攒到 pendingFinal，流结束时
router.drain() 一次性 emit 为 final（splitMessage 按 4000 字硬切）。

真实 trace 验证效果：
- 纯文本 Q&A（1485 字）：8 条 → 1 条
- tool_use + 9121 字长答案：59 条 → 4 条

顺带清理死代码：删除 MIN_BATCH_FLUSH_LEN / SOFT_FLUSH_LIMIT /
endsWithStructuralBoundary / onBlockEnd（provider.ts 同步移除字段
和 content_block_stop 回调，测试相应更新）。

限流逻辑、splitMessage、typing、silence warning、文件自动推送
全部不动。emitText 保留了 commit d6d7d62 引入的失败重试语义
（pendingRetry）。

Spec: docs/superpowers/specs/2026-06-20-message-batching-design.md

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
EOF
)"

验收清单（实现完成后人工跑一遍）

参考 spec § 7.3：

纯 Q&A：微信发"用三段话解释闭包"→ 应收到 1 条完整答案（而非现在的 8 条）。
多 tool_use + 长答案：微信发"分析 src/main.ts 的结构"→ 应收到 1 条 interstitial + N 条最终答案块（按 4000 字切），总条数远少于现在。
Abort：发任务后立即发 /stop → 已发的 interstitial 不丢，部分生成的最终答案按已生成内容推送，无内容丢失。
限流恢复：观察日志，若偶发 emitText send failed, content retained for retry，下一次 emit 应自动重试成功（pendingRetry 机制）。

43 KiB Raw Permalink Blame History Unescape Escape

微信消息分块优化 Implementation Plan

Global Constraints

Task 1: 提取 handleStreamLine 到 provider.ts（纯重构，为可测性铺路）

Task 2: 在 provider.ts 加 onTurnEnd 回调

Task 3: 创建 TurnRouter 状态机

Task 4: 把 TurnRouter 接入 main.ts，删除死代码

验收清单（实现完成后人工跑一遍）

43 KiB

Raw Permalink Blame History

Task 1: 提取 `handleStreamLine` 到 provider.ts（纯重构，为可测性铺路）

Task 2: 在 provider.ts 加 `onTurnEnd` 回调

Task 3: 创建 `TurnRouter` 状态机

Task 4: 把 TurnRouter 接入 `main.ts`，删除死代码