claw-code 源码详细分析：Turn Loop 里的工程细节——多轮对话如何在移植期保持可测试、可回放？

路径：result/04.md（若你习惯写reuslt，对应到仓库的result/目录即可。）
涉及源码：src/runtime.py（run_turn_loop、bootstrap_session）、src/query_engine.py、src/session_store.py、src/transcript.py、src/history.py、src/main.py、tests/test_porting_workspace.py。

1. Turn Loop 在仓库里的定位

PortRuntime.run_turn_loop不是「模拟用户与模型你来我往的真实多轮产品逻辑」，而是一个移植期探针：在同一组路由结果上，对同一个QueryEnginePort实例连续调用多次submit_message，观察：

• 会话状态（mutable_messages、total_usage、transcript_store）如何累积；
• stop_reason何时从completed变为max_turns_reached或max_budget_reached；
• --structured-output打开后，每一轮TurnResult.output是否仍是稳定可解析的 JSON。

这样做的直接好处是：无网络、无 API Key、无随机温度，多轮行为在 CI 里可重复，满足「移植期可测试」；可回放则通过另一条链路——持久化 JSON +load_session/from_saved_session+replay_user_messages——部分实现，下文分述。

2.run_turn_loop：循环与引擎配置的耦合

def run_turn_loop(self, prompt: str, limit: int = 5, max_turns: int = 3, structured_output: bool = False) -> list[TurnResult]: engine = QueryEnginePort.from_workspace() engine.config = QueryEngineConfig(max_turns=max_turns, structured_output=structured_output) matches = self.route_prompt(prompt, limit=limit) command_names = tuple(match.name for match in matches if match.kind == 'command') tool_names = tuple(match.name for match in matches if match.kind == 'tool') results: list[TurnResult] = [] for turn in range(max_turns): turn_prompt = prompt if turn == 0 else f'{prompt} [turn {turn + 1}]' result = engine.submit_message(turn_prompt, command_names, tool_names, ()) results.append(result) if result.stop_reason != 'completed': break return results

2.1 工程细节一：外层循环次数与QueryEngineConfig.max_turns同名同值

CLI 把--max-turns同时传给：

• for turn in range(max_turns)——最多尝试几轮submit_message；
• QueryEngineConfig(max_turns=max_turns)——引擎内部允许写入mutable_messages的最大条数（见query_engine里len(self.mutable_messages) >= self.config.max_turns的闸门）。

在默认用法下，两者一致，因此常见情况是：每一轮都能completed，直到跑满循环次数。若将来有人把「循环次数」与「引擎 max_turns」拆成不同参数，第N轮可能提前拿到stop_reason='max_turns_reached'，run_turn_loop会因result.stop_reason != 'completed'提前break——这是可测试的显式出口，而不是死循环或静默失败。

2.2 工程细节二：路由只做一次，多轮共用command_names/tool_names

route_prompt(prompt, limit)在循环外调用一次，整段 loop 复用同一matches推导出的command_names/tool_names。含义是：

• 移植期：专注测QueryEngine 状态机 + 输出格式，不引入「每轮重新理解意图」的变量。
• 产品期局限：真实多轮里用户意图会变，路由应每轮更新；当前实现是刻意简化，阅读时勿误认为已是完整对话产品。

2.3 工程细节三：合成 turn 文本，保证每轮输入可区分

turn_prompt=promptifturn==0elsef'{prompt}[turn{turn+1}]'

同一语义骨架下微调字符串，使得：

• mutable_messages中各条不相等，便于 diff 与调试；
• UsageSummary.add_turn按词数累加时，每轮输入长度略有变化，预算边界可通过调整max_budget_tokens在测试中触发。

2.4 工程细节四：不传denied_tools（空元组）

循环里固定denied_tools=()，多轮不累积权限拒绝。与bootstrap_session（带_infer_permission_denials）对比，Turn Loop 走的是「纯会话闸门 + 用量」切片测试，权限审计在单轮 bootstrap 报告里练。

3. 可测试性：TurnResult列表即「黄金轨迹」

run_turn_loop返回list[TurnResult]，每一元素包含本轮输入、输出、匹配元数据、用量快照、停止原因：

@dataclass(frozen=True) class TurnResult: prompt: str output: str matched_commands: tuple[str, ...] matched_tools: tuple[str, ...] permission_denials: tuple[PermissionDenial, ...] usage: UsageSummary stop_reason: str

单测/CI 可以断言：

• 轮次数：例如test_turn_loop_cli_runs使用--max-turns 2，检查输出中出现## Turn 1与stop_reason=（见tests/test_porting_workspace.py）。
• 某一固定prompt+ 固定快照下，matched_tools是否非空（test_bootstrap_session_tracks_turn_state针对 bootstrap，同一套路可迁到纯QueryEnginePort单测）。
• structured_output=True时，每轮output是否为合法 JSON（当前实现为json.dumps包一层summary+session_id）。

冻结数据面：命令/工具来自reference_data/*.json，路由算法确定性强，无模型随机性，这是「移植期可测试」的基石。

4. 可回放：三条互补路径

4.1 内存轨迹：results: list[TurnResult]

调用方若拿到run_turn_loop的返回值，已具备按轮重放「当时引擎认为发生了什么」的只读记录（适合单元测试内联断言）。

4.2 转写replay：TranscriptStore.replay()/QueryEnginePort.replay_user_messages

def replay(self) -> tuple[str, ...]: return tuple(self.entries)

def replay_user_messages(self) -> tuple[str, ...]: return self.transcript_store.replay()

submit_message在成功路径上会对mutable_messages与transcript_store同步append，因此用户侧消息序列可通过replay_user_messages()取出，用于不依赖磁盘的轻量回放或与其他模块对拍。

注意：run_turn_loop未调用persist_session，因此默认 Turn Loop 结束后若不做持久化，磁盘上没有该次多轮的 JSON；回放仅限进程内或通过自写测试保存TurnResult。

4.3 磁盘会话：persist_session+load_session+from_saved_session

bootstrap_session在单轮结束后调用engine.persist_session()，把session_id、messages、累计 input/output 伪 token写入.port_sessions/<id>.json：

def persist_session(self) -> str: self.flush_transcript() path = save_session( StoredSession( session_id=self.session_id, messages=tuple(self.mutable_messages), input_tokens=self.total_usage.input_tokens, output_tokens=self.total_usage.output_tokens, ) ) return str(path)

测试链路test_load_session_cli_runs：bootstrap → 取persisted_session_path的 stem →load-session，验证持久化与读取闭环。

@classmethod def from_saved_session(cls, session_id: str) -> 'QueryEnginePort': stored = load_session(session_id) transcript = TranscriptStore(entries=list(stored.messages), flushed=True) return cls( manifest=build_port_manifest(), session_id=stored.session_id, mutable_messages=list(stored.messages), total_usage=UsageSummary(stored.input_tokens, stored.output_tokens), transcript_store=transcript, )

可回放含义：新进程可from_saved_sessionhydrate引擎，继续submit_message（若业务允许），实现跨运行的会话延续。当前StoredSession不保存每轮TurnResult.output与累积permission_denials，因此严格说是「用户消息 + 用量摘要」级回放，不是完整对话录；移植期够用，产品期需扩展 schema。

4.4 人类可读报告：RuntimeSession.as_markdown+HistoryLog

单轮bootstrap_session把上下文、setup、路由、执行 shim、流式事件、turn_result、持久化路径与HistoryLog事件链打成一篇 Markdown：

history.add('routing', f'matches={len(matches)} for prompt={prompt!r}') history.add('execution', f'command_execs={len(command_execs)} tool_execs={len(tool_execs)}') history.add('turn', f'commands={len(turn_result.matched_commands)} tools={len(turn_result.matched_tools)} denials={len(turn_result.permission_denials)} stop={turn_result.stop_reason}') history.add('session_store', persisted_session_path)

def as_markdown(self) -> str: lines = ['# Session History', ''] lines.extend(f'- {event.title}: {event.detail}' for event in self.events) return '\n'.join(lines)

适合PR 审查、故障单附件：一眼看到路由与 stop 原因，无需复现者本地再跑模型。

5. 与bootstrap_session的对比（单轮 vs 多轮）

二者互补：bootstrap 练「审计叙事与落盘」，turn-loop 练「多轮状态累积与早停」。

6. CLI 与测试如何锁住行为

CLI（src/main.py）：

if args.command == 'turn-loop': results = PortRuntime().run_turn_loop(args.prompt, limit=args.limit, max_turns=args.max_turns, structured_output=args.structured_output) for idx, result in enumerate(results, start=1): print(f'## Turn {idx}') print(result.output) print(f'stop_reason={result.stop_reason}') return 0

测试test_turn_loop_cli_runs：子进程跑turn-loop，断言输出含## Turn 1与stop_reason=——端到端、无 mock，与test_bootstrap_cli_runs、test_load_session_cli_runs一起构成会话相关的回归网。

7. 移植期建议与演进方向（基于现状）

1. 可测试：保持TurnResult稳定；路由与快照 JSON 固定；Turn Loop 继续作为确定性压力小工具。
2. 可回放：若要多轮落盘，可在run_turn_loop末尾可选调用persist_session，或扩展StoredSession保存每轮output/stop_reason。
3. 真实多轮：将route_prompt移入循环内，并传入上一轮 assistant 输出（或工具结果），才接近产品语义；当前实现是有意减变量的脚手架。
4. 预算测试：调低QueryEngineConfig.max_budget_tokens（需在run_turn_loop暴露参数或构造专用测试入口），可稳定触发max_budget_reached与提前break分支。

8. 小结

• Turn Loop通过确定性路由 + 冻结的 matched 元组 + 合成 turn 文本，在无 LLM条件下演练多轮状态累积与停止语义，并以list[TurnResult]作为可断言轨迹。
• 可回放依赖TranscriptStore/replay_user_messages（进程内）与persist_session+load_session/from_saved_session（跨进程）；Turn Loop 默认不落盘，与 bootstrap 分工明确。
• HistoryLog+ Markdown 报告提供人类可读的单轮审计切片，与多轮机器可断言结果形成测试金字塔的两层。