第3章 搭建第一套完整的 harness

第2章给出了首请求程序，并解释了调用与结果的消息关系。现在将模型请求、工具接续与真实操作接成一个实际运行的程序：读取两份材料，让模型选择原文，再创建一个退出后仍然保留的摘录文件。读者既可以核对文件，也可以沿着记录看到输入、提议和操作怎样相接。

文件创建使运行系统承担了新的责任。模型的提议要经过解释和授权检查，写入要留下可核对的记录，交付要依据磁盘上的实际内容。遇到失联、短写或授权变更，程序还要决定在哪里停止，以及怎样说明已经发生的效果。完整的 harness 将这些步骤接在同一次运行中，让模型的选择、程序的动作和最终结果有明确的关联。

这里的完整性以摘录任务为范围。两份来源、两个固定工具和一个产物文件，足以呈现请求与效果之间的主要关系，也允许对失败现场作具体检查。实现采用 Python 3.11+ 和标准库，随书文件为 book/v2/harness.py。正文先说明任务和运行结果，再展开建设过程；完整源码、代码查阅入口与真实验证的证据范围分别列出。

3.1摘录任务与运行结果

任务使用两份随书自编的公开练习材料，标识为 fixture-a 和 fixture-b。fixture 是供程序和测试使用的固定输入样本。模型从每份材料各选一段 8 至 200 字符的原文，按 A、B 的顺序提交，程序将它们写入一个新的 Markdown 文件 card.md。文件只收录原文，不增加摘要、观点判断或其他语义结论。

选择留给模型，文件格式由程序确定。card.md 的表头由程序写成「原文摘录」，后面是说明、两份来源及对应摘录，还包含来源 SHA-256、创建请求和产物版本。SHA-256 是从字节计算的内容摘要，在本章用于关联同一份内容。这份文件的可核对性来自这些关联与实际字节，而材料叙述的真实性属于另一类问题。

从随书项目根目录执行离线演示：

python3 --version
python3 book/v2/harness.py --demo

demo 不需要 API Key，也不安装框架或 SDK。已有实测环境是 macOS、Python 3.13.12；源码要求 Python 3.11 及以上，并要求运行环境支持目录打开和 fsync。fsync 向文件系统请求持久同步，本章用它同步文件内容和新建目录项。其他平台尚未实测，Python 版本符合要求只是运行条件的一部分。

默认输出根目录是 book/v2/runs。也可以进入 book/v2 后执行 python3 harness.py --demo，两种方式都依据源码位置确定默认输出。每次启动构成一次运行（run），程序为它创建新的 runs/<本次运行>/ 子目录；输出根目录的父级须已存在。旧目录中的文件和记录保留，新的运行不会接续旧动作。

终端上有两类输出。标准错误显示进度，标准输出给出 JSON 摘要。摘要中的 mode、status 和 stop_reason 描述运行方式及停止结果；artifact.path 指向产物位置。正常 demo 的 mode 为 scripted，模型响应和选段由脚本预先编排，工具调度、目录创建、文件写入和验收则使用真实实现。

一次运行留下三类值得分别查看的对象。

产物是 card.md。 它在进程退出后仍可打开，包含两项按顺序排列的原文和相关元数据。检查这个文件能回答实际留下了什么，不能单独解释所有中间步骤。

持久事件记录是 events.jsonl。 正常路径中，write_intent、write_result、artifact_verified、final_artifact_verified 和 run_stop 分别记录写前意图、创建结果、创建后验收、最终验收和运行停止。事件的先后关系解释文件怎样成为可交付的结果。

通信材料是请求与响应快照。 文件名以 request- 或 response- 开头。第一份请求还没有来源内容；读取工具的结果进入下一份请求后，模型才具备本章认可的选段依据。请求记录中的 disclosed_sources 表示这次输入实际包含哪些来源。

默认脚本使用三次响应：第一条请求读取两份来源，第二条请求创建这个文件，第三条返回完成语。因此正常 demo 有三次模型脚本请求和三次工具调用。这个顺序说明一条完整路径，循环本身仍按收到的响应工作。合法运行可能分开读取材料，也可能经历参数拒绝；提前结束和未能创建也都有相应结果。实现必须处理这些分支，而不能把三步脚本当成所有运行的固定流程。

3.2目标、授权与验收的分工

摘录要求首先规定了目标：得到一个来源、数量、顺序和长度均符合要求的新文件。材料中符合要求的连续原文片段，即子串，通常有多个，模型在这个范围内选择。任务的自动判断以逐字摘录为准，选段的重要性或代表性没有被纳入成功条件。这样的目标既保留模型参与的空间，又给程序提供可执行的检查依据。

达成目标需要授权。本章的许可覆盖读取两份固定材料、使用其内容、将材料发送给选定的外部模型服务，以及在指定输出位置创建一次新文件。本地读取和对外发送是两个不同动作，二者都需要有相应许可。示例使用自编、公开且无敏感信息的材料，使外发范围可以明确说明；它没有获得扫描或发送读者个人文档的权限。

验收负责判断实际结果与这些要求是否一致。它使用来源快照、合格参数生成的预期字节、当前文件和产物绑定，检查原文、顺序、hash、版本及当前授权。目标描述期望，授权确定可执行范围，验收提供交付依据。三者在同一个任务中合作，但变化的后果各不相同。

例如，材料 A 中的“社区阅览室在入口放置了书目清单。清单按照书名排列，方便读者查找。”满足一个候选摘录的形式要求。材料 B 中的“花园观察记录使用统一表格，分别填写日期、天气和植物名称。”也可以作为候选。程序能确认这些字符确实出现在相应输入中；fixture 的内容用于练习摘录机制，并非作者亲历，也不构成现实阅览室或花园活动的事实证据。

目标改变后，原请求是在旧条件下发出的；授权撤回后，旧文件仍是已经发生的动作留下的结果；文件后来被修改，过去的验收又只对应修改前的观察。要正确处理这些变化，需要同时保留动作的来历和当前条件。

第一版由 Coordinator统一持有目标文本、联合修订号 revision 和停止原因。目标与授权共同使用这个修订号，初始值为 1。每次请求保存发出时的 revision，创建动作和文件版本沿用它，响应返回后再与当前状态核对。联合编号减少了这一任务所需的状态组合，也意味着接受新条件时先停止旧任务，而不是建立任意目标的在线切换系统。

3.3固定工具与执行范围

当前任务需要两种能力：取得已绑定的来源内容，以及用两项合格摘录创建文件。接口相应收窄为 read_source 和 create_card。两个工具均由本地固定实现执行，模型提交名称和参数，程序决定提议是否可执行。

read_source 的参数只有 source_id，例如 {"source_id":"fixture-a"}。这个值命中启动时绑定的来源，不参与路径解析，也不指向网址。create_card 的参数只有 entries，其中恰好两项，每项只有 source_id 和 quote，顺序固定为 fixture-a、fixture-b。模型选择 quote，工具生成标题、来源小节、引用前缀和元数据，写入位置始终是当前运行目录的 card.md。

固定接口是由任务需要决定的取舍。通用 read_file(path) 会引入路径解析、符号链接和目录变化等问题；write_file(path, content) 还会引入覆盖许可、写入位置和任意内容验收。shell 则允许进程调用更广的能力。摘录任务没有使用这些自由的必要，因此第一版将它们留在接口之外，集中建设选段、写入和验收之间的关系。

窄接口仍有真实效果。创建工具使用实际文件系统，产物不会在退出时自动清理，也不是只输出一段准备写入的文本。固定内容和位置让文件效果有清楚的边界，失败后的现场也能够与写前预期比较。

这套实现依赖明确的信任前提：Python 进程、标准库、固定工具代码和输出目录都可信，并假设没有其他进程并发篡改该目录。模型没有执行 Python、动态导入模块或启动程序的工具。由此得到的是固定工具效果的约束，任意不可信程序的 sandbox 是另一种能力；执行器或库被替换、目录遭到并发攻击，都超出本章保证范围。

使用标准库有两项直接好处：运行依赖少，调度与效果的责任位置可直接查看。框架可以改变代码组织，但工具名接收、参数检查、写入和结果确认仍需要实现。本章将这些责任保留在可查阅的本地代码中。

3.4来源快照与创建请求的输入依据

读取工具首先需要回答“这个来源标识对应哪份内容”。Harness 初始化时取得两份 fixtures 的字节副本，解码为文本并计算各自的 SHA-256。此后工具返回、原文检查和来源 hash 使用同一组启动快照。一个来源名在本次运行中的含义由这组字节确定。

每份快照上限为 4096 字节，超过就拒绝初始化。这里选择拒绝而非截断，是因为截断会改变作为摘录依据的材料范围。小任务可以直接限定输入大小，无须同时引入长文折叠和部分来源的说明规则。

快照保存在运行内存中，并没有额外生成两份独立来源文件。read_source 返回完整文本和来源 hash，成功结果进入消息历史，再随下一次请求提交。因此核对模型实际收到的材料，应查看请求快照中的工具结果。源码中的 fixture、内存中的已读状态与实际发出的模型输入，各有不同用途。

工具契约也在启动时绑定。TOOLS 向模型描述工具名称、说明和参数 schema；schema 是约定输入结构的描述。MAPPING 保存本地工具映射标记。程序复制二者，在每次请求意图中记录映射、schema 及 schema hash。同一 run 内使用固定副本，没有热更新；返回提议按照发出请求时的绑定处理，避免同名工具的解释在往返中改变。

schema 为模型提供参数约定，本地执行器将它落实为检查。additionalProperties: False 对应字段集合精确匹配，额外字段同样会被拒绝。创建工具检查恰好两项的 entries 列表，并逐项检查字段、顺序、长度和原文。以下为 Harness.tool的节选：

for i, entry in enumerate(args["entries"]):
    if not isinstance(entry, dict) or set(entry) != {"source_id", "quote"} or not isinstance(entry["quote"], str):
        return reject("schema_error")
    if entry["source_id"] != SOURCE_IDS[i]:
        return reject("wrong_source_or_order")
    quote = entry["quote"]
    if not 8 <= len(quote) <= 200:
        return reject("quote_length")
    if quote not in self.texts[SOURCE_IDS[i]]:
        return reject("verbatim_missing")

外层先确认参数对象只有 entries，且它是恰好两项的列表，再由这段代码处理每项摘录。len(quote) 使用 Python 字符串长度，即字符数，区别于 UTF-8 字节数和 token 数。子串检查要求字符、标点及换行与来源一致，因此近义改写不满足原文要求。

通过这些检查还不足以确认选段的输入依据。模型可能在一条响应中同时提出读取 A、读取 B 和创建文件。串行执行时，创建动作排在两次读取之后；但整条响应是在读取结果返回给模型之前生成的。工具执行的先后并不会改变这条响应原来依据的输入。

实现用两个集合描述这一区别。read_sources 表示本地读取工具已经成功返回的来源；disclosed_sources 表示产生当前响应的实际请求已经包含的完整读取结果。创建需要两者都齐备：未读齐返回 sources_not_read，已读齐但尚未进入当前请求返回 sources_not_disclosed。同一响应里的读后创建会被拒绝，读取结果随下一次请求发送后，模型才能重新提出创建。

已发送来源集合由实际载荷推导（源码中为 disclosed_sources）。循环先编码即将发送的 JSON，再解析其中的 messages，查找成功的 tool 结果，核对来源标识、完整内容和来源 hash 是否与启动快照一致。得到的集合与本次 request_id 一起绑定；创建时保存该请求，以及两份来源进入请求的记录。只发送 A 尚未达到条件，A 和 B 都进入产生创建提议的请求，才满足输入要求。

这一机制确认材料进入了哪次模型输入，而不把输入关系扩大为理解程度的证明。对于逐字摘录，来源快照、实际请求和候选子串已经建立了可检查的依据；理解和选择质量可在另有评价标准的任务中处理。

3.5响应协议、整批检查与逐项调度

模型返回内容经过几个不同阶段才会到达工具。Harness.loop固定构造 DeepSeek Chat Completions 请求，模型名为 deepseek-flash，显式设置 thinking={"type":"disabled"}、stream=False 和 max_tokens=1024。采用非流式返回，使程序能取得并检查完整响应后再调度，文件效果不会从尚未收全的消息中启动。

返回字节先保存为经脱敏的响应快照并记录事件。随后程序取得小型元信息、处理 usage（接口报告的 token 用量），经过响应后的控制检查点，再调用 parse_response。消息历史接收的是检查后的助手消息、必要正文和合法调用关联字段，供应商其他字段按各自用途保存，不将整份返回直接作为下一次输入。

JSON 解释与字段语义

外层响应是 JSON，工具 arguments 又是装有 JSON 的字符串。这两层使用同一解释口径。decode_json拒绝重复字段、NaN 和 Infinity 等非标准常量，并检查指数形式的数值转换后是否成为非有限浮点数；encode也禁止输出非有限数值。重复键没有被静默覆盖，数值也不会在不同环节以不一致的含义继续处理。

能精确解码与符合工具参数是两项判断。例如，arguments 为字符串 "null" 时，内部 JSON 合法，解码结果为 None，随后由工具返回 schema_error。若字段本身是 JSON null，则已经违反“arguments 必须为字符串”的响应协议，在调度前拒绝。字段缺失、空字符串与 null 的含义，要放在具体字段的要求中解释。

解析还检查只有一个 choice、消息角色为 assistant、正文类型合法，以及结束原因与调用情况相符。当前空值和结束行为如下：

工具调用编号用于配对提议与结果，要求为 1 至 128 个字母、数字、下划线或连字符。同批编号不能重复，也不能与此前记录的 tool_call_id 重复。名称必须是字符串，调用类型必须为 function，参数则必须完成内部 JSON 解码。协议检查建立了可调度的提议结构，业务合格性仍由工具决定。

整批检查与业务拒绝

调度顺序是先完整解析一批调用，再检查整批是否落在剩余额度内，最后接受并开始执行。第二项的参数 JSON 若残缺，第一项即使结构合法也不会提前执行。整批工具额度不足时，同样在效果发生前停止。这个时点协议可以是 validated，助手消息却尚未成为 accepted；两个状态分别表达检查通过和正式接受。

业务拒绝发生在逐项执行中。参数 JSON 合法但缺少 source_id，或 quote 不满足原文要求，都能得到带原调用编号的明确结果，后续合法调用仍可继续。这让模型有机会在下一次输入中看到拒绝原因并修正提议，协议错误与可解释的参数错误因而有不同处理。

未知工具采用更严格的停止策略。名称未命中固定绑定，当前调用返回 unknown_tool；后面的调用逐项获得 batch_stopped，保持编号配对，却不再启动动作，批次处理后运行停止。执行器只使用固定分派，不按名称反射查找函数、推测替代能力或执行命令。

这些规则没有组成批次事务。未知工具之前已经完成的读取或创建仍然存在，停止阻止的是后续效果。整批结构错误能够在动作前拒绝整个批次，业务拒绝则根据具体结果决定是否继续。正常接受时，助手调用消息先进入账本与历史，每个处理完成的调用再产生匹配的 tool 消息，之后才可能提交下一次请求。记录失败会使运行停止，配对要求不会掩盖实际记录缺口。

通信记录与协议状态也分别保留。model_response.state 表示响应处在等待、收到未持久化、已持久化或传输失败等阶段；protocol_status 表示是否通过验证并被接受。已保存的响应可以被协议拒绝。摘要中的 model_response_unknown 表示响应持久记录未完成，model_protocol_unknown 表示协议未验证或被拒绝，因此确认请求已经不在途，仍需继续确认提议是否可用。

3.6写入意图与新文件的持久创建

本次 run 已有产物绑定时，新的创建提议返回 already_created，一个 run 最多创建一次产物。失败进入未知状态后，运行停止并保留现场。工具只有通过参数、来源发送和当前 revision 条件，才会为首次创建生成预期 UTF-8 字节。

标题、说明和来源小节来自程序固定版式，quote 每行加上引用前缀，来源 hash 来自启动快照。模型只决定合格选段，不提供最终 Markdown、路径或可信 hash。

Card以不可变元数据绑定路径、run_id、action_id、revision、版本、预期内容 hash 和字节数。版本形式为 <run>/<action>/r<revision>。建立这个对象时，磁盘文件可能尚不存在；它首先描述准备创建的产物。摘要是否包含 artifact 与真实创建是否完成，需要分别判断。

写文件和写账本可能分别失败，因此创建有明确的持久顺序。write_intent 先进入 Ledger，记录目标、预期 hash、预期字节数、版本、action、revision 和创建请求。追加检查完整写入，刷新并执行 fsync，成功返回后才启动文件写入。关键顺序为：

self.ledger.append("write_intent", action_id=action, target="card.md", intended_content_hash=self.card.content_hash,
                   intended_content_bytes=self.card.content_bytes, version=version, goal_authority_revision=revision,
                   request_id=self.card_request_id, disclosed_sources=list(self.card_disclosed_sources))
self.effect_in_flight = True
create_new(self.card.path, self.card_bytes)
self.ledger.append("write_result", action_id=action, target="card.md", content_hash=self.card.content_hash,
                   content_bytes=self.card.content_bytes, version=version, outcome="created")
acceptance = self.verify_artifact()
self.ledger.append("artifact_verified", action_id=action, **acceptance)
self.acceptance = acceptance
self.effect_in_flight = False

写前记录给后续核对建立了起点。如果先留下文件再记意图，进程可能在两者之间停止，产物就缺少准备写入内容的关联。意图先持久化，至少能回答目标是什么、预期是什么、动作属于哪次请求；实际写入是否发生，还要继续看文件和结果记录。账本追加一旦失败即被标记为不可继续使用，运行以 ledger_unknown 停止，写前失败不会越过账本直接创建文件。

create_new使用 O_WRONLY | O_CREAT | O_EXCL 和 0o600 创建文件。O_EXCL 要求目标为新对象，已有对象不会被覆盖。实现进行一次 os.write，核对返回字节数，随后同步文件并同步父目录。文件同步对应内容，目录同步对应新文件名与对象的关联，两步共同构成本章声明的持久创建过程。

目录创建遵守同样的范围约定。输出根目录的父级必须已存在，程序只在需要时创建根目录本身，再独占创建本次 run 子目录，并同步相应父目录。源码没有递归创建与同步整串祖先的逻辑，目录同步失败就停止。这也解释了运行前要求父目录存在的原因。

写入边界的失败不能都归为“没有文件”。短写可能留下部分内容；文件或目录同步失败时，完整内容也可能已经可读，但约定的持久过程没有得到确认。实现将这些情况报告为 artifact_unknown，保留现场，不补写、不删除重建，也不重试。文件可见与持久提交完成是不同事实。

文件完整写好以后，write_result 或验收事件仍可能写入失败，结果为 ledger_unknown。实际字节可以帮助核对发生了什么，却不能替代缺失的可靠记录。创建、中断和记账分别处在哪个边界，决定了停止时能够承诺到哪里。

重新启动总会使用全新 run。程序不扫描旧账本、不认领旧动作，也不提供自动恢复。O_EXCL 防止当前目标覆盖，但不会跨 run 去重：旧运行可能已经创建文件，新运行仍可再创建一个。未知结果后应先人工检查旧目录，再决定是否重新发起任务。新的运行编号只区分运行，不保证业务效果恰好一次。

第一版因此先建立记录和停止方式，把恢复留在范围之外。账本是持久证据，不是自动恢复器；安全地继续旧动作还需要接管、重试和去重条件。这些条件尚未实现时，停止能避免在不清楚已有结果的情况下继续扩大效果。

3.7当前文件验收与运行结束

创建后的验收由程序重新打开 card.md 完成。“独立”指判断依据独立于模型完成语：它使用预先绑定的预期字节和产物绑定检查实际文件，并非调用第二个 agent 或模型给产物评分。

内容约束在效果前后衔接。写前确认 quote 是对应快照的精确子串，并满足长度、来源和顺序；写后确认当前文件正是这些合格参数生成的内容。固定版式使两步可以用字节关联，免去了任意 Markdown 所需的额外语义判断。

Harness.verify_artifact最多读取预期字节数加一。少写会不一致，多写一个或更多字节也会被发现。随后同时核对实际与预期字节、hash、版本、路径、run、字节数、读取集合、创建请求的发送来源和当前授权：

checks = dict(content_check=actual == self.card_bytes, hash_check=sha(actual) == c.content_hash,
              version_check=c.version == expected_version and f"- version: {expected_version}\n".encode() in actual,
              binding_check=c.run_id == self.run_id and c.path == self.directory / "card.md" and c.content_bytes == len(actual),
              sources_check=self.read_sources == set(SOURCE_IDS),
              disclosure_check=self.card_disclosed_sources == SOURCE_IDS and self.card_request_id is not None,
              current_authority=c.revision == self.coordinator.revision and self.coordinator.reason is None)
if not all(checks.values()):
    raise Stop("artifact_verification_failed", "artifact_unknown")

字节比较确认实际内容，hash 关联原意图，版本和绑定指明产物属于哪次动作，revision 则确认当前条件仍适用。文件变化后为它计算一个新 hash，只能描述新内容；验收要求的是与原来绑定的预期一致。当前授权也需要成立，历史动作身份不能代替当前许可。

创建后验收通过并可靠记录，工具返回 verified=True。接下来结果送回模型，模型以无工具调用、finish_reason=stop 的非空白文本结束。最终响应到达时再次读取并验收，记录 final_artifact_verified，随后尝试持久记录成功停止。因此创建完成、创建后验收、模型给出结束语和成功停止之间仍有步骤。

几个分支体现了这种分工。模型正常结束却没有产物，结果为 finished_without_artifact；已有产物却返回 null、空字符串或全空白最终正文，协议仍不能正常结束。若创建后已有验收，后来因模型次数、工具次数或载荷预算停止，程序会重新验收，成功时报告 artifact_created_not_finished。这个状态保留已确认的产物，也明确接续没有完整结束。

摘要应结合当前状态和历史验收阅读。顶层 status、stop_reason 说明停止结果；artifact.verified_at_stop 表示本次停止分支是否确认产物，currently_authorized 表示产物是否仍属于当前有效授权；acceptance 保留具体检查和绑定。历史 acceptance 或 outcome 可以继续表明此前发生过验收，却不能替当前状态宣告成功。

创建后撤权就是一个例子。文件保留，旧验收也保留，但 revision 已经改变，currently_authorized 失效，verified_at_stop 不会凭旧验收变为 true。另一方面，停止后的外部修改仍须重新核对：这个标志描述停止分支的判断，不是持续监视文件的机制。

摘要查阅：预算停止分支

当前实现的预算停止分支会复验，但没有为这次复验另存完整验收事件。摘要中的 acceptance_scope 固定写着“最后一次已持久化的验收”，在这个分支不能准确描述内存复验与持久事件的关系。核对时应一起查看具体停止状态、已有验收事件与 run_stop，不单靠这一字段判断是否有新的持久回执。

3.8条件修订的生效时点与旧响应失效

目标和授权可以在运行中变化。交互式 demo 使用同一实现呈现条件修订过程：

python3 book/v2/harness.py --demo --interactive

响应返回并处理到控制输入这一步后，终端提示输入。Enter 继续，goal:<新要求> 接受新目标并停止旧任务，revoke 撤回授权，cancel 取消。生产交互与测试 hook 都通过 Coordinator.accept改变状态，共用同一个入口。

接受新条件也以持久记录为先。程序将 control_accepted 写入账本，再增加 revision、更新目标与停止原因，然后核对请求发出时保存的 revision。新条件已经生效，旧响应就不会进入调度，即使其中的创建参数原本完全合格。修订号使请求依据与当前许可可以直接比较。

goal: 的能力是接受变化并结束旧任务，而不是原地执行任意新目标。新要求可能改变工具与验收，单独替换目标文本不足以建立这些能力；再次启动本章 CLI，执行的仍是固定摘录练习。第一版选择联合修订号和停止旧任务，保持了条件变更的明确含义。

程序读取控制输入的位置由 Harness.control_boundary 决定。网络请求同步进行，在途时不读终端命令；同批工具之间也不再读。用户敲键与程序读到命令并非同一时刻。--interactive 因而提供响应后、工具调度前的粗粒度控制，不是实时撤权、异步抢占或持续授权监听。

若文件已经创建，到最终响应返回时才接受 revoke，旧文件不会被删除。程序同时保留已发生动作和当前授权失效这两个事实。已提交效果没有回滚，运行也不再以当前要求的成功结果交付旧文件。

Ctrl+C 根据在途对象分类。请求尚未取得可靠响应时，中断为 model_unknown，服务是否已经接收或计费仍未确定；写入效果尚未可靠确认时，中断为 artifact_unknown，现场可能有部分或全部内容。没有这两类在途状态时，可以作为普通取消处理。终端 cancel 则经协调者接受，其停止原因和摘要分支应按实际记录阅读。

这样的停止优先限制后续效果，并说明已经知道和仍不确定的部分。它能使旧响应失效，却不自动提供撤销、补偿或抢占；这些能力各自需要新的动作条件。

3.9预算对象与停止边界

模型可能重复读取、修正参数或继续提出动作。运行投入需要上限，而且上限应对应它所约束的对象。本章分别限制请求次数、工具次数、来源大小、提交载荷、响应读取量和输出 tokens：

每次请求重新编码完整历史，包含用户消息、助手调用、工具结果和 schema。24000 字节针对即将发送的完整 JSON，不只针对新正文，也不是精确 token 上界。重复读取和拒绝结果不断加入历史，即使模型每次回答很短，后续载荷仍可能超限。

载荷超限时直接停止，来源超限也直接拒绝。实现保留已声明的输入范围，不在预算检查中暗中摘要材料或删除调用配对。长任务的历史选择和压缩可以另外建设，当前拒绝策略让输入依据保持可解释。

工具次数按进入已接受批次处理的调用计数。参数被拒绝、重复读取，以及未知工具之后产生 batch_stopped 的项都消耗额度，计数不是成功次数。整批放不下则在接受与执行前停止，避免只做半批后因次数退出。这项额度判断与业务拒绝后是否继续是两种规则。

6 次模型请求表示最多重新提交六次当前历史，区别于六个用户 turn。max_tokens 约束输出 tokens，不能约束整份响应 JSON 字节；响应字节上限也不能推导完整计费输入。将对象与数值一起说明，才能判断限制实际覆盖了什么。

30 秒 timeout 同样不能乘以 6 后当作任务总时长。终端等待、文件同步、解析以及多次网络操作都参与运行，源码没有总时钟、进程级资源隔离或任意工具强制终止。标准库网络参数提供的是局部等待约束。

预算耗尽只决定停止投入。已有创建后验收的产物按前述分支复验并保留产物事实，没有产物就保留未创建的结果；程序不会代替模型补出未执行的摘录，也不会为了让运行结束而降低验收要求。

3.10用量记录与日期化费用基线

预算控制未来投入，usage 描述已经取得的用量信息。一次动作失败可能已有模型用量，一次文件创建成功也可能没有可靠费用信息，因此二者分别累计。

本章沿用第2章的日期化基线：2026年11月1日，人民币，每百万 tokens。默认 peak 的缓存命中输入、未命中输入和输出单价分别为 0.04 元、2 元、8 元；显式选择 --tariff offpeak 时分别为 0.02 元、1 元、4 元。

这些数字服务于全书实践的统一比较，是固定基线投影。源码没有实际价格查询或自动账单估算，也不会按请求日期、当天时段或节假日选择 tariff。request_time 是本地请求意图记录的 UTC 时间戳，供事后核对，区别于服务端确认的计费时间。实际账户支出仍需依据请求当时的适用规则另行核对。

Usage分别累计 prompt_tokens、completion_tokens、prompt_cache_hit_tokens 与 prompt_cache_miss_tokens。相关已报告字段必须为非负整数，布尔值不按整数接受；缓存命中与未命中都报告时，二者之和必须等于输入总量。金额用 Decimal 计算，将缓存命中、未命中和输出数量分别乘单价，相加后除以一百万。

输入与输出总量有效，但缓存细分未报告完整时，程序把全部输入按未命中计算，标记 cache_missing_all_prompt_assumed_miss。这是一项保守假设，不是观测到缓存命中为零。已经出现的缓存字段若为 null、负值或非整数，仍属于畸形值并拒绝，不能利用另一个字段缺失跳过检查。

整份 usage 缺失或为 null，则本次用量未知。其他请求的有效累计和 known_usage_cost 保留，但 total_cost 为 null，unknown_cost 为 true。空对象缺少有效输入输出总量，属于 usage_malformed；字段 null 也不等于零。网络失联时本地没有取得用量，未知费用同样不能补为零。

默认 demo 没有报告实测 usage，usage.origin 明示 scripted_fixture_not_measured。它可以创建真实文件，但不是一次零元真实模型调用。测试合成的 token 数只检查累计与费用算术，不能证明缓存命中率或账单。本章尚未完成真实 API 联调，因此不提供 live 请求次数、真实 token 数或实际支出，也不以产物字节数推算全书成本。

3.11真实模型路径与通信记录

--live 是需要显式选择的运行路径。使用者应具备账户和有效 DEEPSEEK_API_KEY，允许将两份材料发往选定服务，并接受付费请求。CLI 要求选择 demo 或 live，已有环境密钥不会自动启用付费模式。

下面保留真实调用方式，供在项目根目录的交互终端使用。执行它会主动发起付费 API 请求，现有验证尚未执行这一步。

python3 -c '
import getpass, os, subprocess, sys
if not sys.stdin.isatty():
    raise SystemExit("请在交互终端运行。")
env = os.environ.copy()
env["DEEPSEEK_API_KEY"] = getpass.getpass("DeepSeek API Key: ")
raise SystemExit(subprocess.call(
    [sys.executable, "book/v2/harness.py", "--live", "--interactive"], env=env))
'

外层 Python 隐藏读取密钥，通过子进程环境传递，避免将密钥写进聊天、源码、命令参数或 shell 历史。--live 默认读取环境变量，--key-stdin 可以覆盖，但仅供 live 使用；与 --interactive 合用须有真实终端，管道密钥不能同时占用交互控制输入。

HTTPTransport只请求固定 HTTPS Chat Completions 目标，禁用自动重定向和环境代理，并且不自动重试。API Key 只用于请求头和脱敏，不进入提示词或工具结果。当前工具没有 shell 或继承环境的外部程序入口；以后若增加外部工具，需要单独构造不含模型凭证的工具环境。

请求和响应保存前经过脱敏。账本为快照记录相对路径、hash、字节数和脱敏标志，请求意图另外保留实际载荷 hash。脱敏改变了快照时，保存文件的 hash 与原始载荷 hash 对应不同字节，核对时必须区分。快照提供通信内容的检查材料，但仍在同一个可信目录中，不是防篡改的独立审计服务。

错误信息也有记录范围。HTTP 错误体在限定读取量内处理，脱敏后最多保留 512 字符；网络异常、工具读写错误和异常诊断输出前同样脱敏。规则覆盖已知密钥及若干凭证形式，不能据此推断所有敏感内容都受到保护。hash 不提供访问控制，输出目录的可信与保密要求仍然存在。

供应商元信息按观测单独保存。请求配置为 deepseek-flash，不能据此补出返回模型版本或指纹。实现只接受长度不超过 256 的小型、可打印 ASCII 模型名或指纹，以及非负整数生成时间；缺失、null、空值或不可用值保留 null。scripted 元信息明确标记脚本来源，与供应商返回分开。

现有 HTTP 测试使用 fake response 或 fake HTTPS handler，可以检查固定目标、请求头、读取上限、timeout 参数、拒绝重定向和错误脱敏。真实服务可用性、返回兼容性、模型自主选段及账户消费仍须 live 联调。代码表明这条路径如何构造，真实调用证据则需要实际执行后取得，两者在本章分别说明。

3.12源码结构与查阅入口

正文中的三段 Python 节选分别说明选段检查、持久创建和当前验收，不组成第二份可运行程序。完整实现保存在 book/v2/harness.py，章末也逐字附录。代码查阅可以从循环的责任关系进入，再定位具体机制。

用户目标与授权 → Coordinator：当前 revision
                         ↓
Harness：材料快照、工具绑定、预算、消息历史
  请求快照／意图 → HTTPTransport → 响应快照／usage
                                      ↓
                           响应后控制边界
                                      ↓
                           完整解析调用批次
                                      ↓
                read_source 或 create_card
                                      ↓
             真实文件 → 独立验收 → 下一次请求
                                      ↓
                    正常结束响应 → 再验收 → 停止记录

Ledger 持久记录上述关键边界；不是自动恢复器。

这幅图表达结构，不代替一次实际轨迹。Harness 将来源、请求、提议、效果和停止条件连接起来；HTTPTransport 负责通信，Usage 负责用量与基线计算，Card 绑定产物，Ledger 持久记录关键边界。具体运行中可能在任何声明的失败或控制分支停止，事件记录才说明实际经过了哪些步骤。

消息历史与账本保存的信息有交集，用途却不同。历史让模型看到先前调用及结果，维持下一次输入的接续；账本让进程结束后的核对能够定位写入开始、结果确认和失败位置。聊天记录支持模型交互，解释外部效果还需要动作边界的持久记录。

编号也对应不同对象。run_id 标识一次启动，request_id 标识一次模型通信，tool_call_id 配对模型提议与工具结果，action_id 标识一次创建副作用。版本进一步将 run、action 和 revision 关联起来。请求失联、编号重复或旧文件保留时，这些身份使核对能够指向具体请求与效果，而非一个含义混杂的任务号。

阅读某项机制时，可以将正文中的设计原因与对应代码对照；确认它在一次运行中是否成立，则查看该 run 的实际文件、请求快照和事件。源码查阅、故障测试和真实服务联调提供不同层面的依据。

3.13本地测试与既有验证记录

随书测试文件为 book/v2/test_harness.py。下面的命令从项目根目录运行本章测试，会执行本地检查并留下新的证据目录；后面的验证记录则对应此前执行的结果。

python3 -m unittest discover -s book/v2 -p test_harness.py -v

测试用 scripted 响应驱动同一 Harness，在项目内创建真实目录和文件。每次证据保留在独立的 test-evidence-… 目录，包含正常产物和失败现场，不自动清理。这样既能检查返回状态，也能在测试结束后读取字节与事件。脚本 quote 由测试编写者指定，工具真实执行与模型自主选择仍然分属不同证据。

测试围绕建设中的关键边界安排。正常路径核对原文、固定顺序、来源 hash、版本和调用配对；参数用例覆盖额外字段、错误来源或顺序、长度、非原文和未读齐。来源发送用例再检查同批读后创建、只发送一份来源，以及不能从可变的读取集合冒称材料已经进入请求。

协议与批次用例将错误放在不同位置：畸形参数 JSON 阻止本批第一项提前执行，重复编号拒绝，整批额度不足不启动动作，未知工具之后给出带编号的 batch_stopped。它们检验的是具体调度规则，而不仅是程序能否退出。

持久性用例检查写前账本失败无文件效果、短写保留部分文件且不重试、文件或目录同步失败不报告成功，以及效果后账本失败保留未知。最终验收前修改内容、版本或 hash 元数据，用于确认旧验收不能替变化后的对象继续作保证。这些构造故障覆盖声明的分支，没有穷尽所有文件系统故障。

控制用例共用生产环境的同一入口，检查目标变更、撤回、取消、旧响应不调度、旧文件保留和当前授权失效。中断用例区分网络与写入在途。最终正文的缺失、null、空字符串和全空白，以及 usage 算术、缓存假设与未知费用，也分别由测试约束。

已验证记录：环境、结果与证据范围

以下是2026年11月1日针对当前随书实现取得的验证记录。环境为macOS（Darwin 24.5.0，arm64）、Python 3.13.12。测试进行真实文件创建、读取和同步，模型及HTTP响应由测试构造，没有使用真实API Key。

这次基础套件的完整测试命令为 python3 -m unittest discover -s book/v2 -p "test_*.py"，返回110项全部通过，构成为首次请求程序12项与本章harness 98项。110对应当时这两份测试，不代表随书目录加入其他模块后的全部测试数量，也不是系统正确性的总分。

既有 demo 目录保留了 card.md、三对请求／响应快照和 events.jsonl，可核对两份材料进入请求的变化，以及写前意图、创建结果、两次验收和成功停止的顺序。748 字节是那次脚本选段与元数据形成的产物大小，合法的文件并不都具有这个长度。

本地构造测试说明程序在所列输入和故障下的行为，fake HTTP 不验证服务可用性，合成 usage 不验证实际计费，scripted 摘录不验证模型自主选择。修改代码后需要为相应边界取得新证据；本章保留的历史记录不能自动适用于变化后的程序，也不能把固定工具扩大为任意代码隔离。

本章判据

一套完整的摘录 harness，应能将任务要求与一次运行留下的结果逐项对应。判据既覆盖正常路径，也覆盖无法确认或继续执行的边界：

• ☐目标、读取与外发许可、创建范围分别明确，验收只承担已声明的内容与绑定检查。
• ☐模型只选择固定来源和 quote，工具实际创建持久的新文件，读写路径和任意程序执行不进入提议接口。
• ☐来源与工具在启动时绑定，请求记录 schema、映射和 revision，两份材料的发送情况从实际提交载荷推导。
• ☐响应及整批调用完整解析，精确 JSON、非有限数值、missing、null 和空值按字段语义处理，整批额度在动作前检查。
• ☐正常接受并处理完成的调用有配对结果；未知工具停止后续动作，先前效果保留；记录失败如实停止。
• ☐write_intent 在效果前完成 fsync，文件以独占方式新建且不覆盖，短写、同步失败、记录失败和中断保留已知与未知的区别。
• ☐未知状态停止并保留现场，没有自动重试或跨重启接管；当前文件独占不被误解为跨 run 业务去重。
• ☐验收重新读取当前字节，关联预期内容、hash、来源、顺序、创建请求、run、版本和当前授权。
• ☐创建与运行结束分别表达，status、verified_at_stop 和历史 acceptance 结合阅读，旧文件与预算停止后的产物事实均得到说明。
• ☐目标及授权变化通过同一个入口与 revision 生效，响应后输入时点的时间范围明确。
• ☐请求次数、工具次数、字节限制、输出 tokens 和网络 timeout 各有对象；未知 usage 不补成零，费用保留日期化基线口径。
• ☐scripted、本地故障测试与真实 API 联调的证据分开，验证数字指向对应记录，未发生的 live 用量和支出不被补写。

这些条件把有限任务接成了完整运行：模型获得材料并提出选择，程序依据绑定和当前条件调度，真实效果留下持久记录，交付再由当前文件确认。完整性来自这条关系可执行、可停止、可核对，而不是工具数量或源码篇幅。

本章没有回答什么

逐字摘录没有覆盖摘要准确性、观点可信度和选段代表性。来源 hash、版本与发送记录解决内容及请求的关联，语义质量仍需另行定义证据和评价方法。研究报告、事实核查或综合写作不能直接套用这个摘录文件的成功条件。

本章也没有建立任意代码 sandbox、内核级资源隔离、恶意库防护或面对并发攻击者的文件系统安全。固定工具运行在可信进程与可信目录中；shell、动态插件、任意目录输入和第三方工具会改变这些前提，不属于当前任务未说明的附加能力。

恢复、重试协议、跨 run 业务去重、效果补偿和实时撤权尚未实现。写前账本提供核对材料，不能自动认领旧动作；同步 timeout 也不是任务 deadline。未知后人工检查旧文件和记录，是这版程序明确的处理方式。

长任务的历史选择与压缩、其他模型接口以及 thinking 接续也在本章范围之外。材料和载荷超限直接拒绝，非空 reasoning 返回按模式不匹配停止。DeepSeek live 路径尚未联调；预算复验与 acceptance_scope 的记录差异仍保留，历史验收必须结合当前停止分支理解。

这些范围使下一步建设的问题更具体：循环中哪些值描述当前条件，哪些记录说明已经发生的事实，哪些检查允许启动下一步。第4章将以本章运行为基础整理状态与事实的关系，而不是先扩充工具。

源码附录：完整随书实现

下面原样保留 book/v2/harness.py 的完整 718 行源码，与正文三个节选属于同一实现。首次请求程序和两份测试文件也位于代码包的 book/v2，本章不内嵌测试源码。

已有随书项目时，直接运行项目中的文件。需要从本章保存实现时，只保存下方 Python 代码块的完整内容，并保持 book/v2/harness.py 的随书相对位置。三个短节选用于解释局部机制，不能独立运行。查阅具体机制可使用 3.12 的入口表；正文与证据说明已在前面完成，附录供保存和定位，不要求按行顺序阅读。

"""第3章：两个固定工具的摘录任务，Python 3.11+，无第三方依赖。
仅实现声明的效果，不是任意代码 sandbox；进程、库、输出目录及其无并发篡改假设都可信。
只在同步请求返回后的控制边界接收纠正，不能撤销已提交动作；每次启动全新 run，不支持恢复。
自动验收只证明原文、来源、顺序、版本和材料已发送，不证明模型理解、摘要正确或观点真实。
output_root 的父目录必须已存在且可信，只创建并同步一个新的根目录，不递归创建祖先。
model_response_unknown 指响应持久记录未完成；protocol_status 单独区分未验证、拒绝和已接受。
24000 是 UTF-8 JSON 载荷字节预算，不是 token 精确上限；max_tokens 仅约束输出。
thinking 禁用；非空 reasoning_content 按模式不匹配停止，终结必须有非空文本及已验收产物。
费用只是2026-11-01固定价格基线投影，本书代码没有实际价格查询或自动账单估算。
"""
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import re
import sys
import traceback
import urllib.error
import urllib.request
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal

ENDPOINT = "https://api.deepseek.com/chat/completions"
RESPONSE_LIMIT, PAYLOAD_LIMIT = 128 * 1024, 24000
SOURCE_IDS = ("fixture-a", "fixture-b")
FIXTURES = {
    "fixture-a": ("随书自编公开练习材料 A（fixture，非个人亲历）。\n"
                  "社区阅览室在入口放置了书目清单。清单按照书名排列，方便读者查找。\n"
                  "管理员每天闭馆前核对借阅登记，并把归还的书放回对应书架。\n").encode(),
    "fixture-b": ("随书自编公开练习材料 B（fixture，非个人亲历）。\n"
                  "花园观察记录使用统一表格，分别填写日期、天气和植物名称。\n"
                  "记录员只抄录观察到的现象，不在表格中添加原因推断。\n").encode(),
}
GOAL = "读取两份 fixture，各选一段8至200字符的原文，按 fixture-a、fixture-b 顺序创建全新摘录文件；不生成语义结论。"
SYSTEM = ("你执行固定摘录任务，只能使用所列两个工具。材料是随书自编公开 fixture，非个人亲历。"
          "必须先逐份读取，等读取结果进入下一次请求后再选择 exact substring；同批读取后创建会拒绝。"
          "每源一段，按固定顺序提交；发送不代表理解。"
          "最多创建一次卡，不作摘要、观点判断或语义结论。创建成功后用简短完成语结束。"
          "工具拒绝不是成功；不得把自然语言声称完成当作文件完成。")
TOOLS = [
    {"type": "function", "function": {"name": "read_source", "description": "读取启动时绑定的可信材料快照，不接受路径。", "parameters": {
        "type": "object", "properties": {"source_id": {"type": "string", "enum": list(SOURCE_IDS)}},
        "required": ["source_id"], "additionalProperties": False}}},
    {"type": "function", "function": {"name": "create_card", "description": "两源读取结果必须已进入本次请求；同响应读后创建会拒绝。按固定顺序创建一次全新原文摘录文件。", "parameters": {
        "type": "object", "properties": {"entries": {"type": "array", "minItems": 2, "maxItems": 2, "items": {
            "type": "object", "properties": {"source_id": {"type": "string", "enum": list(SOURCE_IDS)},
            "quote": {"type": "string", "minLength": 8, "maxLength": 200}},
            "required": ["source_id", "quote"], "additionalProperties": False}}},
        "required": ["entries"], "additionalProperties": False}}},
]
MAPPING = {"read_source": "1", "create_card": "2"}


def encode(value):
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8")


def sha(data):
    return hashlib.sha256(data).hexdigest()


def decode_json(raw):
    def unique(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("JSON 重复字段")
            result[key] = value
        return result
    def invalid_constant(value):
        raise ValueError("JSON 非有限数值")
    def finite_float(value):
        parsed = float(value)
        if not math.isfinite(parsed):
            raise ValueError("JSON 非有限浮点数")
        return parsed
    return json.loads(raw, object_pairs_hook=unique, parse_constant=invalid_constant, parse_float=finite_float)


def sync_directory(path):
    fd = os.open(path, os.O_RDONLY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def redact(text, key=""):
    if key:
        for representation in (key, json.dumps(key)[1:-1]):
            text = text.replace(representation, "[REDACTED]")
    text = re.sub(r"(?i)bearer\s+[^\s\"'<>]+|\bsk-[A-Za-z0-9_-]+", "[REDACTED]", text)
    text = re.sub(r"(?i)\bauthorization\s*[:=]\s*[^\r\n\"'<>]+", "[REDACTED]", text)
    return re.sub(r'(?i)("(?:api[_-]?key|authorization|access_token|secret)"\s*:\s*")[^"]*', r'\1[REDACTED]', text)


class Stop(Exception):
    def __init__(self, reason, status="stopped", *, diagnostic=None):
        self.reason, self.status, self.diagnostic = reason, status, diagnostic
        super().__init__(reason)


class LedgerUnknown(Stop):
    def __init__(self):
        super().__init__("ledger_write_failed", "ledger_unknown")


class TransportFailure(Stop):
    def __init__(self, reason, status=None, body=None, *, diagnostic=None):
        super().__init__(reason, "model_unknown", diagnostic=diagnostic)
        self.http_status, self.body = status, body


def create_new(path, data):
    """一次写入并同步文件及目录项；短写、异常、中断留存现场，不得重试。"""
    try:
        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        try:
            written = os.write(fd, data)
            if written != len(data):
                raise Stop("partial_write", "artifact_unknown", diagnostic=f"文件短写：{written}/{len(data)} 字节")
            os.fsync(fd)
        finally:
            os.close(fd)
        sync_directory(Path(path).parent)
    except KeyboardInterrupt:
        raise Stop("new_file_write_interrupted", "artifact_unknown") from None
    except OSError as exc:
        raise Stop("new_file_write_failed", "artifact_unknown",
                   diagnostic=redact(f"{type(exc).__name__}: {exc}")) from None


class Ledger:
    def __init__(self, directory, run_id):
        self.run_id, self.sequence, self.poisoned = run_id, 0, False
        fd = os.open(directory / "events.jsonl", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        self.file = os.fdopen(fd, "wb", buffering=0)
        try:
            sync_directory(directory)
        except BaseException as exc:
            self.file.close()
            raise LedgerUnknown() from exc

    def append(self, kind, **fields):
        if self.poisoned:
            raise LedgerUnknown()
        try:
            data = encode(dict(seq=self.sequence, run_id=self.run_id, event=kind, **fields)) + b"\n"
            if self.file.write(data) != len(data):
                raise OSError("账本短写")
            self.file.flush()
            os.fsync(self.file.fileno())
            self.sequence += 1
        except BaseException as exc:
            self.poisoned = True
            raise LedgerUnknown() from exc

    def close(self):
        self.file.close()


class Coordinator:
    """单一授权持有者；accept 也是生产交互和测试 hook 共用的入口。"""
    def __init__(self, goal=GOAL):
        self.goal, self.revision, self.reason = goal, 1, None

    def check(self, expected):
        if self.reason or expected != self.revision:
            raise Stop(self.reason or "stale_response")

    def accept(self, command, ledger):
        command = command.strip()
        if not command:
            return
        if command.startswith("goal:") and command[5:].strip():
            reason, goal = "goal_changed", command[5:].strip()
        elif command in ("revoke", "cancel"):
            reason, goal = command, self.goal
        else:
            raise Stop("invalid_control_command")
        ledger.append("control_accepted", previous_revision=self.revision,
                      goal_authority_revision=self.revision + 1, reason=reason,
                      goal_sha256=sha(goal.encode()), acceptance_boundary="after_response_before_tools")
        self.revision += 1
        self.reason, self.goal = reason, goal


class HTTPTransport:
    """固定 HTTPS 目标、禁重定向、无重试；密钥只供请求头与脱敏使用。"""
    class NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self, req, fp, code, msg, headers, newurl):
            return None

    def __init__(self, api_key):
        if not isinstance(api_key, str) or not api_key or any(not 33 <= ord(c) <= 126 for c in api_key):
            raise ValueError("密钥必须为非空、无空白的可打印 ASCII 字符串")
        self._key = api_key
        # 禁止隐式环境代理，避免从环境取凭据或改变可信目标。
        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), self.NoRedirect())

    def scrub(self, text):
        return redact(text, self._key)

    def send(self, payload):
        request = urllib.request.Request(ENDPOINT, data=payload, method="POST", headers={
            "Content-Type": "application/json", "Authorization": "Bearer " + self._key})
        try:
            with self.opener.open(request, timeout=30) as response:
                body = response.read(RESPONSE_LIMIT + 1)
                if len(body) > RESPONSE_LIMIT:
                    raise TransportFailure("response_byte_limit")
                if response.status != 200:
                    raise TransportFailure("http_error", response.status,
                                           self.scrub(body.decode("utf-8", "replace"))[:512])
                declared = response.headers.get("Content-Length")
                if declared is not None and int(declared) != len(body):
                    raise TransportFailure("response_truncated")
                return body
        except urllib.error.HTTPError as exc:
            diagnostic = self.scrub(f"HTTPError: HTTP {exc.code}: {exc.reason}")
            try:
                body = exc.read(RESPONSE_LIMIT + 1)
                text = self.scrub(body.decode("utf-8", "replace"))[:512]
            except Exception as read_error:
                text = "[错误体不可读]"
                diagnostic += self.scrub(f"；{type(read_error).__name__}: {read_error}")
            finally:
                exc.close()
            raise TransportFailure("http_error", exc.code, text, diagnostic=diagnostic) from None
        except TransportFailure:
            raise
        except Exception as exc:
            raise TransportFailure("network_error", diagnostic=self.scrub(f"{type(exc).__name__}: {exc}")) from None


class Usage:
    def __init__(self, tariff):
        self.tariff, self.known, self.fallback = tariff, 0, 0
        self.totals = dict.fromkeys(("prompt_tokens", "completion_tokens", "prompt_cache_hit_tokens", "prompt_cache_miss_tokens"), 0)

    def record(self, value):
        if value is None:
            return "missing"
        if not isinstance(value, dict):
            raise Stop("usage_malformed", "model_unknown")
        keys = tuple(self.totals)
        if any(type(value.get(k)) is not int or value[k] < 0 for k in keys[:2]):
            raise Stop("usage_malformed", "model_unknown")
        for k in keys[2:]:
            if k in value and (type(value[k]) is not int or value[k] < 0):
                raise Stop("usage_malformed", "model_unknown")
        values = {k: value[k] for k in keys[:2]}
        if not all(k in value for k in keys[2:]):
            values.update(prompt_cache_hit_tokens=0, prompt_cache_miss_tokens=value["prompt_tokens"])
            self.fallback += 1
            basis = "cache_missing_all_prompt_assumed_miss"
        else:
            values.update({k: value[k] for k in keys[2:]})
            if values[keys[2]] + values[keys[3]] != values[keys[0]]:
                raise Stop("usage_malformed", "model_unknown")
            basis = "reported"
        for k in keys:
            self.totals[k] += values[k]
        self.known += 1
        return basis

    def summary(self, attempted):
        hit, miss, output = ("0.04", "2", "8") if self.tariff == "peak" else ("0.02", "1", "4")
        total = sum(Decimal(self.totals[k]) * Decimal(p) for k, p in zip(
            ("prompt_cache_hit_tokens", "prompt_cache_miss_tokens", "completion_tokens"), (hit, miss, output))) / Decimal(1000000)
        unknown = attempted != self.known
        usage = dict(self.totals, requests_with_usage=self.known, requests_unknown_usage=attempted - self.known,
                     cache_fallback_requests=self.fallback, tokens_scope="仅有有效 usage 的请求，非未知请求的零用量")
        cost = dict(baseline_date="2026-11-01", tariff=self.tariff, currency="CNY", per_million=dict(hit=hit, miss=miss, output=output),
                    known_usage_cost=str(total), total_cost=None if unknown else str(total), unknown_cost=unknown,
                    actual_price=None, automatic_bill_estimate=False, estimate_type="dated_baseline_projection",
                    basis="仅2026-11-01价格基线投影；tariff 为显式 baseline tariff，peak 为基线最大单价。"
                          "本书代码没有实际价格查询或自动账单估算，不按请求日期、当天时段或节假日选择价格")
        return usage, cost


def parse_response(raw, seen):
    try:
        response = decode_json(raw.decode("utf-8"))
        choices = response["choices"]
        if not isinstance(choices, list) or len(choices) != 1:
            raise ValueError()
        choice = choices[0]
        if choice["finish_reason"] == "length":
            raise Stop("finish_length", "model_unknown")
        message = choice["message"]
        if message["role"] != "assistant" or (message.get("content") is not None and not isinstance(message["content"], str)):
            raise ValueError()
        reasoning = message.get("reasoning_content")
        if reasoning is not None and not isinstance(reasoning, str):
            raise ValueError()
        if reasoning:
            raise Stop("thinking_mode_mismatch", "model_unknown")
        calls = message.get("tool_calls", [])
        if calls is None:
            calls = []
        if not isinstance(calls, list) or choice["finish_reason"] != ("tool_calls" if calls else "stop"):
            raise ValueError()
        if not calls and not (isinstance(message.get("content"), str) and message["content"].strip()):
            raise Stop("final_text_missing")
        parsed, ids = [], set()
        for call in calls:
            cid, function = call["id"], call["function"]
            if not isinstance(cid, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", cid):
                raise ValueError()
            if cid in ids or cid in seen:
                raise Stop("duplicate_tool_call_id", "model_unknown")
            if call["type"] != "function" or not isinstance(function["name"], str) or not isinstance(function["arguments"], str):
                raise ValueError()
            args = decode_json(function["arguments"])
            ids.add(cid)
            parsed.append((cid, function["name"], args))
        # 不转发供应商其它字段，保留合法 assistant 与 tool_call_id 的协议配对。
        accepted = {"role": "assistant", "content": message.get("content")}
        if calls:
            accepted["tool_calls"] = [{"id": c[0], "type": "function", "function": {
                "name": c[1], "arguments": json.dumps(c[2], ensure_ascii=False, allow_nan=False)}} for c in parsed]
        encode(accepted)
        return response, accepted, parsed
    except Stop:
        raise
    except (ValueError, KeyError, TypeError, UnicodeError, RecursionError):
        raise Stop("response_malformed", "model_unknown") from None


@dataclass(frozen=True)
class Card:
    path: Path
    run_id: str
    action_id: str
    revision: int
    version: str
    content_hash: str
    content_bytes: int


class Harness:
    def __init__(self, transport, output_root, *, coordinator=None, after_response=None,
                 interactive=False, tariff="peak", max_models=6, max_tools=12):
        if tariff not in ("peak", "offpeak") or not 1 <= max_models <= 6 or not 0 <= max_tools <= 12:
            raise ValueError("预算或计价参数不合法")
        self.transport, self.root = transport, Path(output_root)
        self.mode = "live" if isinstance(transport, HTTPTransport) else "scripted"
        self.coordinator = coordinator or Coordinator()
        self.after_response, self.interactive = after_response, interactive
        self.max_models, self.max_tools = max_models, max_tools
        self.sources = {sid: bytes(FIXTURES[sid]) for sid in SOURCE_IDS}
        self.texts = {sid: data.decode("utf-8") for sid, data in self.sources.items()}
        if any(len(data) > 4096 for data in self.sources.values()):
            raise ValueError("材料超出4KB，不截断")
        self.hashes = {sid: sha(data) for sid, data in self.sources.items()}
        self.schemas, self.mapping = decode_json(encode(TOOLS)), dict(MAPPING)
        self.run_id = uuid.uuid4().hex
        self.directory = self.root / self.run_id
        self.ledger = None
        self.models = self.tools = 0
        self.trace = []
        self.read_sources, self.seen = set(), set()
        self.disclosed_sources = frozenset()
        self.current_request_id = self.card_request_id = None
        self.card_disclosed_sources = ()
        self.card, self.card_bytes, self.acceptance = None, None, None
        self.effect_in_flight = self.request_pending = False
        self.response_state = self.protocol_status = "not_requested"
        self.meter = Usage(tariff)
        self.history = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": self.coordinator.goal}]

    def scrub(self, text):
        return getattr(self.transport, "scrub", redact)(text)

    def snapshot(self, name, data):
        sanitized = self.scrub(data.decode("utf-8", "surrogateescape")).encode("utf-8", "surrogateescape")
        create_new(self.directory / name, sanitized)
        return dict(path=name, sha256=sha(sanitized), bytes=len(sanitized), redacted=sanitized != data)

    def accept_control(self, command):
        self.coordinator.accept(command, self.ledger)

    def control_boundary(self, revision):
        if self.after_response:
            self.after_response(self)
        if self.interactive:
            print("边界已到：Enter / goal:<新要求> / revoke / cancel（不撤销已提交动作）", file=sys.stderr, flush=True)
            try:
                self.accept_control(input())
            except EOFError:
                raise Stop("control_input_closed") from None
        self.coordinator.check(revision)

    def verify_artifact(self):
        c = self.card
        if c is None:
            raise Stop("artifact_missing", "artifact_unknown")
        try:
            with c.path.open("rb") as file:
                actual = file.read(len(self.card_bytes) + 1)
        except OSError as exc:
            raise Stop("artifact_unreadable", "artifact_unknown",
                       diagnostic=self.scrub(f"{type(exc).__name__}: {exc}")) from None
        expected_version = f"{self.run_id}/{c.action_id}/r{c.revision}"
        checks = dict(content_check=actual == self.card_bytes, hash_check=sha(actual) == c.content_hash,
                      version_check=c.version == expected_version and f"- version: {expected_version}\n".encode() in actual,
                      binding_check=c.run_id == self.run_id and c.path == self.directory / "card.md" and c.content_bytes == len(actual),
                      sources_check=self.read_sources == set(SOURCE_IDS),
                      disclosure_check=self.card_disclosed_sources == SOURCE_IDS and self.card_request_id is not None,
                      current_authority=c.revision == self.coordinator.revision and self.coordinator.reason is None)
        if not all(checks.values()):
            raise Stop("artifact_verification_failed", "artifact_unknown")
        return dict(checks, card_content_hash=sha(actual), committed_version=c.version,
                    source_order=list(SOURCE_IDS), source_sha256=self.hashes,
                    creation_request_id=self.card_request_id, disclosed_sources=list(self.card_disclosed_sources),
                    proves="每源一段原文、来源、顺序、版本及材料已进入创建请求；不证明模型理解、摘要正确或观点真实")

    def tool(self, name, args, revision):
        self.coordinator.check(revision)
        reject = lambda code: {"ok": False, "reason": code}
        if name not in self.mapping:
            return reject("unknown_tool")
        if not isinstance(args, dict):
            return reject("schema_error")
        if name == "read_source":
            if set(args) != {"source_id"} or not isinstance(args["source_id"], str):
                return reject("schema_error")
            sid = args["source_id"]
            if sid not in self.sources:
                return reject("wrong_source")
            self.read_sources.add(sid)
            return dict(ok=True, source_id=sid, content=self.texts[sid], source_sha256=self.hashes[sid], fixture=True)
        if self.card is not None:
            return reject("already_created")
        if set(args) != {"entries"} or not isinstance(args["entries"], list) or len(args["entries"]) != 2:
            return reject("schema_error")
        for i, entry in enumerate(args["entries"]):
            if not isinstance(entry, dict) or set(entry) != {"source_id", "quote"} or not isinstance(entry["quote"], str):
                return reject("schema_error")
            if entry["source_id"] != SOURCE_IDS[i]:
                return reject("wrong_source_or_order")
            quote = entry["quote"]
            if not 8 <= len(quote) <= 200:
                return reject("quote_length")
            if quote not in self.texts[SOURCE_IDS[i]]:
                return reject("verbatim_missing")
        if self.read_sources != set(SOURCE_IDS):
            return reject("sources_not_read")
        if not set(SOURCE_IDS).issubset(self.disclosed_sources):
            return reject("sources_not_disclosed")
        self.card_disclosed_sources = tuple(sid for sid in SOURCE_IDS if sid in self.disclosed_sources)
        self.card_request_id = self.current_request_id
        action = uuid.uuid4().hex
        version = f"{self.run_id}/{action}/r{revision}"
        lines = ["# 原文摘录", "", "随书自编公开 fixture；非个人亲历。仅收录原文，不生成语义结论。", "",
                 f"- run_id: {self.run_id}", f"- action_id: {action}", f"- version: {version}",
                 f"- goal_authority_revision: {revision}", f"- creation_request_id: {self.card_request_id}", ""]
        for i, entry in enumerate(args["entries"], 1):
            sid = entry["source_id"]
            lines += [f"## {i}. {sid}", f"- source_sha256: {self.hashes[sid]}", ""]
            lines += ["> " + line for line in entry["quote"].split("\n")] + [""]
        self.card_bytes = ("\n".join(lines) + "\n").encode()
        self.card = Card(self.directory / "card.md", self.run_id, action, revision, version, sha(self.card_bytes), len(self.card_bytes))
        self.ledger.append("write_intent", action_id=action, target="card.md", intended_content_hash=self.card.content_hash,
                           intended_content_bytes=self.card.content_bytes, version=version, goal_authority_revision=revision,
                           request_id=self.card_request_id, disclosed_sources=list(self.card_disclosed_sources))
        self.effect_in_flight = True
        create_new(self.card.path, self.card_bytes)
        self.ledger.append("write_result", action_id=action, target="card.md", content_hash=self.card.content_hash,
                           content_bytes=self.card.content_bytes, version=version, outcome="created")
        acceptance = self.verify_artifact()
        self.ledger.append("artifact_verified", action_id=action, **acceptance)
        self.acceptance = acceptance
        self.effect_in_flight = False
        return dict(ok=True, artifact="card.md", content_hash=self.card.content_hash, version=version, verified=True)

    def loop(self):
        while self.models < self.max_models:
            revision = self.coordinator.revision
            self.coordinator.check(revision)
            payload = encode(dict(model="deepseek-flash", thinking={"type": "disabled"}, stream=False,
                                  max_tokens=1024, tools=self.schemas, messages=self.history))
            if len(payload) > PAYLOAD_LIMIT:
                raise Stop("payload_byte_budget")
            disclosed = set()
            # 从实际提交的 payload 推导发送，不信执行时的 read_sources 可变集合。
            for message in decode_json(payload)["messages"]:
                if message["role"] == "tool":
                    result = decode_json(message["content"])
                    sid = result.get("source_id")
                    if (result.get("ok") is True and isinstance(sid, str) and sid in self.sources
                            and result.get("content") == self.texts[sid] and result.get("source_sha256") == self.hashes[sid]):
                        disclosed.add(sid)
            request_id = uuid.uuid4().hex
            snapshot = self.snapshot(f"request-{request_id}.json", payload)
            self.ledger.append("model_request_intent", request_id=request_id, request_time=datetime.now(timezone.utc).isoformat(),
                               goal_authority_revision=revision,
                               tools_mapping=self.mapping, tools_schema=self.schemas, tools_schema_hash=sha(encode(self.schemas)),
                               payload_hash=sha(payload), payload_bytes=len(payload), snapshot=snapshot,
                               disclosed_sources=[sid for sid in SOURCE_IDS if sid in disclosed])
            self.disclosed_sources = frozenset(disclosed)
            self.current_request_id = request_id
            self.models += 1
            self.request_pending = True
            self.response_state, self.protocol_status = "awaiting_response", "not_validated"
            print(f"run {self.run_id}：请求 {self.models}/{self.max_models}", file=sys.stderr, flush=True)
            try:
                raw = self.transport.send(payload)
            except TransportFailure as exc:
                self.response_state = "transport_failed"
                self.ledger.append("model_transport_failure", request_id=request_id, reason=exc.reason,
                                   http_status=exc.http_status, error_body=self.scrub(exc.body or "")[:512],
                                   diagnostic=self.scrub(exc.diagnostic) if exc.diagnostic else None)
                raise
            self.response_state = "received_not_persisted"
            if len(raw) > RESPONSE_LIMIT:
                raise Stop("response_byte_limit", "model_unknown")
            saved = self.snapshot(f"response-{request_id}.json", raw)
            self.ledger.append("model_response", request_id=request_id, snapshot=saved)
            self.request_pending = False
            self.response_state = "persisted"
            # 先持久化，再验证；脱敏后的消息才进入后续请求和 assistant 账本。
            try:
                clean = self.scrub(raw.decode("utf-8")).encode()
                envelope = decode_json(clean)
            except (ValueError, UnicodeError, RecursionError):
                self.control_boundary(revision)
                raise Stop("response_malformed", "model_unknown") from None
            metadata = dict(request_id=request_id, model=None, system_fingerprint=None, created=None,
                            origin="provider_response" if self.mode == "live" else "scripted_fixture")
            if isinstance(envelope, dict):
                # 只取小型、可打印 ASCII 元信息；缺失/不可用为 null，不猜配置值。
                for field in ("model", "system_fingerprint"):
                    value = envelope.get(field)
                    if isinstance(value, str) and len(value) <= 256 and value.isascii() and value.isprintable():
                        metadata[field] = self.scrub(value)
                if type(envelope.get("created")) is int and envelope["created"] >= 0:
                    metadata["created"] = envelope["created"]
            self.ledger.append("model_response_metadata", **metadata)
            self.trace.append(metadata)
            reported = envelope.get("usage") if isinstance(envelope, dict) else None
            try:
                basis = self.meter.record(reported)
            except Stop:
                self.control_boundary(revision)
                raise
            self.ledger.append("model_usage", request_id=request_id, basis=basis,
                               usage=None if reported is None else {k: reported[k] for k in self.meter.totals if k in reported})
            self.control_boundary(revision)
            _, assistant, calls = parse_response(clean, self.seen)
            self.protocol_status = "validated"
            if len(calls) > self.max_tools - self.tools:
                raise Stop("tool_budget")
            self.ledger.append("assistant_accepted", request_id=request_id, message=assistant)
            self.protocol_status = "accepted"
            self.history.append(assistant)
            self.seen.update(c[0] for c in calls)
            if not calls:
                if self.card is None:
                    raise Stop("finished_without_artifact", "failed")
                self.acceptance = self.verify_artifact()
                self.ledger.append("final_artifact_verified", **self.acceptance)
                return "success", "assistant_finished_after_verification"
            batch_stopped = False
            for cid, name, args in calls:
                self.tools += 1
                result = {"ok": False, "reason": "batch_stopped"} if batch_stopped else self.tool(name, args, revision)
                self.ledger.append("tool_result", tool_call_id=cid, name=name, result=result)
                self.history.append(dict(role="tool", tool_call_id=cid, content=encode(result).decode()))
                batch_stopped |= result.get("reason") == "unknown_tool"
            if batch_stopped:
                raise Stop("unknown_tool")
        raise Stop("model_budget")

    def run(self):
        status, reason, diagnostic = "stopped", "not_started", None
        try:
            if not self.root.parent.is_dir():
                raise Stop("output_parent_missing", "failed", diagnostic="output_root 的父目录必须已存在；未创建任何祖先或请求模型")
            if not self.root.is_dir():
                self.root.mkdir(exist_ok=True)
            sync_directory(self.root.parent)
            self.directory.mkdir()  # 独占随机 run；从不查找或接续旧 run。
            sync_directory(self.root)
            self.ledger = Ledger(self.directory, self.run_id)
            self.ledger.append("run_start", mode=self.mode, goal_authority_revision=self.coordinator.revision,
                               goal_sha256=sha(self.coordinator.goal.encode()), source_sha256=self.hashes,
                               limits=dict(model_requests=self.max_models, tool_requests=self.max_tools, payload_bytes=PAYLOAD_LIMIT),
                               assumptions="可信进程和库、可信且无人并发篡改的输出目录；无恢复，无重试，非任意代码隔离")
            status, reason = self.loop()
        except KeyboardInterrupt:
            status = "artifact_unknown" if self.effect_in_flight else "model_unknown" if self.request_pending else "cancelled"
            reason = "interrupt_in_flight" if self.effect_in_flight or self.request_pending else "keyboard_interrupt"
        except Stop as exc:
            status, reason = exc.status, exc.reason
            if reason in ("response_byte_limit", "response_truncated", "response_malformed", "finish_length",
                          "thinking_mode_mismatch", "duplicate_tool_call_id", "usage_malformed", "final_text_missing"):
                self.protocol_status = "rejected"
            if exc.diagnostic:
                diagnostic = self.scrub(exc.diagnostic)
                print(diagnostic, file=sys.stderr, flush=True)
            if isinstance(exc, LedgerUnknown):
                print(self.scrub(traceback.format_exc()), file=sys.stderr, flush=True)
        except Exception:
            status = "artifact_unknown" if self.effect_in_flight else "model_unknown" if self.request_pending else "stopped"
            reason = "unexpected_error"
            diagnostic = self.scrub(traceback.format_exc())
            print(diagnostic, file=sys.stderr, flush=True)
        if reason in ("model_budget", "tool_budget", "payload_byte_budget") and self.acceptance is not None:
            try:
                self.acceptance = self.verify_artifact()
                status = "artifact_created_not_finished"
            except Stop as exc:
                status, reason = exc.status, exc.reason
                diagnostic = self.scrub(exc.diagnostic) if exc.diagnostic else None
                if diagnostic:
                    print(diagnostic, file=sys.stderr, flush=True)
        try:
            if self.ledger is None and reason != "output_parent_missing":
                raise LedgerUnknown()
            if self.ledger is not None:
                self.ledger.append("run_stop", status=status, reason=reason, diagnostic=diagnostic, request_pending=self.request_pending,
                                   response_state=self.response_state, protocol_status=self.protocol_status,
                                   effect_in_flight=self.effect_in_flight, goal_authority_revision=self.coordinator.revision)
        except (LedgerUnknown, OSError):
            status, reason = "ledger_unknown", "ledger_write_failed"
            print(self.scrub(traceback.format_exc()), file=sys.stderr, flush=True)
        finally:
            if self.ledger is not None:
                try:
                    self.ledger.close()
                except (OSError, KeyboardInterrupt):
                    status, reason = "ledger_unknown", "ledger_close_failed"
                    print(self.scrub(traceback.format_exc()), file=sys.stderr, flush=True)
        usage, cost = self.meter.summary(self.models)
        usage["origin"] = "provider_reported_or_unknown" if self.mode == "live" else "scripted_fixture_not_measured"
        artifact = None if self.card is None else dict(path=str(self.card.path), content_hash=self.card.content_hash,
            version=self.card.version, run_id=self.card.run_id, action_id=self.card.action_id,
            goal_authority_revision=self.card.revision, bytes=self.card.content_bytes, acceptance=self.acceptance,
            acceptance_scope="最后一次已持久化的验收；停止状态不继承成功声明",
            verified_at_stop=status in ("success", "artifact_created_not_finished"),
            outcome="durably_verified" if self.acceptance else "intended_or_unknown",
            currently_authorized=self.card.revision == self.coordinator.revision and self.coordinator.reason is None)
        return dict(run_id=self.run_id, status=status, stop_reason=reason, mode=self.mode, artifact=artifact, trace=self.trace,
                    usage=usage, baseline_cost=cost, goal_authority_revision=self.coordinator.revision,
                    model_requests=self.models, tool_requests=self.tools,
                    model_response=dict(scope="latest_request", state=self.response_state, protocol_status=self.protocol_status,
                        record_persisted=self.response_state == "persisted", accepted=self.protocol_status == "accepted"),
                    known_unknown=dict(unknown_cost=cost["unknown_cost"], model_response_unknown=self.request_pending,
                        model_protocol_unknown=self.models > 0 and self.protocol_status in ("not_validated", "rejected"),
                        effect_unknown=self.effect_in_flight or status == "artifact_unknown", ledger_unknown=status == "ledger_unknown",
                        stop_durably_recorded=self.ledger is not None and status != "ledger_unknown",
                        success_durably_recorded=status == "success", no_resume=True,
                        live_ability_tested=self.mode == "live" and status == "success"))


class ScriptedTransport:
    """仅验证循环及文件，不是 live 能力证据；脚本中的 quote 由测试编写者指定。"""
    def __init__(self, responses):
        self.responses, self.payloads = iter(responses), []

    def send(self, payload):
        self.payloads.append(payload)
        response = next(self.responses)
        if isinstance(response, BaseException):
            raise response
        return response if isinstance(response, bytes) else encode(response)


def scripted_response(calls=(), usage=None, content=None):
    message = dict(role="assistant", content=content)
    if calls:
        message["tool_calls"] = [dict(id=cid, type="function", function=dict(name=name, arguments=json.dumps(args, ensure_ascii=False)))
                                 for cid, name, args in calls]
    return dict(choices=[dict(finish_reason="tool_calls" if calls else "stop", message=message)], usage=usage)


def demo_responses():
    entries = [dict(source_id=sid, quote=FIXTURES[sid].decode().splitlines()[1]) for sid in SOURCE_IDS]
    return [scripted_response([(f"read-{i}", "read_source", dict(source_id=sid)) for i, sid in enumerate(SOURCE_IDS)]),
            scripted_response([("create-1", "create_card", dict(entries=entries))]), scripted_response(content="摘录文件已创建。")]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    selection = parser.add_mutually_exclusive_group(required=True)
    selection.add_argument("--demo", action="store_true", help="离线脚本演示，不验证真实模型选摘录能力")
    selection.add_argument("--live", action="store_true", help="显式启用付费固定 DeepSeek API，默认从 DEEPSEEK_API_KEY 取密钥")
    parser.add_argument("--key-stdin", action="store_true", help="可选覆盖环境配置：仅 live 从 stdin 读一行密钥；与 interactive 同用需真实终端")
    parser.add_argument("--interactive", action="store_true", help="响应后的粗粒度控制边界，输入不在请求进行中处理；推荐环境密钥方式")
    parser.add_argument("--tariff", choices=("peak", "offpeak"), default="peak",
                        help="2026-11-01 baseline tariff，默认 peak；不是当天实际价格或自动账单估算")
    parser.add_argument("--output-dir", type=Path, default=Path(__file__).parent / "runs", help="可信且无人并发篡改的输出目录；其父目录必须已存在，不递归创建祖先")
    args = parser.parse_args()
    if args.key_stdin and not args.live:
        parser.error("--key-stdin 仅供 --live 使用")
    if args.key_stdin and args.interactive and not sys.stdin.isatty():
        parser.error("不支持管道 --key-stdin 与 --interactive 同用；请改用环境密钥或真实终端")
    if args.live:
        key = sys.stdin.readline().rstrip("\r\n") if args.key_stdin else os.environ.get("DEEPSEEK_API_KEY")
        try:
            transport = HTTPTransport(key)
        except ValueError:
            parser.error("密钥缺失或格式无效：DEEPSEEK_API_KEY 或 --key-stdin 必须提供无空白的可打印 ASCII 密钥")
    else:
        transport = ScriptedTransport(demo_responses())
    summary = Harness(transport, args.output_dir, interactive=args.interactive, tariff=args.tariff).run()
    print(encode(summary).decode(), flush=True)
    return 0 if summary["status"] == "success" else 1


if __name__ == "__main__":
    sys.exit(main())

完整源码到此结束。第4章继续分析同一循环：当前状态如何形成，已发生事实如何保留，以及程序依据什么允许下一步或决定停止。摘录文件、请求快照和事件记录将为这些关系提供具体对象。