Mini Harness：模块装配与可恢复执行
用本地工单任务装配模型适配器、MCP 工具、SQLite 状态和独立验收器。各模块按同一个任务契约接线；固定响应实验验证控制与恢复，不代表已经验证真实模型的任务能力。
最近修订：2026.09.07
阅读导航 · 本篇目录
用本地工单任务装配模型适配器、MCP 工具、SQLite 状态和独立验收器。各模块按同一个任务契约接线；固定响应实验验证控制与恢复，不代表已经验证真实模型的任务能力。
要搭一个能完成任务的 Harness，第一步是画清模块，以及它们交换什么。实战采用一个具体起点：单用户、单 Worker、Python、本地执行、允许重启恢复。在这个范围内，优先选择少量显式函数、严格数据校验和 SQLite；出现明确瓶颈后再换组件。
八个模块怎样拆
| 模块 | 输入 → 输出 | 必须负责的判断 | 实验中的落点 |
|---|---|---|---|
| 任务契约 | 用户目标 → 可验收目标、允许动作 | 什么才算完成 | lab.py: execute() 的 expected 和运行配置 |
| 上下文组装 | 契约、任务材料、已知状态 → 模型输入 | 哪些信息本轮需要，来源是否可信 | model.py: propose()，目前只输入固定任务 |
| 模型适配 | 模型输入 → 候选动作、调用元信息 | 响应是否完整，工具名与字段是否有效 | propose()、parse_response()、validate_action() |
| 运行控制 | 候选动作、状态 → 下一步或终态 | 继续、等待、失败、恢复、预算耗尽 | lab.py: execute() |
| 工具执行 | 工具名、参数 → 结构化结果 | 调用什么，如何报告失败 | session_for()、call() 和 server.py |
| 执行策略 | 动作、操作者配置 → 允许或拒绝 | 动作是否在授权范围内 | 目标一致性检查、execute_write；尚无系统沙箱 |
| 状态存储 | 契约、意图、事件 → 可恢复记录 | 重启后哪些事实仍成立 | store.py: Runs，以及服务端 Tickets |
| 结果验收 | 可信目标、实际资源 → 通过或失败 | 实际交付是否符合要求 | 工具查询后对照 expected |
这些模块的依赖方向应清楚：模型返回提案；运行器决定是否派发；工具改变资源；验收器读取资源。运行器把关键状态和事件写入存储。模型输出不能自行变成执行许可或验收标准。
Mini Harness 的运行器接收模型提案，经策略检查后通过 MCP 调用工具服务；Runs 保存意图与进度，Tickets 保存工单与操作键，查询结果交给验收器。
当前适合这个起点的默认组合
“最佳”必须有适用条件。下面是面向当前教学场景的工程建议，不是经过统一基准测试的工具排行榜。
| 模块 | 新项目的默认选择 | 为什么先这样选 | 何时替换或扩展 |
|---|---|---|---|
| 任务与数据契约 | Python 数据结构 + Pydantic 或严格校验函数 | 字段、长度、枚举和额外字段策略容易核对 | 跨语言共享时统一维护 JSON Schema |
| 上下文 | 显式消息列表 + 当前任务材料 + 最近工具结果 | 可以检查每条信息为什么进入模型 | 材料超出预算且检索确实提升验收率时增加检索 |
| 模型 | 一家供应方的官方 SDK，封装成小适配器 | 先稳定输出协议、错误分类和用量记录 | 在同一任务集上证明第二模型有收益后加路由 |
| 循环 | 显式状态机或有步数限制的循环 | 中断点和终止条件容易定位 | 分支、人工暂停和恢复节点增多时考虑 LangGraph |
| 工具 | 同进程先直接调用函数；跨客户端共享时用 MCP SDK | 本地函数调试简单，MCP 解决跨进程契约 | 多用户共享服务再引入 HTTP、身份与服务治理 |
| 策略与隔离 | 固定工具白名单、参数约束、执行前检查 | 可以限制当前业务动作 | 一旦开放 Shell 或不可信代码，就加入操作系统隔离 |
| 存储 | SQLite，运行状态与事件同事务写入 | 单 Worker 容易部署、检查和重放 | 多 Worker 时增加所有权与并发控制，按需迁移数据库 |
| 验收与观测 | 确定性断言 + 独立资源查询 + 有序事件 | 能指出哪项要求失败 | 语义任务增加人工标注样本；跨服务再加 Trace |
官方 SDK 是新项目建议；配套实验为展示原始请求，在线模型适配器使用 Python 标准库 HTTP。实验的严格校验也由普通函数实现，不要求先引入抽象框架。
LangGraph 的 checkpointer 保存线程运行状态，store 保存跨线程信息；它可以承担更多状态编排，但不会自动替业务写入提供幂等保证。是否采用它，取决于运行流程复杂度。LangGraph 持久化文档
用什么任务把它们组装起来
实战的第一块可运行切片是：给定 source 与 title，创建一张本地待核验工单，重启后同一业务操作仍只对应一张工单。
这个任务的执行路径已经确定，因此它是单动作工作流：模型负责产生结构化提案，运行器执行固定的查询、创建和验收步骤。实际业务中，如果字段已经完全确定，可以直接调用创建函数。这里保留模型适配，是为了单独练习生成结果如何进入受控执行链路。
在此基础上，真正需要 Agent 决策的扩展任务可以是“读取三份材料，比较冲突，生成待核验问题，再创建工单”。这时模型要根据工具结果选择下一步，才需要多轮循环；模型与循环给出循环接口与消息往返，本 Mini 实验包尚未实现该扩展。架构对照实验另外实现了读取两份资料、根据观察推进、批准后创建的控制链路，模型仍用固定响应，便于隔离编排差异。
从模型到验收的装配顺序
| 顺序 | 动手内容 | 这一层完成的标准 |
|---|---|---|
| 1. 模块与选型 | 确定八项职责、默认组件和任务边界 | 每项职责能对应到代码或明确的待实现项 |
| 2. 模型与循环 | 定义上下文、动作 Schema 和循环退出条件 | 普通文本、截断、多动作不能误入执行 |
| 3. 工具与策略 | 接入查询、创建及 MCP 传输 | 工具响应可校验，写入许可由应用决定 |
| 4. 状态与验收 | 保存意图，加入幂等创建与恢复查询 | 三个退出点都能得到可解释的资源结果 |
| 5. 组装与运行 | 跑命令、读数据库、替换一个模块 | 正常路径和故障路径都有实际证据 |
Skills、长期记忆和多 Agent 都是可选扩展。只有重复流程值得沉淀、跨任务信息确实有用、任务拆分收益超过协调成本时，再分别接入。关于循环和任务边界，可对照Harness 工程主线继续阅读。
需要比较自写循环、图式编排与持久工作流时，继续阅读架构选型与对照实验。
从任务契约构造上下文
工单实验的可信目标只有两个字段：
{"source":"source-001","title":"核验 Agent Harness 资料"}运行 ID、数据库位置、执行开关和派发预算由应用另外保存。它们不属于模型可以改写的参数。上下文组装器在这个阶段只需要任务说明、目标字段和可用工具定义。
扩展为资料研究任务时，按四层组织输入：稳定的任务规则；当前用户目标；带来源标识的材料；本轮状态和最近工具结果。检索材料中的指令应按资料内容处理，不能用它改写执行策略。长输出保留摘要与可重新读取的引用，原始证据另存；不要把每轮日志全部追加进去。
默认先用显式消息列表。只有在真实任务中发现“相关材料找不到”或上下文预算不足，才增加检索、压缩或分层记忆。这样可以判断收益到底来自模型、材料还是运行逻辑。
模型适配器输出什么
实验中的实际接口是：
# model.py 中已有的接口；provider 默认为 fixture
propose(expected, provider, model) -> (action, metadata)action 必须恰好包含非空的 source、title，每项长度不超过 300 字符。metadata 保存模型来源；在线模式还记录响应 ID、模型和用量，便于排查同一动作是怎样产生的。
parse_response() 检查停止原因是 tool_use，且恰好出现一个名为 propose_review_ticket 的工具调用；随后由 validate_action() 检查字段。运行器最后把动作与可信目标比较。Schema 正确只能说明结构合规，不能证明目标正确。
| 模型返回 | 处理方式 | 理由 |
|---|---|---|
| 一份完整、字段正确的提案 | 保存后进入运行器 | 提案获得了明确结构 |
| 普通文字“已创建” | 拒绝 | 没有可执行提案，也没有资源证据 |
| 被 token 上限截断的响应 | 拒绝 | 不能按部分输出猜测动作 |
| 两个工具调用 | 拒绝 | 超出本实验单动作契约 |
增加 approved 或 operation_key | 拒绝 | 模型不能扩充自身权限或改变去重标识 |
| 字段合法，但标题与目标不同 | 运行器拒绝 | 结构正确仍可能做错任务 |
新项目可用供应方官方 SDK 处理连接与响应类型，但应保留这层应用契约。工具名称、用途说明和输入 Schema 是模型选工具的依据，名称对上并不足以约束参数语义。Claude 工具定义文档
现有实验是一轮提案解析。下文的多轮循环属于扩展设计，不应据此认为代码已自动研究资料。
多轮 Agent Loop 应怎样接
如果任务需要“读材料以后再决定查哪里”，单次 propose() 就不够。推荐把适配器扩展为接收消息历史和本轮可用工具，返回统一的三类结果：候选工具调用、最终答案、无法继续。循环控制不依赖供应方具体的停止原因名称。
下面是设计伪代码，展示模块连接；它不是实验包中可直接调用的 API：
读取已保存的任务、状态和消息
重复，最多 N 轮：
先处理历史未确认写入：只读对账或等待，不生成替代写入
检查任务截止时间、剩余额度和取消状态
组装本轮上下文，调用模型，校验完整响应
若模型提出最终答案：执行独立验收，再决定成功或继续
若模型提出工具调用：
校验工具与参数，检查当前执行权限
为业务动作确定稳定操作键，保存意图与预算消耗
仅在可以安全执行时派发工具
保存工具结果，按供应方协议关联原 tool call ID
将模型的调用消息与对应工具结果加入下一轮上下文
若无法继续：保存原因并退出
轮数耗尽：保存 budget_exhausted，交给用户或后续流程最容易遗漏的是工具结果回传。只调用工具再打印结果，不构成“观察结果后再决策”的循环。另一处是身份：供应方的 tool call ID 用于消息关联，业务操作键用于副作用去重，两者不能混用。一次写入在重试时应复用同一业务键；不同业务动作必须分配不同键。
预算怎样放进组合
至少区分模型请求次数、工具派发次数和任务总截止时间。模型超时、工具超时、任务超时也要分别处理。写入超时可能意味着结果未知，应进入对账流程，不能一律重试。
配套实验已经持久化工具派发预算，上限为 2；模型适配器是一次非流式请求，尚无持久化模型预算、多轮消息或总截止时间。扩展循环时先补这些控制，再增加自动重试；否则一次重启就可能重置额度。
先换模型，保持其他模块不变
默认 fixture 返回固定响应，用来稳定测试运行器；anthropic 是可选的在线提案适配器。两者输出同一份动作契约，因此工具、存储和验收可以复用。
选模型时，用同一批任务比较合法提案率、目标符合率、完成成本和延迟。格式经常出错时先检查 Schema 与工具说明；提案正确但资源错误时检查执行和验收。不要把所有失败都归因于模型能力。更多网关设计见模型网关。
先定义两个业务工具
本实验只暴露两个工具，把读取和写入分开：
| 工具 | 参数 | 输出 | 副作用 |
|---|---|---|---|
lookup_review_ticket | operation_key | {"ticket": null} 或工单对象 | 读取 |
create_review_ticket | operation_key, source, title | {"ticket": {"id": 整数, "action": {...}}} | 创建，或返回同键的既有工单 |
同键同参数返回同一资源；同键不同参数拒绝执行。这条规则由服务端 Tickets.create() 和唯一约束落实，不能只写在工具说明里。
工具层不要顺便判断任务完成。创建接口可能成功创建了错误标题；是否符合任务仍由验收器对照原始目标判断。工具错误也要显式返回或抛出，不能把失败信息包装成看似成功的业务对象。
把本地函数包成 MCP 服务
server.py 使用官方 Python SDK 中的 FastMCP，将 Tickets.lookup() 与 Tickets.create() 包装成两个工具。返回值通过 TicketEnvelope 提供结构化 ticket 字段。
客户端的连接顺序是：启动传输 → 建立 ClientSession → initialize() → 读取工具目录 → 检查需要的工具 → 发起调用。实验的 session_for() 管连接生命周期，call() 管工具结果解析，execute() 管业务流程。这三层分开后，换传输不必改任务验收。
# 在实验目录中使用的真实客户端接口片段
async with session_for(root) as (session, version):
ticket = await call(
session,
'lookup_review_ticket',
{'operation_key': 'review-001:ticket'},
)root 是状态目录的 Path；完整运行入口见组装与运行。调用返回后，call() 先检查 isError，再检查 structuredContent 中是否包含 ticket。MCP 的工具结果可以包含文本与结构化内容；本应用主动选择后者作为机器判断接口。MCP 工具规范
连接、工具调用和业务验收分别有明确责任。查到既有工单时可直接验收，无需再次创建。
stdio 与 HTTP 如何选择
| 条件 | 默认组合 | 应验证的边界 |
|---|---|---|
| Harness 自己启动本地工具服务 | SDK stdio 客户端 + 子进程 FastMCP | 启停、退出码、标准输出不能混入普通日志 |
| 多个客户端访问共享工具服务 | Streamable HTTP + 服务身份与认证 | 会话、超时、代理、访问控制与资源归属 |
| 一次性同进程业务函数 | 直接调用函数 | 参数、错误、副作用与结果契约 |
实验固定 mcp==1.29.1 与 MCP 2025-11-25 协商基线，用于复现已记录结果，不把这个固定版本称为所有新项目的最佳版本。对应 SDK
stdio 测试会真正启动服务进程；HTTP 测试使用本机回环地址。两者复用工具逻辑，但 HTTP 当前只覆盖正常路径。生产 HTTP 还需要认证、会话失效、撤权和代理环境测试。本机监听不提供身份认证。
执行策略应插在哪里
顺序是：校验模型动作 → 检查业务目标 → 检查执行许可 → 保存派发意图 → 调用工具。策略检查应紧邻实际派发；恢复后也要重新判断当前是否允许新增写入。
实验里，--execute 控制是否允许创建本地工单，查询不需要该开关。操作键由运行器根据 run ID 生成，数据库路径由命令行配置。模型只提供工单内容，没有通用 Shell，也不能指定任意文件路径。
这是一份小型业务白名单。将来增加文件或 Shell 工具时，需要同时加入路径限制、环境变量过滤、网络范围和操作系统隔离；仅在提示词里写“不要越权”不能限制进程权限。Hook 可以在配置匹配的调用路径执行检查，但不能替代服务端授权或沙箱。执行安全篇
怎样替换为真实业务系统
保留运行器面对的查询、创建和结果接口，在服务端替换 Tickets 的存储实现。接 CRM、任务系统或工单 SaaS 前，先确认四件事：是否支持业务幂等键；是否能按该键查询；写入后多久查询可见；冲突与限流怎样报告。
两份存储各管什么
| 存储 | 内容 | 它能回答的问题 |
|---|---|---|
runs.sqlite | 任务契约、状态、已保存提案、派发次数、有序事件 | Harness 原本打算做什么，走到了哪里 |
tickets.sqlite | 操作键、资源 ID、工单字段 | 业务资源到底是否存在，内容是什么 |
实验中客户端维护 Runs，工具服务维护 Tickets。分开存储是为了真实呈现“两次提交不能靠同一个本地事务包住”的问题。运行状态和事件在运行库内同事务提交；它们与工单创建不构成跨库原子事务。
状态机只保留有行动意义的状态
| 状态 | 下一次应该做什么 |
|---|---|
new | 生成并校验提案 |
planned | 使用已保存提案，进入资源查询 |
reconciling | 根据业务键查工单，判断是否还需创建 |
waiting_execution | 等待允许新增写入；资源已存在时可以只验收 |
dispatching | 派发意图已落盘；重启后先查询，不能假定未写入 |
reconcile_required | 派发预算用尽；继续查询或转人工处理 |
succeeded / failed | 返回已保存的终态 |
interrupted 是命令行捕获异常后的输出摘要，不是数据库里的新状态。异常发生时，数据库可能仍停在 dispatching。恢复应以持久化记录为依据。
本实验成功终态重读会直接返回，不持续监测工单以后是否被删除。若业务要求“始终存在”，应另设巡检任务，不能把一次验收扩大成永久保证。
一次写入怎样组装
实际执行顺序如下：
Runs.open()保存任务契约；同一个 run ID 更换目标、供应方或模型时拒绝恢复。- 提案校验成功后保存；恢复时复用提案。
- 根据
run + ':ticket'得到稳定操作键，查询工单。 - 如果工单不存在，检查执行许可和剩余派发预算。
- 用
Runs.record()在同一事务内保存意图、事件并消耗一次预算。 - 调用创建工具，再用查询工具读取实际工单。
- 验收 ID 类型及字段与原始目标一致后，保存成功终态。
第 5 步记录的是派发尝试预算，因此即使进程在网络调用前退出，也已经消耗一次。这个保守选择可以避免重启无限重置额度；它不是服务端实际执行次数的精确统计。
服务端创建则在 SQLite 事务中检查操作键，并以唯一约束防止重复资源。同键出现不同参数会报冲突。事务与连接管理的具体行为应以 Python SQLite 接口为准。Python sqlite3 文档
验收器为什么必须独立
本实验验收的是“待核验工单创建正确”，判定式是：查询到工单、ID 为整数、action 与可信 expected 相等。它没有核验来源内容的真实性。
验收独立指：标准来自用户目标，证据来自资源读取，不接受模型一句“已完成”作为成功条件。这里查询与创建仍由同一测试服务提供，所以它可以发现丢响应和错误内容，不能抵抗恶意服务伪造结果。
换成代码任务时，验收可以是指定测试、构建产物与 Diff；换成研究任务时，需要把来源是否支持结论拆成可复核的检查。验收复杂度跟任务走，不能用“工具调用成功”统一代替。
用三个退出点检查恢复
| 故障位置 | 退出时资源数 | 恢复动作 | 最终资源数 / 派发预算消耗 |
|---|---|---|---|
| 客户端保存意图后、调用前 | 0 | 查询为空，获准后再次派发 | 1 / 2 |
| 服务端提交后、响应前 | 1 | 查询到资源并验收 | 1 / 1 |
| 客户端收到结果后、验收前 | 1 | 查询到资源并验收 | 1 / 1 |
这三种情况在实验中通过真实进程退出重现，不只是函数抛异常。恢复命令要复用状态目录和 run ID；把状态目录换掉，就不是恢复原运行。
什么时候要换持久化方案
单 Worker 可以使用这里的小状态机。多 Worker 会引入新的问题：谁拥有运行、谁能消耗预算、旧 Worker 恢复后是否还能写。此时需要租约、版本或 fencing 等所有权设计，单靠工单唯一键只能阻止同键重复资源，管不了调度竞争。
分支、暂停和消息历史增多时，可以把运行状态交给工作流框架；工具的幂等和业务验收仍需保留。先以并发与一致性中的反例确定需要的保证，再换数据库或编排器。
本实验创建一张本地待核验工单：来源为 source-001，标题为“核验 Agent Harness 资料”。运行后应只存在一条字段匹配的工单；写入后进程退出，再次运行也不能重复创建。下面按模块安装、执行并检查这个结果。
下载 Mini Harness 实验包。包内默认使用固定模型响应，MCP 子进程、传输和 SQLite 写入真实执行。
先对照装配清单
lab.py: execute() 组装入口与运行状态机
├─ model.py: propose() 上下文、模型调用与严格提案解析
├─ store.py: Runs 契约、提案、意图、预算与事件
├─ session_for() → call() MCP 连接与结构化结果
│ └─ server.py 查询、创建两个工具
│ └─ store.py: Tickets 幂等业务资源
└─ 查询结果对照 expected 完成验收与记录终态没有单独的策略服务：目标一致性检查与 execute_write 在运行器内。没有独立上下文框架：单动作任务的输入在 propose() 内组装。先保留这种紧凑实现，职责增长后再拆文件。
正常路径用两条命令完成。响应丢失后的路径复用运行记录和操作键，验收回到真实工单。
安装并验证各模块契约
解压进入 harness-integration 目录；已记录的实验环境为 Python 3.11.9。以下使用 macOS/Linux 路径：
python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt
.venv/bin/python -m unittest discover -p 'test_*.py' -vWindows 使用虚拟环境中的 Scripts/python.exe。固定依赖包含 mcp==1.29.1；客户端检查协议基线 2025-11-25。安装需要网络，默认测试不调用外部模型。
16 项测试分为：8 项模型输入输出边界、2 项默认与正常运行、3 项真实退出恢复、2 项目标一致性、1 项本机 HTTP 正常路径。测试文件与 VERIFICATION.md 提供对应证据。
按两步完成一次任务
.venv/bin/python lab.py --state-dir ./state-demo
.venv/bin/python lab.py --state-dir ./state-demo --execute第一条把模型提案交给状态库，查询为空后返回 waiting_execution。第二条复用已保存提案，允许创建，独立查询通过后返回 succeeded。事件中应依次出现提案保存、会话初始化、执行意图和验收通过；跨两次运行会有多次会话初始化。
这时检查 attempts 为 1，并读实际资源：
.venv/bin/python - <<'PYCODE'
import sqlite3
with sqlite3.connect('state-demo/tickets.sqlite') as db:
print(db.execute('SELECT id, op, payload FROM tickets').fetchall())
PYCODE应只有一行，操作键为 review-001:ticket，字段匹配默认目标。查询数据库用于学习和测试；在真实远端系统中应使用受授权的资源查询接口。
从代码调用组装入口
在实验目录另存 run_example.py，即可用不同任务调用同一组模块：
import asyncio
from pathlib import Path
from lab import execute
result = asyncio.run(execute(
Path('state-example'),
run='review-002',
expected={'source': 'source-002', 'title': '检查报告的来源'},
provider='fixture',
execute_write=True,
))
print(result['state'], result['attempts'])用 .venv/bin/python run_example.py 运行。这个接口示例对应现有实现；复用同一 run ID 时保持任务配置一致。
让模块在退出后重新接上
每个故障选择一个从未使用过的状态目录：
.venv/bin/python lab.py --state-dir ./state-crash --execute --fault server-after-commit
.venv/bin/python lab.py --state-dir ./state-crash第一条预期非零退出；第二条查询已提交的工单并验收，attempts 仍为 1。再分别尝试 client-after-result 与 client-after-intent：前者恢复只需查询；后者资源尚不存在，恢复命令需要追加 --execute，最终 attempts 为 2。
检查状态与验收的资源数表，而不只看退出码。已经成功的运行直接返回终态，所以旧的成功目录不能用于重新触发故障。
第一次替换：接在线模型
在终端设置 ANTHROPIC_API_KEY 和账号实际可用的 ANTHROPIC_MODEL，再运行：
.venv/bin/python lab.py --state-dir ./state-live --provider anthropic --model "$ANTHROPIC_MODEL"这会把合成任务发往 Anthropic Messages API，并可能产生模型费用。默认只保存提案和查询；确认动作后，保留相同参数追加 --execute。其他模块无需改动。
本次没有调用在线模型，不能据此给出真实模型的成功率、费用或延迟。接入后应另外保留响应类型、提案通过率和用量证据，密钥与完整敏感输入不要进入事件日志。
下一步按需求扩展哪一块
| 你要增加的任务能力 | 先改的模块 | 增加的验收证据 |
|---|---|---|
| 读材料后再决定下一步 | 模型接口、消息历史、有界循环 | 工具结果进入下一轮，轮数和总截止时间有效 |
| 更换真实工单系统 | 工具服务与幂等映射 | 响应丢失后能查询；同键同参数不重复 |
| 长任务暂停后继续 | 状态存储、消息与预算检查点 | 重启不丢决定性上下文，不重置额度 |
| 开放文件与 Shell | 执行策略、沙箱 | 越界路径和网络动作被执行层拒绝 |
| 多 Worker 调度 | 所有权、并发状态更新 | 旧 Worker 不能继续提交过期动作 |
| 引入 Skills 或记忆 | 上下文组装与来源管理 | 对相同任务集确实提高验收率或减少成本 |
第一版的完成标准是：你能替换模型或工具，同时保持任务契约、执行策略、恢复和验收不被绕开。它仍是单用户、单动作的本地基线；上表提供了从这块切片扩展到多轮系统的具体路线。