claw-code 源码分析：从「清单」到「运行时」——Harness 为什么必须先做 inventory 再做 I/O？

说明本文分析对象为开源仓库claw-codeREADME 中Rewriting Project Claw Code的 Python/Rust 移植工作区。1. 问题在问什么Inventory清单在 Harness 里指「系统承认存在的命令名、工具名及其元数据」的有穷集合——谁算内置、谁算插件、谁可被模型调用、各自职责与来源提示是什么。I/O输入输出指真正对外部世界产生副作用的行为——读盘、起进程、调网络 API、改用户仓库等。核心论点是在智能体系统里若没有一个稳定、可枚举、可过滤的清单就直接开放 I/O会把「命名空间」「路由」「权限」「审计」全部绑死在即兴逻辑上后期每一次加工具/加命令都会变成全库手术。claw-code 的 Python 移植层用快照 JSON → 内存元组 → 路由/注册表 →再模拟执行的链条把这一顺序写进了代码结构本身。2. 源码中的「清单层」长什么样2.1 数据源头reference_data快照命令与工具的权威枚举来自版本库内的 JSON而不是运行时再扫描磁盘猜名字src/reference_data/commands_snapshot.jsonsrc/reference_data/tools_snapshot.jsoncommands.py/tools.py在模块加载时读取 JSON解析为不可变元组PORTED_COMMANDS、PORTED_TOOLS并缓存lru_cache(maxsize1) def load_command_snapshot() - tuple[PortingModule, ...]: raw_entries json.loads(SNAPSHOT_PATH.read_text()) return tuple( PortingModule( nameentry[name], responsibilityentry[responsibility], source_hintentry[source_hint], statusmirrored, ) for entry in raw_entries ) PORTED_COMMANDS load_command_snapshot()lru_cache(maxsize1) def load_tool_snapshot() - tuple[PortingModule, ...]: raw_entries json.loads(SNAPSHOT_PATH.read_text()) return tuple( PortingModule( nameentry[name], responsibilityentry[responsibility], source_hintentry[source_hint], statusmirrored, ) for entry in raw_entries ) PORTED_TOOLS load_tool_snapshot()学习点清单与代码解耦——JSON 可 diff、可审计、可随 parity 演进Python 侧只消费「已镜像」条目避免运行时动态发现带来的不可重复性。2.2 在清单之上做「视图」过滤、简单模式、权限get_tools()不重新发明清单而是在PORTED_TOOLS上叠加策略simple mode、是否包含 MCP、权限上下文def get_tools( simple_mode: bool False, include_mcp: bool True, permission_context: ToolPermissionContext | None None, ) - tuple[PortingModule, ...]: tools list(PORTED_TOOLS) if simple_mode: tools [module for module in tools if module.name in {BashTool, FileReadTool, FileEditTool}] if not include_mcp: tools [module for module in tools if mcp not in module.name.lower() and mcp not in module.source_hint.lower()] return filter_tools_by_permission_context(tuple(tools), permission_context)tool_pool.py的assemble_tool_pool只是把上述「当前允许的子集」包装成报告对象——先有全集inventory再有池policy 下的视图def assemble_tool_pool( simple_mode: bool False, include_mcp: bool True, permission_context: ToolPermissionContext | None None, ) - ToolPool: return ToolPool( toolsget_tools(simple_modesimple_mode, include_mcpinclude_mcp, permission_contextpermission_context), simple_modesimple_mode, include_mcpinclude_mcp, )学习点权限与产品模式是清单上的过滤器不是散落在每个 I/O 调用点里的 if-else没有 inventory过滤器无处附着。2.3 命令「图」仍是清单的划分command_graph.py根据source_hint把同一批PORTED_COMMANDS分成 builtin / plugin-like / skill-like——拓扑来自元数据字段而不是执行时行为def build_command_graph() - CommandGraph: commands get_commands() builtins tuple(module for module in commands if plugin not in module.source_hint.lower() and skills not in module.source_hint.lower()) plugin_like tuple(module for module in commands if plugin in module.source_hint.lower()) skill_like tuple(module for module in commands if skills in module.source_hint.lower()) return CommandGraph(builtinsbuiltins, plugin_likeplugin_like, skill_likeskill_like)3. 「I/O 层」在本仓库里如何被刻意推迟3.1 执行入口execute_*首先是名字校验 描述性消息真正的危险 I/O 并未接在execute_tool上当前实现是mirrored shim只在清单里找到名字时返回「将会如何处理」的字符串def execute_tool(name: str, payload: str ) - ToolExecution: module get_tool(name) if module is None: return ToolExecution(namename, source_hint, payloadpayload, handledFalse, messagefUnknown mirrored tool: {name}) action fMirrored tool {module.name} from {module.source_hint} would handle payload {payload!r}. return ToolExecution(namemodule.name, source_hintmodule.source_hint, payloadpayload, handledTrue, messageaction)命令同理execute_command。学习点Harness 演进的标准节奏是——先让「调用约定」在清单内跑通名字、payload、返回结构再接真实后端若颠倒顺序调试时无法区分「路由错了」还是「I/O 错了」。3.2 注册表ExecutionRegistry完全由清单构造build_execution_registry()遍历PORTED_COMMANDS/PORTED_TOOLS生成可查找对象注册表容量 清单条目数def build_execution_registry() - ExecutionRegistry: return ExecutionRegistry( commandstuple(MirroredCommand(module.name, module.source_hint) for module in PORTED_COMMANDS), toolstuple(MirroredTool(module.name, module.source_hint) for module in PORTED_TOOLS), )运行时拿路由结果去 registry 里取执行器——没有 inventoryregistry 无法构建路由结果也无法落到稳定 handler。4.PortRuntime路由与清单的硬依赖PortRuntime.route_prompt的输入是用户prompt但匹配对象只能是PORTED_COMMANDS与PORTED_TOOLS中的模块它用 token 与name/source_hint/responsibility做打分产出有限条RoutedMatchclass PortRuntime: def route_prompt(self, prompt: str, limit: int 5) - list[RoutedMatch]: tokens {token.lower() for token in prompt.replace(/, ).replace(-, ).split() if token} by_kind { command: self._collect_matches(tokens, PORTED_COMMANDS, command), tool: self._collect_matches(tokens, PORTED_TOOLS, tool), } selected: list[RoutedMatch] [] for kind in (command, tool): if by_kind[kind]: selected.append(by_kind[kind].pop(0)) leftovers sorted( [match for matches in by_kind.values() for match in matches], keylambda item: (-item.score, item.kind, item.name), ) selected.extend(leftovers[: max(0, limit - len(selected))]) return selected[:limit]bootstrap_session的流程顺序非常清晰构建上下文与 setup环境自省QueryEnginePort.from_workspace()再拉 manifest / summary 相关状态history 记下commands{len(PORTED_COMMANDS)}, tools{len(PORTED_TOOLS)}——显式把清单规模当作会话元数据route_prompt→build_execution_registry()→ 仅对匹配到的名字执行 shim再把matched_commands/matched_tools/ 推断的denials交给QueryEnginePort的submit_message/stream_submit_messagedef bootstrap_session(self, prompt: str, limit: int 5) - RuntimeSession: context build_port_context() setup_report run_setup(trustedTrue) setup setup_report.setup history HistoryLog() engine QueryEnginePort.from_workspace() history.add(context, fpython_files{context.python_file_count}, archive_available{context.archive_available}) history.add(registry, fcommands{len(PORTED_COMMANDS)}, tools{len(PORTED_TOOLS)}) matches self.route_prompt(prompt, limitlimit) registry build_execution_registry() command_execs tuple(registry.command(match.name).execute(prompt) for match in matches if match.kind command and registry.command(match.name)) tool_execs tuple(registry.tool(match.name).execute(prompt) for match in matches if match.kind tool and registry.tool(match.name)) denials tuple(self._infer_permission_denials(matches)) stream_events tuple(engine.stream_submit_message( prompt, matched_commandstuple(match.name for match in matches if match.kind command), matched_toolstuple(match.name for match in matches if match.kind tool), denied_toolsdenials, )) turn_result engine.submit_message( prompt, matched_commandstuple(match.name for match in matches if match.kind command), matched_toolstuple(match.name for match in matches if match.kind tool), denied_toolsdenials, )学习点路由routing是定义在有穷 inventory 上的搜索问题I/O 只应作用于路由后的已解析符号。若先写 I/O常见反模式是「字符串里猜路径」「正则提取 shell 片段」——不可枚举、不可审计。权限拒绝示例_infer_permission_denials同样建立在已匹配的工具名上例如 bash 类工具说明deny-list / gate 需要名字语义而名字来自清单。5.QueryEnginePort会话与预算——仍以「匹配集合」为输入submit_message并不自己去「发现」工具它接收调用方已经算好的matched_commands、matched_tools与denied_tools再写入摘要、用量、转写与压缩策略summary_lines [ fPrompt: {prompt}, fMatched commands: {, .join(matched_commands) if matched_commands else none}, fMatched tools: {, .join(matched_tools) if matched_tools else none}, fPermission denials: {len(denied_tools)}, ] output self._format_output(summary_lines) projected_usage self.total_usage.add_turn(prompt, output) stop_reason completed if projected_usage.input_tokens projected_usage.output_tokens self.config.max_budget_tokens: stop_reason max_budget_reached self.mutable_messages.append(prompt) self.transcript_store.append(prompt) self.permission_denials.extend(denied_tools) self.total_usage projected_usage self.compact_messages_if_needed() return TurnResult( promptprompt, outputoutput, matched_commandsmatched_commands, matched_toolsmatched_tools, permission_denialsdenied_tools, usageself.total_usage, stop_reasonstop_reason, )render_summary()再次聚合manifest command/tool backlog仍来自清单说明「给用户/维护者看的系统面」与 inventory 同源。6. Bootstrap 阶段叙事bootstrap_graph把顺序写死build_bootstrap_graph()用字符串阶段描述了整个启动链其中「setup commands/agents 并行加载」在「query engine submit loop」之前def build_bootstrap_graph() - BootstrapGraph: return BootstrapGraph( stages( top-level prefetch side effects, warning handler and environment guards, CLI parser and pre-action trust gate, setup() commands/agents parallel load, deferred init after trust, mode routing: local / remote / ssh / teleport / direct-connect / deep-link, query engine submit loop, ) )这与前文代码一致先加载/信任/模式再进入 submit loop。在更完整的产品里「agents parallel load」就是 inventory policy 的装配没有这一步query loop 没有稳定工具面可展示给模型或用户。7. Parity Audit清单是「可度量一致性」的锚parity_audit.py将归档侧与当前 Python 树的根文件、目录、命令条目、工具条目做比例统计——命令/工具覆盖率直接绑定commands_snapshot/tools_snapshot与归档dataclass(frozenTrue) class ParityAuditResult: archive_present: bool root_file_coverage: tuple[int, int] directory_coverage: tuple[int, int] total_file_ratio: tuple[int, int] command_entry_ratio: tuple[int, int] tool_entry_ratio: tuple[int, int] missing_root_targets: tuple[str, ...] missing_directory_targets: tuple[str, ...]学习点没有 inventory就没有「条目覆盖率」这种工程指标移植进度会沦为感受而不是数据。8. 结论为什么必须先 inventory 再 I/O结合本仓库维度若先做 I/O先做 inventory本仓库做法命名空间任意字符串都可能触发副作用仅PORTED_*内名字可进入执行链路由难以定义「匹配到什么算合法工具」route_prompt在固定模块集合上打分权限权限逻辑散落在具体 syscallToolPermissionContext、denial 推断附着在模块名与元数据审计/回放日志与真实能力面脱节history / TurnResult 记录「匹配了哪些已登记符号」移植无法做 parity 与 snapshot diffJSON 快照 audit 量化进度演进每加一个工具改多处增删 JSON 条目 → 注册表与路由自动继承claw-code 当前用mirrored shim把真实 I/O 推到清单与路由之后是这一原则的极端清晰演示先把「系统承认什么」钉死再谈「能对世界做什么」。9. 建议阅读顺序动手src/reference_data/commands_snapshot.json、tools_snapshot.json— 感受清单体量与字段。src/commands.py、src/tools.py— 加载、过滤、execute shim。src/execution_registry.py、src/runtime.py— 注册表与bootstrap_session顺序。src/query_engine.py—submit_message如何把「已匹配集合」纳入会话与用量。src/parity_audit.py— 清单与归档的量化对照。