构建订单与退款交付验收助手
本实战将在智能体开发服务平台（AgentWorks）中构建一个面向电商数据接收方的交付验收助手。合作方每天交付订单或退款批次后，使用者只需说明交付 ID；助手会按约定生成本次报告的名称，先从已批准的校验器清单中准备一份可核对的计划，再在人工允许后运行按固定规则编写的校验程序，解释 PASS 或 FAIL 结果，并把 JSON 报告写入对象存储。
准备条件已经满足时，完成本实战通常需要 75–120 分钟。首次配置对象存储挂载或沙箱时可能需要更长时间。
重要
本实战依赖外部对象存储
本实战不能只在 AgentWorks 内完成。开始前，需要准备由组织批准的阿里云 OSS 或火山引擎 TOS 非生产测试存储桶、最小权限访问凭证，以及能够从存储端下载并核对报告的方式。
AgentWorks 通过沙箱挂载访问这些目录；对象存储账号、服务可用性、费用、数据保留和访问审计仍由相应服务及你的组织管理。没有获批测试范围时，请勿继续。AccessKey 和 Secret Access Key 只填写到 AgentWorks 的凭证管理中，不要写入文件、命令、提示词、截图或验收记录。
本实战将帮助你理解
- 如何把一句验收请求转换成可核对的执行计划，并从已批准的校验器清单中精确选择订单或退款校验规则。
- 为什么计划和实际校验都要经过人工审批，以及校验程序、助手和审批者分别决定什么。
- 如何区分数据校验的 PASS 或 FAIL 与工具执行失败，并从对象存储端确认报告已经持久写入。
理解 AgentWorks 如何完成数据交付验收
可以把“数据交付验收”理解为数据进入业务系统前的收货检验。合作方交付每日订单或退款文件，接收方用自己批准的规则检查文件是否齐全、格式是否正确、内容是否与 manifest（批次清单）一致，再决定是否导入、退回或等待重新提交。
本实战让同一个助手处理两类交付。使用者不需要记住校验脚本和策略文件的位置；助手先调用只读的 plan 命令，由确定性脚本根据 manifest 中的交付类型和版本精确匹配已批准的校验器清单，再把准备使用的校验器、策略、哈希和报告路径展示出来。使用者确认后，助手才请求实际运行。
validator-catalog.json 和 accept_delivery.py 是本实战提供的接收方工件，不是 AgentWorks 的内置校验器管理能力。用于实际业务时，它们应由数据接收方维护和审核；AgentWorks 在这里提供沙箱挂载、工具调用审批、执行记录和结果解释。
只运行一个固定校验脚本当然更简单。本实战使用小型合成数据，是为了展示交付类型和规则逐渐增多后仍然适用的处理方式：使用者通过同一个入口发起验收，执行前可以看到匹配结果，程序只接受已经登记并固定哈希的校验器，最后留下便于复核的结构化报告。增加这些步骤不是为了让模型判断数据，而是为了约束它可以运行什么，并让人员看清实际运行内容。
下图展示一次验收中的职责和数据流。两个黄色审批节点分别表示计划命令和实际运行命令都要经过人工审批；右侧的导入、退回或重新提交决定发生在 AgentWorks 之外。
图中有四条需要保持的边界：
- 计划只说明准备怎样验收。
plan读取 manifest 和已批准的校验器清单，但不运行校验器，也不生成 PASS 或 FAIL 报告。 - 提示词和校验器清单本身不是执行约束。 实际运行时，
accept_delivery.py会重新核对清单匹配、文件路径、哈希、validator_id和plan_sha256；不一致时安全停止。 - 确定性校验器决定 PASS 或 FAIL。 这里的校验器是按固定规则运行的 Python 程序；助手解释结构化结果，但不会代替订单或退款规则判断数据是否合格。
- 后续业务动作仍在 AgentWorks 之外。 本实战不会自动导入数据、退回批次或通知交付方。
准备样例和存储
本阶段先确认 AgentWorks 和外部对象存储满足测试条件，再上传固定的订单和退款样例。
确认准备条件
开始前确认：
- 模型凭证和模型能够正常回复。尚未验证时，先完成创建第一个智能体并完成基础验证。
- 已获得创建对象存储凭证、沙箱实例、智能体模板和智能体实例，以及查看 Trace 和 沙箱运行记录的权限。
- 租户已有包含 Python 3 和命令工具的沙箱规格与环境。
- 存储所有者已经批准一个不含生产数据、个人信息、密钥或客户资料的阿里云 OSS 或火山引擎 TOS 测试范围。
- 平台管理员已经说明测试沙箱的运行超时、空闲回收和计费方式，并确认完成后由谁处理沙箱、凭证和外部测试对象。
本实战包含四种操作角色：存储所有者批准对象存储范围和凭证范围，AgentWorks 构建者创建沙箱与智能体，测试人员发起验收并核对结果，审批者在每次工具运行前核对命令。可以由同一人承担多个角色，但不能省略相应检查。
准备对象存储目录和样例
本阶段由存储所有者或获准操作测试存储的人员完成。AgentWorks 构建者记录获批的存储桶、子目录和凭证范围，供下一阶段配置挂载。
在测试存储桶（Bucket）中准备三个互不重叠的子目录：
| 用途 | Bucket 子目录示例 | 沙箱目录 | 权限 |
|---|---|---|---|
| 已批准的校验器清单、策略和程序 | agentworks-recipes/data-delivery/team-a/rules | /delivery-rules | 只读 |
| 合作方交付的 manifest 和数据文件 | agentworks-recipes/data-delivery/team-a/input | /delivery-input | 只读 |
| 验收报告 | agentworks-recipes/data-delivery/team-a/reports | /delivery-reports | 读写 |
将 team-a 换成本次测试专用的短名称。对象存储凭证应允许三个目录均可列出和读取，只允许在报告目录创建测试对象，不授予删除权限。由于同一沙箱表单中的挂载共用一个凭证，请通过对象存储权限限制各子目录的写入范围，并为每次运行使用新的报告对象名。
下载订单与退款交付验收样例包和样例清单。
也可以在这里查看或下载样例清单，核对包和文件的校验值。
manifest.jsonJSON2.8 KiB下载原始文件 json{ "name": "agentworks-data-delivery-acceptance-assistant", "version": "2.0.0", "description": "Retail orders and refunds deliveries, approved validator catalog, deterministic plan-and-run dispatcher, fixed report projections, mount expectations, and disposition inventory for the AgentWorks data-delivery acceptance recipe.", "bundle": { "path": "data-delivery-acceptance-assistant-v2.zip", "sha256": "e77aa23e307dd594123dee0333a969e6acdaac13ed5adb91463e152258418834" }, "files": [ { "path": "README.md", "sha256": "29d31adbfb273dbb2b25c14a88cb9663fb8e8e092a8eaf62c7d54c71050a4dda" }, { "path": "cleanup-inventory.json", "sha256": "79062f1e5af32d402cc9bfa7f174dfa48c21b673b9964e0071d2c2a1449eb183" }, { "path": "expected-reports.json", "sha256": "e057f5f3ce40e639205c81d8ba6500e20ce410b850f0a1f76ddd1761229cbb6a" }, { "path": "input/RETAIL-INVENTORY-UNSUPPORTED-001/inventory.csv", "sha256": "e101b24882a206a79fd6e74b2e2e60efb33b3c38274fa7ca328baa1c66326ed5" }, { "path": "input/RETAIL-INVENTORY-UNSUPPORTED-001/manifest.json", "sha256": "36b8b4062d6661af1dcee70aeba3ef7e4f8be261c0235b9407e5fac34ef8d9eb" }, { "path": "input/RETAIL-ORDERS-PASS-001/manifest.json", "sha256": "711a8f2972a52de384ce4e3b26642496268143d43ac44195c99cc1eef8a517c7" }, { "path": "input/RETAIL-ORDERS-PASS-001/orders.csv", "sha256": "62170a26120e5a2c39eba2b4e57603eb07786e7e88714881651f2c3ac42ea9c1" }, { "path": "input/RETAIL-REFUNDS-FAIL-001/manifest.json", "sha256": "69159185c72d2d32cc9deb85c29dfd7b6ab0dfb2fc4f8fe8bac2fc49558f9db5" }, { "path": "input/RETAIL-REFUNDS-FAIL-001/refunds.csv", "sha256": "0a9b8bac27f2c607943eb5e63c8ca821991da21cf5199ce843e61e948bb86410" }, { "path": "mounts.json", "sha256": "234e8ed250e95c86efc78005f6c7a02e5386d2ed4f2b28ab20d4bfb4def4e288" }, { "path": "rules/accept_delivery.py", "sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79" }, { "path": "rules/policies/retail-orders-v1.json", "sha256": "3abaa0b0848be84b844c668ade74d2b98d21e3b56993371fb56054cf35095016" }, { "path": "rules/policies/retail-refunds-v1.json", "sha256": "21ce1c471cddc0c2d988ea044a478945f69da5117e2f731165a28946553a239e" }, { "path": "rules/validator-catalog.json", "sha256": "7ddd51da8da9479732a501fee43f04ff8fbf9118d8b15b48506a4381a17c5334" }, { "path": "rules/validators/validate_csv.py", "sha256": "b0a33a9a4fa4dced476bb30680a97f140e82f30fec4c64efc34313d3ff79fc4f" }, { "path": "test-cases.json", "sha256": "26facd0885f10556624972d53cc8648e162964274636f1ab49cd2c7d66a68ca1" } ] }计算 ZIP 的 SHA-256，与样例清单中的
bundle.sha256比较；一致后再解压。保持相对目录不变，把以下内容上传到对应的对象存储子目录：
rules/accept_delivery.pyrules/validator-catalog.jsonrules/validators/validate_csv.pyrules/policies/retail-orders-v1.jsonrules/policies/retail-refunds-v1.jsoninput/RETAIL-ORDERS-PASS-001/input/RETAIL-REFUNDS-FAIL-001/input/RETAIL-INVENTORY-UNSUPPORTED-001/（仅用于可选验证）
在对象存储端确认上述对象存在。后面的
plan命令会读取 manifest，并核对清单匹配、相关文件路径和 SHA-256；任一检查失败时，请修正上传或挂载，不要改用未经批准的文件。确认报告目录中没有
RETAIL-ORDERS-PASS-001-run-001.json和RETAIL-REFUNDS-FAIL-001-run-001.json。已有同名对象时，请为本次运行使用新的后缀。
注意
本实战中的验收规则
订单和退款批次使用同一个确定性 CSV 校验程序，但分别匹配 retail-orders-v1 和 retail-refunds-v1，并加载各自的策略文件。真实流程还需要验证交付方身份和业务数据真实性：请从受信系统取得预期值，或者使用经过认证或签名的 manifest。
创建专用沙箱
本阶段由 AgentWorks 构建者在 AgentWorks 控制台完成。操作账号需有相应资源权限，租户需已分配可用沙箱规格和环境。
本实战直接创建一个专用沙箱实例，以便先确认 Python 环境、三个挂载和最小权限凭证，再把同一个沙箱关联到测试智能体。只有需要让多个智能体实例根据相同配置分别创建沙箱时，才需要改用沙箱模板。
按挂载对象存储中的凭证创建步骤，创建名为
recipe-data-delivery-storage-v2的对象存储凭证。按创建沙箱实例确认配置要求，再打开新建沙箱实例页面，创建名为
recipe-data-delivery-sandbox-v2的沙箱实例。选择管理员提供的测试规格和包含 Python 3 的环境。在 存储挂载 中开启挂载，选择
recipe-data-delivery-storage-v2，再选择 新增挂载点 两次。配置三个挂载点：
- 将规则目录挂载到
/delivery-rules，选择 只读。 - 将输入目录挂载到
/delivery-input，选择 只读。 - 将报告目录挂载到
/delivery-reports，选择 读写。
- 将规则目录挂载到
确认三个 本地目录互不重叠，再选择 创建沙箱实例。校验失败时，修正凭证、Bucket、子目录或本地目录后重新提交，不要扩大凭证权限。
不要把这个测试沙箱关联到其他智能体。对象存储挂载是否正确，将由后面的只读计划和存储端报告核验共同确认。
创建订单与退款交付验收助手
本阶段由 AgentWorks 构建者在 AgentWorks 控制台完成。先创建关联专用沙箱的智能体模板，再从该模板创建一个测试智能体实例。
创建智能体模板
打开创建智能体模板页面。
也可以打开 智能体模板，再选择 创建智能体模板。
将 模板名称设为
recipe-data-delivery-assistant-v2，并选择已经验证的 模型凭证和模型。在 系统提示词 中输入：
language-text你是订单与退款交付验收助手。 用户提供 delivery-id 后，将本次教程的 report-key 设为 <delivery-id>-run-001。若用户明确给出另一个唯一 report-key，则使用该值。 然后先且只运行： python3 /delivery-rules/accept_delivery.py plan \ --delivery-id <delivery-id> \ --report-key <report-key> 概括 phase、status、delivery_id、delivery_type、schema_version、selected_validator_id、 dispatcher_sha256、provenance 中的文件路径和 SHA-256、plan_basis、proposed_run、proposed_run_argv、 report_object_name、mount_output_path 和 plan_sha256，然后等待用户明确要求继续。 计划失败或没有精确匹配时， 停止并说明原因，不要猜测校验器或策略。 用户确认后，只运行： python3 /delivery-rules/accept_delivery.py run \ --delivery-id <delivery-id> \ --report-key <report-key> \ --validator-id <plan 的 proposed_run.validator_id> \ --plan-sha256 <plan 返回的 plan_sha256> 不直接运行其他脚本，不改写目录、路径、哈希或 validator_id，不执行交付目录中的文件， 不删除或覆盖对象，不访问网络，也不执行导入、退回、通知或其他下游操作。 运行完成后，概括结构化结果中的 delivery_id、status、failed_checks、 failed_check_count、validator_id、delivery_type、schema_version、dispatcher_sha256、 plan_sha256、provenance_sha256、 report_object_name、mount_output_path 和 next_action。只有确定性校验器的结构化结果可以表示 PASS 或 FAIL； 工具、挂载或报告写入失败时，说明无法确认验收结果。 请在当前调用中完成，不要委派其他任务。关闭默认开启的 启用文件系统。本实战只使用专用沙箱，关闭另一个文件工作区可以避免混淆输入和报告位置。
在 插件配置 中添加 沙箱 - 实例，选择
recipe-data-delivery-sandbox-v2。在 工具权限 中，明确选择并保存 默认行为为 审批。不要添加自动允许规则。
重要
本实战保留两次可见审批
plan是只读操作，但仍会调用命令工具，因此第一次审批用于确认它只读取固定目录并生成计划。第二次审批用于确认实际运行仍使用计划中的delivery_id、report_key、validator_id和plan_sha256。任一次工具、命令、参数或路径不符合本实战时，请选择拒绝。工具审批不会扩大对象存储凭证的权限，参见工具审批规则能代替工具服务的访问授权吗？。
选择 创建智能体模板，确认模板已经保存。
创建智能体实例
打开创建智能体实例页面。
选择刚创建的智能体模板，将智能体实例命名为
recipe-data-delivery-assistant-test-v2，再完成创建。从控制台侧边栏打开 网页版对话，在新标签页选择刚创建的测试实例；确认实例名称后选择 开启新对话。
为订单批次生成只读验收计划
测试人员在网页版对话中核对目标实例并开启新对话，发送：
language-text
请验收批次 RETAIL-ORDERS-PASS-001。命令工具暂停后，审批者确认实际调用为：
language-bash
python3 /delivery-rules/accept_delivery.py plan \
--delivery-id RETAIL-ORDERS-PASS-001 \
--report-key RETAIL-ORDERS-PASS-001-run-001确认没有其他命令、重定向、网络访问或后台执行，再选择 批准。计划结果应表明已经找到精确匹配，并列出以下结构化信息：
phase=plan、status=READY、delivery_id、delivery_type和schema_versionselected_validator_id=retail-orders-v1provenance中的accept_delivery.py、manifest、数据文件、校验器清单、校验器和策略路径及其 SHA-256plan_basis、proposed_run、proposed_run_argv、报告对象名、挂载内报告路径和plan_sha256
plan 会读取 manifest，并核对已批准校验器清单中的匹配、路径和文件哈希。它不会运行 CSV 校验器或写入报告。计划失败、没有精确匹配或任一字段与本实战不一致时，请停止并修正上传或挂载；不要让助手改用其他路径。
plan_sha256 代表这次计划所依据的 accept_delivery.py、manifest、数据文件、校验器清单、校验器、策略和输出参数。实际运行时，accept_delivery.py 会重新计算它；两次审批之间任一相关内容发生变化时，运行会安全停止。
运行订单 PASS 和退款 FAIL 验收
测试人员在网页版对话中发起验收，审批者核对并处理实际运行命令，存储所有者或获准操作人员从对象存储端下载报告。每次验收都使用新的报告键。
危险
从对象存储下载并核对报告
报告写入后，请在对象存储控制台或组织批准的客户端中按完整对象路径下载，并核对内容和 SHA-256。同一沙箱中的读取结果、Trace 或沙箱运行记录可以定位调用，但不能代替存储端核验。
找不到报告、无法读取或哈希不一致时，请把本次写入视为失败，保留沙箱、Trace、运行记录和现有测试对象。修正问题后使用替代测试沙箱或新的报告键重试，不要覆盖已有对象。
使用本页列出的报告键时，可以对照订单和退款报告的固定字段核对交付类型、校验器、状态、失败项、报告对象名和下一步代码。模型生成的解释可以改变措辞，这些结构化字段不应改变。如果为避免重名而更换报告键，报告对象名、挂载内路径和 plan_sha256 也会随之改变。
下面可以直接查看或下载预期报告字段。
json
{
"version": "2.0.0",
"projection_note": "Compare these fixed decision fields with each report downloaded from the object-storage provider. The full report also contains pinned provenance and per-check evidence.",
"reports": [
{
"delivery_id": "RETAIL-ORDERS-PASS-001",
"delivery_type": "retail.orders",
"schema_version": "1.0",
"validator_id": "retail-orders-v1",
"dispatcher_sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79",
"plan_sha256": "9c7156c6e7a79697dcd16da649453954ad6854934014d45589243fd79967dbf8",
"status": "PASS",
"failed_check_count": 0,
"failed_checks": [],
"next_action_code": "REVIEW_FOR_DOWNSTREAM_IMPORT",
"downstream_action_performed": false,
"report_key": "RETAIL-ORDERS-PASS-001-run-001",
"report_object_name": "RETAIL-ORDERS-PASS-001-run-001.json",
"mount_output_path": "/delivery-reports/RETAIL-ORDERS-PASS-001-run-001.json"
},
{
"delivery_id": "RETAIL-REFUNDS-FAIL-001",
"delivery_type": "retail.refunds",
"schema_version": "1.0",
"validator_id": "retail-refunds-v1",
"dispatcher_sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79",
"plan_sha256": "52eba8fede82b2e7767c9a232ae1d44b278995b7d2bd216e2ef1ccf6f237e6f3",
"status": "FAIL",
"failed_check_count": 1,
"failed_checks": ["reason_code_allowed_values"],
"next_action_code": "REQUEST_CORRECTED_DELIVERY",
"downstream_action_performed": false,
"report_key": "RETAIL-REFUNDS-FAIL-001-run-001",
"report_object_name": "RETAIL-REFUNDS-FAIL-001-run-001.json",
"mount_output_path": "/delivery-reports/RETAIL-REFUNDS-FAIL-001-run-001.json"
}
],
"expected_safe_stops": [
{
"delivery_id": "RETAIL-INVENTORY-UNSUPPORTED-001",
"delivery_type": "retail.inventory",
"schema_version": "1.0",
"status": "NOT_READY",
"dispatcher_sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79",
"reason_code": "NO_APPROVED_VALIDATOR",
"failed_checks": ["approved_validator_match"],
"next_action_code": "STOP_AND_ADD_APPROVED_VALIDATOR",
"proposed_run": null,
"report_written": false
}
]
}验收合格的订单批次
在已经显示订单计划的会话中发送：
language-text按刚才的计划继续验收。命令工具暂停后，审批者确认实际调用为：
language-bashpython3 /delivery-rules/accept_delivery.py run \ --delivery-id RETAIL-ORDERS-PASS-001 \ --report-key RETAIL-ORDERS-PASS-001-run-001 \ --validator-id retail-orders-v1 \ --plan-sha256 9c7156c6e7a79697dcd16da649453954ad6854934014d45589243fd79967dbf8validator_id和plan_sha256必须与计划完整一致。若实际调用仍有占位符、字段不同，或者包含其他命令、参数或路径，请拒绝，不要在审批窗口中代为修补。accept_delivery.py会重新计算计划哈希，并核对校验器清单、文件路径、文件哈希和校验器 ID，然后才调用确定性 CSV 校验器。确认结构化结果中的
status=PASS、failed_check_count=0，并包含delivery_id、delivery_type、schema_version、validator_id、dispatcher_sha256、plan_sha256、provenance_sha256、报告对象名、挂载内报告路径和next_action。provenance_sha256应分别列出accept_delivery.py、manifest、数据文件、校验器清单、校验器和策略的 SHA-256。按结果中的报告对象名从对象存储端下载 JSON 报告，记录 SHA-256，并确认报告内容与助手解释一致。输入目录中的 manifest 和订单文件不应发生变化。
在 Trace 和 沙箱运行记录中确认计划和实际运行是两次独立、经过审批的命令调用。
识别不合格的退款批次
打开另一个 网页版对话新会话，发送：
language-text请验收批次 RETAIL-REFUNDS-FAIL-001。审批者确认
plan命令只包含该交付 ID 和报告键，再选择 批准。计划应精确匹配selected_validator_id=retail-refunds-v1，并列出退款策略、共享 CSV 校验器及其 SHA-256。测试人员发送：
language-text按刚才的计划继续验收。命令工具暂停后，审批者确认实际调用为：
language-bashpython3 /delivery-rules/accept_delivery.py run \ --delivery-id RETAIL-REFUNDS-FAIL-001 \ --report-key RETAIL-REFUNDS-FAIL-001-run-001 \ --validator-id retail-refunds-v1 \ --plan-sha256 52eba8fede82b2e7767c9a232ae1d44b278995b7d2bd216e2ef1ccf6f237e6f3确认
validator_id和plan_sha256与退款计划完整一致。仍有占位符或任一字段不一致时，请拒绝，不要在审批窗口中代为修补。确认结构化结果中的
status=FAIL，failed_checks指出不受支持的reason_code，failed_check_count与失败项一致，报告保留计划中的plan_sha256，next_action建议修正后重新提交。助手可以解释失败原因，但不能把 FAIL 改成 PASS，也不能自行退回或修改批次。从对象存储端下载退款 FAIL 报告并记录 SHA-256。确认它没有覆盖订单 PASS 报告，输入文件也没有变化；Trace 和沙箱运行记录应指向本次计划和实际运行。
命令工具、Python、挂载、清单匹配或报告写入失败时，请先恢复运行环境，再使用新的报告键重新验收。只有确定性校验器返回的不合格结果才记为数据验收 FAIL。
完成结果：同一个助手调用接收方维护的计划脚本，由脚本从已批准的校验器清单中分别匹配订单和退款校验配置；对合格订单生成 PASS 报告，对包含不受支持 reason_code 的退款批次生成 FAIL 报告。两份报告使用不同对象名，并且都已从对象存储端下载、核对内容和 SHA-256。
完成可选验证
前面的订单 PASS 和退款 FAIL 已经构成完整教程。以下检查用于进一步理解安全停止和持久化边界，不影响本实战的完成结果。
验证没有清单匹配时安全停止
固定验收用例还包含清单中没有匹配项的 RETAIL-INVENTORY-UNSUPPORTED-001。在 网页版对话新会话中发送：
language-text
请验收批次 RETAIL-INVENTORY-UNSUPPORTED-001。审批者确认调用的仍然只是固定 plan 命令，再选择 批准。由于 manifest 中的交付类型和版本在已批准的校验器清单中没有精确匹配项，命令应返回 status=NOT_READY 和 reason_code=NO_APPROVED_VALIDATOR，proposed_run 应为空。助手不应提出其他校验器，报告目录中也不应出现对应对象。
这个结果不是数据验收 FAIL，因为确定性校验器尚未运行。它证明 accept_delivery.py 在无法选择已批准校验器时停止，而不是由模型猜测一个路径。
完整的固定验收用例如下，也可以下载后逐项核对。
json
{
"version": "2.0.0",
"shared_verification": {
"before_each_plan": "Confirm the exact report object key is absent through the object-storage provider.",
"plan_review": "Inspect selected_validator_id, dispatcher_sha256, all provenance hashes, plan_sha256, and the exact proposed_run fields before approving execution.",
"after_each_run": "Stat and download the exact provider object, compare the fixed report projection, and record its SHA-256.",
"downstream_boundary": "Do not import, publish, replace, or delete downstream data as part of this sample."
},
"cases": [
{
"id": "retail-orders-pass",
"delivery_id": "RETAIL-ORDERS-PASS-001",
"report_key": "RETAIL-ORDERS-PASS-001-run-001",
"expected_validator_id": "retail-orders-v1",
"expected_dispatcher_sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79",
"expected_plan_sha256": "9c7156c6e7a79697dcd16da649453954ad6854934014d45589243fd79967dbf8",
"plan_argv": [
"python3",
"/delivery-rules/accept_delivery.py",
"plan",
"--delivery-id",
"RETAIL-ORDERS-PASS-001",
"--report-key",
"RETAIL-ORDERS-PASS-001-run-001"
],
"run_argv": [
"python3",
"/delivery-rules/accept_delivery.py",
"run",
"--delivery-id",
"RETAIL-ORDERS-PASS-001",
"--report-key",
"RETAIL-ORDERS-PASS-001-run-001",
"--validator-id",
"retail-orders-v1",
"--plan-sha256",
"9c7156c6e7a79697dcd16da649453954ad6854934014d45589243fd79967dbf8"
],
"expected_status": "PASS",
"expected_failed_checks": [],
"expected_next_action_code": "REVIEW_FOR_DOWNSTREAM_IMPORT"
},
{
"id": "retail-refunds-business-rule-fail",
"delivery_id": "RETAIL-REFUNDS-FAIL-001",
"report_key": "RETAIL-REFUNDS-FAIL-001-run-001",
"expected_validator_id": "retail-refunds-v1",
"expected_dispatcher_sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79",
"expected_plan_sha256": "52eba8fede82b2e7767c9a232ae1d44b278995b7d2bd216e2ef1ccf6f237e6f3",
"plan_argv": [
"python3",
"/delivery-rules/accept_delivery.py",
"plan",
"--delivery-id",
"RETAIL-REFUNDS-FAIL-001",
"--report-key",
"RETAIL-REFUNDS-FAIL-001-run-001"
],
"run_argv": [
"python3",
"/delivery-rules/accept_delivery.py",
"run",
"--delivery-id",
"RETAIL-REFUNDS-FAIL-001",
"--report-key",
"RETAIL-REFUNDS-FAIL-001-run-001",
"--validator-id",
"retail-refunds-v1",
"--plan-sha256",
"52eba8fede82b2e7767c9a232ae1d44b278995b7d2bd216e2ef1ccf6f237e6f3"
],
"expected_status": "FAIL",
"expected_failed_checks": ["reason_code_allowed_values"],
"expected_next_action_code": "REQUEST_CORRECTED_DELIVERY"
}
],
"optional_cases": [
{
"id": "retail-inventory-no-approved-validator",
"delivery_id": "RETAIL-INVENTORY-UNSUPPORTED-001",
"report_key": "RETAIL-INVENTORY-UNSUPPORTED-001-run-001",
"plan_argv": [
"python3",
"/delivery-rules/accept_delivery.py",
"plan",
"--delivery-id",
"RETAIL-INVENTORY-UNSUPPORTED-001",
"--report-key",
"RETAIL-INVENTORY-UNSUPPORTED-001-run-001"
],
"expected_status": "NOT_READY",
"expected_dispatcher_sha256": "90509e93490da59e5a40dee03bbc4a02a605b5152131171378566fc737bbff79",
"expected_reason_code": "NO_APPROVED_VALIDATOR",
"expected_failed_checks": ["approved_validator_match"],
"expected_next_action_code": "STOP_AND_ADD_APPROVED_VALIDATOR",
"expected_proposed_run": null,
"expected_report_written": false
}
]
}从新沙箱读回报告
主流程已经要求从对象存储端直接核对报告，不需要再创建沙箱。组织还要求独立验证新挂载时，请按验证对象存储挂载另建只读验证路径，读取一个已确认的 PASS 报告和一个 FAIL 报告，再把读取内容的 SHA-256 与存储端记录比较。不要放宽本实战助手只运行 plan 和 run 的限制。
需要更换仍然有效的凭证时，参见轮换对象存储凭证。
排查验收失败
从最早失败的步骤开始检查。挂载配置或运行环境状态仍不明确时，请新建替代测试沙箱并重新配置；可能已经写入报告时，请使用新的报告键。
| 现象 | 检查和恢复 |
|---|---|
| 沙箱创建或 Bucket 预检失败 | 核对存储类型、区域、Endpoint、Bucket、子目录、本地目录和凭证范围。修正配置后重新提交，不要扩大到 Bucket 全局权限。 |
plan 无法读取 manifest、目录或规则文件 | 保持规则和输入挂载只读，核对上传对象、相对目录、挂载路径和凭证读取范围。不要让助手寻找其他文件。 |
plan 返回 status=NOT_READY 和 reason_code=NO_APPROVED_VALIDATOR | 核对 manifest 中的交付类型和版本，以及已批准校验器清单中的对应项。不要通过修改 validator_id 绕过停止结果。 |
run 拒绝目录、路径、哈希或 validator_id | 返回最近一次计划重新核对字段。修正被更改的对象或重新生成计划，不要直接调用校验器。 |
| 命令工具或 Python 3 不可用 | 选择管理员已经准备并允许使用的环境，不要让模型安装未审核依赖或猜测其他工具名称。 |
| 报告目录不能写入，或者无法直接下载报告 | 把写入视为执行失败，保留现有测试资源和运行记录。修正配置后使用替代测试沙箱和新的报告键重试。 |
| 助手回答、报告、Trace 或运行记录不一致 | 以校验器结构化结果和对象存储端报告为准，先确认原调用是否已经写入报告，再决定是否重试。不要补写 PASS。 |
| 新沙箱读取的报告缺失或哈希不同 | 停止使用该挂载，记录完整对象路径并联系管理员。不要删除原沙箱或报告来重新测试。 |
调整为自己的交付规则
本实战用小型合成订单和退款 CSV 讲清完整流程。迁移到其他交付类型时，由数据接收方维护 manifest 约定、已批准的校验器清单、策略和确定性校验程序；交付方只提交 manifest 和数据文件。
至少明确：
- 用哪两个 manifest 字段选择规则，以及没有精确匹配时怎样停止。
- 必需文件、格式、字符编码、列名、数据类型、行数和校验和规则。
- 校验器和策略如何发布版本、固定 SHA-256，并加入已批准的校验器清单。
- 哪些 FAIL 可以修复后重新提交，哪些需要人工复核。
- 报告键的唯一性、报告保留、读取、审计、备份和恢复要求。
- 并发交付、覆盖、删除和对象版本由哪个外部系统控制。
样例校验器面向小型 CSV。真实校验器应流式处理数据，并限制文件大小、行数、执行时间、内存和输出大小。交付 ID 和报告键应作为经过校验的结构化参数传入；不要把未经校验的值拼接到 Shell 命令中，也不要自动批准任意 Shell 命令。
将这一模式用于其他业务场景
同样的计划、审批、确定性校验和报告边界可以用于其他文件交付，但每个场景都需要接收方自己的 manifest 约定、已批准的校验器清单、策略和校验器：
| 行业场景 | 一批待验收的数据 | 可以写入接收方规则的检查 | 通过后通常交给的系统 |
|---|---|---|---|
| 电商与零售 | 每日订单、退款或库存文件 | 文件名、列结构、金额格式、数据行数和校验和 | 数仓、结算或库存系统 |
| 制造与供应链 | 发货清单、批次和质检数据 | 必需文件、批次号、字段格式、记录数量和校验和 | WMS、MES 或质量系统 |
| 金融与支付 | 对账或清算批次 | 币种、笔数、总金额，以及与受信控制数的一致性 | 对账或清算系统 |
| AI 与数据工程 | 训练数据分片或评测集 | 文件清单、格式、分片数量、元数据列和校验和 | 训练、评测或数据处理流程 |
PASS 只表示本批文件通过已经配置的检查，不自动证明交付方身份或业务数据真实，也不表示 AgentWorks 已经完成导入、结算或放行。
上线前完成验证
本页提供的是合成数据教程，不是已经针对你的生产规则完成的验证。上线前，请在非生产范围使用最终的最小权限凭证和正式批准的校验器清单，重新运行一个 PASS 和一个 FAIL 用例，并从对象存储端确认两份报告。
同时确认生产发布检查中的权限、恢复、观测和回退要求。还应明确校验规则的负责人、目录变更审批、凭证轮换、报告保留、对象清理、费用管理，以及下游系统如何消费或拒绝报告。
完成后处理测试资源
本实战不包含删除凭证、资源或对象的步骤。完成验证并保留所需报告后，请记录本实战创建的对象存储凭证、沙箱实例、智能体模板、智能体实例和测试对象，并移交给准备条件中确认的资源负责人。
负责人应先确认这些资源没有被其他任务引用，再按组织流程保留或清理。对象存储中的报告和样例由存储所有者按照既定的数据保留策略处理；不要为了清理测试对象而扩大本实战凭证的删除权限。沙箱的回收和费用边界参见说明沙箱如何启动、回收和计费。
查看样例校验实现
需要理解或调整验收规则时，可以查看或下载下面两份样例程序。运行本实战时仍使用完整样例包，保留程序、策略和数据的相对目录。
计划与运行入口
accept_delivery.py 生成验收计划，并在运行前重新核对清单、路径、哈希和校验器 ID。
显示代码隐藏代码，文件 accept_delivery.py605 行
python
import argparse
import hashlib
import json
import re
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
ACCEPTANCE_RUNNER_VERSION = "2.1.0"
CATALOG_VERSION = "2.0.0"
TOKEN_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
CATALOG_REQUIRED_FIELDS = (
"validator_id",
"delivery_type",
"schema_version",
"validator_path",
"validator_sha256",
"policy_path",
"policy_sha256",
)
class ContractError(ValueError):
pass
@dataclass(frozen=True)
class RuntimeRoots:
rules_root: Path = Path("/delivery-rules")
input_root: Path = Path("/delivery-input")
report_root: Path = Path("/delivery-reports")
def read_bytes(path, label):
try:
return path.read_bytes()
except OSError as error:
raise ContractError(f"{label} is not readable: {error}") from error
def sha256_bytes(value):
return hashlib.sha256(value).hexdigest()
def parse_json_object(value, label):
try:
parsed = json.loads(value.decode("utf-8"))
except (UnicodeError, json.JSONDecodeError) as error:
raise ContractError(f"{label} is not readable JSON: {error}") from error
if not isinstance(parsed, dict):
raise ContractError(f"{label} root must be an object")
return parsed
def validate_token(value, label):
if not isinstance(value, str) or not TOKEN_PATTERN.fullmatch(value) or value in {".", ".."}:
raise ContractError(f"{label} contains unsupported characters")
def require_nonempty_string(value, label):
if not isinstance(value, str) or not value or value.strip() != value:
raise ContractError(f"{label} must be a non-empty string without surrounding whitespace")
return value
def require_normalized_relative_path(value, label):
value = require_nonempty_string(value, label)
if "\\" in value:
raise ContractError(f"{label} must use a normalized POSIX relative path")
relative = PurePosixPath(value)
if relative.is_absolute() or relative.as_posix() != value or any(part in {"", ".", ".."} for part in relative.parts):
raise ContractError(f"{label} must use a normalized POSIX relative path")
return value
def resolve_exact_child(root, name, label, expect_directory=False):
validate_token(name, label)
resolved_root = root.resolve()
candidate = (resolved_root / name).resolve()
if candidate.parent != resolved_root:
raise ContractError(f"{label} is outside the allowed root")
if expect_directory and not candidate.is_dir():
raise ContractError(f"{label} directory is unavailable")
return candidate
def resolve_contained_file(root, relative_path, label):
relative_path = require_normalized_relative_path(relative_path, label)
resolved_root = root.resolve()
candidate = (resolved_root / Path(relative_path)).resolve()
try:
candidate.relative_to(resolved_root)
except ValueError as error:
raise ContractError(f"{label} is outside the allowed root") from error
if not candidate.is_file():
raise ContractError(f"{label} is unavailable")
return candidate
def require_sha256(value, label):
if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
raise ContractError(f"{label} must be a lowercase SHA-256 value")
return value
def validate_catalog(catalog_path, rules_root):
catalog_bytes = read_bytes(catalog_path, "validator catalog")
catalog = parse_json_object(catalog_bytes, "validator catalog")
if catalog.get("version") != CATALOG_VERSION:
raise ContractError(f"validator catalog version must be exactly {CATALOG_VERSION}")
validators = catalog.get("validators")
if not isinstance(validators, list) or not validators:
raise ContractError("validator catalog validators must be a non-empty array")
validated = []
seen_validator_ids = set()
seen_type_versions = set()
for index, raw_entry in enumerate(validators):
label = f"validator catalog entry {index}"
if not isinstance(raw_entry, dict):
raise ContractError(f"{label} must be an object")
entry = {
field: require_nonempty_string(raw_entry.get(field), f"{label} {field}")
for field in CATALOG_REQUIRED_FIELDS
}
validate_token(entry["validator_id"], f"{label} validator_id")
if entry["validator_id"] in seen_validator_ids:
raise ContractError("validator catalog validator_id values must be unique")
seen_validator_ids.add(entry["validator_id"])
type_version = (entry["delivery_type"], entry["schema_version"])
if type_version in seen_type_versions:
raise ContractError("validator catalog delivery_type and schema_version pairs must be unique")
seen_type_versions.add(type_version)
entry["validator_path"] = require_normalized_relative_path(
entry["validator_path"],
f"{label} validator_path",
)
entry["policy_path"] = require_normalized_relative_path(
entry["policy_path"],
f"{label} policy_path",
)
entry["validator_sha256"] = require_sha256(
entry["validator_sha256"],
f"{label} validator_sha256",
)
entry["policy_sha256"] = require_sha256(
entry["policy_sha256"],
f"{label} policy_sha256",
)
validator_path = resolve_contained_file(
rules_root,
entry["validator_path"],
f"{label} validator_path",
)
policy_path = resolve_contained_file(
rules_root,
entry["policy_path"],
f"{label} policy_path",
)
validator_bytes = read_bytes(validator_path, f"{label} validator")
policy_bytes = read_bytes(policy_path, f"{label} policy")
if sha256_bytes(validator_bytes) != entry["validator_sha256"]:
raise ContractError(f"{label} validator SHA-256 does not match the catalog")
if sha256_bytes(policy_bytes) != entry["policy_sha256"]:
raise ContractError(f"{label} policy SHA-256 does not match the catalog")
policy = parse_json_object(policy_bytes, f"{label} policy")
for field in ("validator_id", "delivery_type", "schema_version"):
if policy.get(field) != entry[field]:
raise ContractError(f"{label} policy {field} does not match the catalog")
entry["validator_path_resolved"] = validator_path
entry["policy_path_resolved"] = policy_path
entry["policy"] = policy
validated.append(entry)
return catalog_bytes, validated
def plan_basis(context):
return {
"contract_version": "2.0",
"delivery_id": context["delivery_id"],
"delivery_type": context["delivery_type"],
"schema_version": context["schema_version"],
"report_key": context["report_key"],
"validator_id": context["validator_id"],
"selected_paths": {
"dispatcher": "rules/accept_delivery.py",
"catalog": "rules/validator-catalog.json",
"manifest": f"input/{context['delivery_id']}/manifest.json",
"data": f"input/{context['delivery_id']}/{context['manifest']['file']}",
"validator": f"rules/{context['validator_relative_path']}",
"policy": f"rules/{context['policy_relative_path']}",
"report_object": context["report_path"].name,
},
"provenance_sha256": {
"dispatcher": context["provenance"]["dispatcher_sha256"],
"catalog": context["provenance"]["catalog_sha256"],
"manifest": context["provenance"]["manifest_sha256"],
"data": context["provenance"]["data_sha256"],
"validator": context["provenance"]["validator_sha256"],
"policy": context["provenance"]["policy_sha256"],
},
}
def canonical_sha256(value):
encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def inspect_delivery(delivery_id, report_key, roots, allow_no_match=False):
validate_token(delivery_id, "delivery-id")
validate_token(report_key, "report-key")
rules_root = roots.rules_root.resolve()
input_root = roots.input_root.resolve()
report_root = roots.report_root.resolve()
if not rules_root.is_dir():
raise ContractError("rules root is unavailable")
if not input_root.is_dir():
raise ContractError("input root is unavailable")
if not report_root.is_dir():
raise ContractError("report root is unavailable")
dispatcher_path = resolve_contained_file(rules_root, "accept_delivery.py", "dispatcher path")
if dispatcher_path != Path(__file__).resolve():
raise ContractError("running dispatcher does not match the rules-root dispatcher")
dispatcher_bytes = read_bytes(dispatcher_path, "dispatcher")
dispatcher_sha256 = sha256_bytes(dispatcher_bytes)
delivery_dir = resolve_exact_child(input_root, delivery_id, "delivery-id", expect_directory=True)
manifest_path = resolve_contained_file(delivery_dir, "manifest.json", "manifest")
manifest_bytes = read_bytes(manifest_path, "manifest")
manifest = parse_json_object(manifest_bytes, "manifest")
if manifest.get("delivery_id") != delivery_id:
raise ContractError("manifest delivery_id does not match --delivery-id")
delivery_type = require_nonempty_string(manifest.get("delivery_type"), "manifest delivery_type")
schema_version = require_nonempty_string(manifest.get("schema_version"), "manifest schema_version")
declared_file = require_normalized_relative_path(manifest.get("file"), "manifest file")
data_path = resolve_contained_file(delivery_dir, declared_file, "manifest file")
if data_path.parent != delivery_dir or data_path.name != declared_file:
raise ContractError("manifest file must be a direct child of the delivery directory")
data_bytes = read_bytes(data_path, "data file")
expected_data_sha256 = require_sha256(manifest.get("sha256"), "manifest sha256")
actual_data_sha256 = sha256_bytes(data_bytes)
if actual_data_sha256 != expected_data_sha256:
raise ContractError("data-file SHA-256 does not match the manifest")
if not isinstance(manifest.get("rows"), int) or manifest["rows"] < 0:
raise ContractError("manifest rows must be a non-negative integer")
report_path = resolve_exact_child(report_root, f"{report_key}.json", "report object")
if report_path.exists():
raise ContractError("report object already exists; choose a new report-key")
catalog_path = resolve_contained_file(rules_root, "validator-catalog.json", "validator catalog")
catalog_bytes, validators = validate_catalog(catalog_path, rules_root)
matches = [
entry
for entry in validators
if entry["delivery_type"] == delivery_type and entry["schema_version"] == schema_version
]
shared_provenance = {
"dispatcher_path": str(dispatcher_path),
"dispatcher_sha256": dispatcher_sha256,
"catalog_path": str(catalog_path),
"catalog_sha256": sha256_bytes(catalog_bytes),
"manifest_path": str(manifest_path),
"manifest_sha256": sha256_bytes(manifest_bytes),
"data_path": str(data_path),
"data_sha256": actual_data_sha256,
}
if not matches and allow_no_match:
return {
"delivery_id": delivery_id,
"delivery_type": delivery_type,
"schema_version": schema_version,
"report_key": report_key,
"report_path": report_path,
"manifest_path": manifest_path,
"manifest": manifest,
"data_path": data_path,
"catalog_path": catalog_path,
"dispatcher_path": dispatcher_path,
"validator_id": None,
"provenance": shared_provenance,
}
if len(matches) != 1:
raise ContractError("validator catalog must contain exactly one matching delivery_type and schema_version")
selected = matches[0]
if selected["policy"].get("required_file") != declared_file:
raise ContractError("manifest file does not match the selected policy")
validator_path = selected["validator_path_resolved"]
policy_path = selected["policy_path_resolved"]
provenance = {
**shared_provenance,
"validator_path": str(validator_path),
"validator_sha256": selected["validator_sha256"],
"policy_path": str(policy_path),
"policy_sha256": selected["policy_sha256"],
}
return {
"delivery_id": delivery_id,
"delivery_type": delivery_type,
"schema_version": schema_version,
"report_key": report_key,
"report_path": report_path,
"dispatcher_path": dispatcher_path,
"catalog_path": catalog_path,
"manifest_path": manifest_path,
"manifest": manifest,
"data_path": data_path,
"validator_id": selected["validator_id"],
"validator_path": validator_path,
"validator_relative_path": selected["validator_path"],
"policy_path": policy_path,
"policy_relative_path": selected["policy_path"],
"provenance": provenance,
}
def build_plan(context):
if context["validator_id"] is None:
return {
"phase": "plan",
"status": "NOT_READY",
"reason_code": "NO_APPROVED_VALIDATOR",
"delivery_id": context["delivery_id"],
"delivery_type": context["delivery_type"],
"schema_version": context["schema_version"],
"selected_validator_id": None,
"dispatcher_sha256": context["provenance"]["dispatcher_sha256"],
"provenance": context["provenance"],
"proposed_run": None,
"report_object_name": context["report_path"].name,
"mount_output_path": str(context["report_path"]),
"failed_checks": ["approved_validator_match"],
"next_action": {
"code": "STOP_AND_ADD_APPROVED_VALIDATOR",
"summary": "Do not run this delivery. Ask the rule owner to add and approve an exact type-and-version catalog entry first.",
},
}
basis = plan_basis(context)
plan_sha256 = canonical_sha256(basis)
proposed_run = {
"subcommand": "run",
"delivery_id": context["delivery_id"],
"report_key": context["report_key"],
"validator_id": context["validator_id"],
"plan_sha256": plan_sha256,
}
return {
"phase": "plan",
"status": "READY",
"delivery_id": context["delivery_id"],
"delivery_type": context["delivery_type"],
"schema_version": context["schema_version"],
"selected_validator_id": context["validator_id"],
"dispatcher_sha256": context["provenance"]["dispatcher_sha256"],
"provenance": context["provenance"],
"plan_basis": basis,
"plan_sha256": plan_sha256,
"proposed_run": proposed_run,
"proposed_run_argv": [
"python3",
"/delivery-rules/accept_delivery.py",
"run",
"--delivery-id",
context["delivery_id"],
"--report-key",
context["report_key"],
"--validator-id",
context["validator_id"],
"--plan-sha256",
plan_sha256,
],
"report_object_name": context["report_path"].name,
"mount_output_path": str(context["report_path"]),
"failed_checks": [],
"next_action": {
"code": "REVIEW_AND_APPROVE_RUN",
"summary": "Review the selected validator, pinned hashes, and proposed run fields before allowing execution.",
},
}
def next_action_for(status):
if status == "PASS":
return {
"code": "REVIEW_FOR_DOWNSTREAM_IMPORT",
"summary": "Review the evidence and decide whether the batch may enter the downstream system; no import was performed.",
}
return {
"code": "REQUEST_CORRECTED_DELIVERY",
"summary": "Ask the supplier to correct the failed checks and resubmit under a new delivery ID; no downstream action was performed.",
}
def provenance_sha256(context):
return {
name: context["provenance"][f"{name}_sha256"]
for name in ("dispatcher", "catalog", "manifest", "data", "validator", "policy")
}
def read_verified_source(path, expected_sha256, label):
value = read_bytes(path, label)
if sha256_bytes(value) != expected_sha256:
raise ContractError(f"{label} changed after inspection")
return value
def write_private_snapshot(path, value):
try:
with path.open("xb") as target:
target.write(value)
except OSError as error:
raise ContractError(f"unable to create private validation snapshot: {error}") from error
def invoke_validator_snapshot(context):
expected = context["provenance"]
read_verified_source(context["dispatcher_path"], expected["dispatcher_sha256"], "dispatcher")
read_verified_source(context["catalog_path"], expected["catalog_sha256"], "validator catalog")
validator_bytes = read_verified_source(
context["validator_path"],
expected["validator_sha256"],
"validator",
)
manifest_bytes = read_verified_source(
context["manifest_path"],
expected["manifest_sha256"],
"manifest",
)
data_bytes = read_verified_source(context["data_path"], expected["data_sha256"], "data file")
policy_bytes = read_verified_source(context["policy_path"], expected["policy_sha256"], "policy")
try:
temporary_directory = tempfile.TemporaryDirectory(prefix="agentworks-delivery-validation-")
except OSError as error:
raise ContractError(f"unable to create private validation directory: {error}") from error
with temporary_directory as temporary_path:
snapshot_root = Path(temporary_path)
validator_snapshot = snapshot_root / "validate_csv.py"
manifest_snapshot = snapshot_root / "manifest.json"
data_snapshot = snapshot_root / context["data_path"].name
policy_snapshot = snapshot_root / context["policy_path"].name
for path, value in (
(validator_snapshot, validator_bytes),
(manifest_snapshot, manifest_bytes),
(data_snapshot, data_bytes),
(policy_snapshot, policy_bytes),
):
write_private_snapshot(path, value)
return subprocess.run(
[
sys.executable,
str(validator_snapshot),
"--manifest",
str(manifest_snapshot),
"--data-file",
str(data_snapshot),
"--policy",
str(policy_snapshot),
"--validator-id",
context["validator_id"],
],
check=False,
capture_output=True,
text=True,
timeout=30,
)
def run_validation(context, supplied_validator_id, supplied_plan_sha256):
validate_token(supplied_validator_id, "validator-id")
if supplied_validator_id != context["validator_id"]:
raise ContractError("--validator-id does not match the validator selected from the approved catalog")
supplied_plan_sha256 = require_sha256(supplied_plan_sha256, "plan-sha256")
actual_plan_sha256 = canonical_sha256(plan_basis(context))
if supplied_plan_sha256 != actual_plan_sha256:
raise ContractError("--plan-sha256 does not match the current delivery, catalog, validator, policy, data, or dispatcher state")
completed = invoke_validator_snapshot(context)
if completed.returncode != 0:
raise ContractError(f"validator failed to complete: {completed.stdout.strip() or completed.stderr.strip()}")
try:
result = json.loads(completed.stdout)
except json.JSONDecodeError as error:
raise ContractError("validator output is not JSON") from error
if not isinstance(result, dict) or result.get("status") not in {"PASS", "FAIL"}:
raise ContractError("validator output does not contain a supported status")
if result.get("validator_id") != context["validator_id"]:
raise ContractError("validator output validator_id does not match the approved selection")
failed_checks = result.get("failed_checks")
checks = result.get("checks")
if not isinstance(failed_checks, list) or not all(isinstance(item, str) for item in failed_checks):
raise ContractError("validator output failed_checks must be an array of strings")
if not isinstance(checks, list):
raise ContractError("validator output checks must be an array")
status = result["status"]
if (status == "PASS") != (len(failed_checks) == 0):
raise ContractError("validator status and failed_checks disagree")
next_action = next_action_for(status)
dispatcher_sha256 = context["provenance"]["dispatcher_sha256"]
report = {
"report_schema_version": "2.0",
"acceptance_runner_version": ACCEPTANCE_RUNNER_VERSION,
"validator_engine_version": result.get("validator_engine_version"),
"delivery_id": context["delivery_id"],
"delivery_type": context["delivery_type"],
"schema_version": context["schema_version"],
"validator_id": context["validator_id"],
"dispatcher_sha256": dispatcher_sha256,
"plan_sha256": actual_plan_sha256,
"provenance_sha256": provenance_sha256(context),
"provenance": context["provenance"],
"status": status,
"failed_check_count": len(failed_checks),
"failed_checks": failed_checks,
"checks": checks,
"next_action": next_action,
"downstream_action_performed": False,
"report_key": context["report_key"],
"report_object_name": context["report_path"].name,
"mount_output_path": str(context["report_path"]),
}
try:
with context["report_path"].open("x", encoding="utf-8") as target:
json.dump(report, target, ensure_ascii=False, indent=2)
target.write("\n")
except FileExistsError as error:
raise ContractError("report object already exists; choose a new report-key") from error
return {
"phase": "run",
"status": status,
"delivery_id": context["delivery_id"],
"delivery_type": context["delivery_type"],
"schema_version": context["schema_version"],
"validator_id": context["validator_id"],
"dispatcher_sha256": dispatcher_sha256,
"plan_sha256": actual_plan_sha256,
"provenance_sha256": provenance_sha256(context),
"failed_check_count": len(failed_checks),
"failed_checks": failed_checks,
"next_action": next_action,
"report_object_name": context["report_path"].name,
"mount_output_path": str(context["report_path"]),
"downstream_action_performed": False,
}
def parse_args(argv=None):
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="phase", required=True)
plan_parser = subparsers.add_parser("plan")
plan_parser.add_argument("--delivery-id", required=True)
plan_parser.add_argument("--report-key", required=True)
run_parser = subparsers.add_parser("run")
run_parser.add_argument("--delivery-id", required=True)
run_parser.add_argument("--report-key", required=True)
run_parser.add_argument("--validator-id", required=True)
run_parser.add_argument("--plan-sha256", required=True)
return parser.parse_args(argv)
def execute(argv=None, roots=None):
args = parse_args(argv)
runtime_roots = roots or RuntimeRoots()
context = inspect_delivery(
args.delivery_id,
args.report_key,
runtime_roots,
allow_no_match=args.phase == "plan",
)
if args.phase == "plan":
return build_plan(context)
return run_validation(context, args.validator_id, args.plan_sha256)
def main(argv=None, roots=None):
try:
result = execute(argv, roots)
except (ContractError, subprocess.TimeoutExpired) as error:
print(json.dumps({"status": "ERROR", "error": str(error)}, ensure_ascii=False))
return 2
print(json.dumps(result, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())CSV 校验程序
validate_csv.py 根据所选策略检查 CSV 文件，并生成结构化验收结果。
显示代码隐藏代码，文件 validate_csv.py161 行
python
import argparse
import csv
import json
from datetime import date
from decimal import Decimal, InvalidOperation
from pathlib import Path
VALIDATOR_ENGINE_VERSION = "2.0.0"
def load_json_object(path, label):
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as error:
raise ValueError(f"{label} is not readable JSON: {error}") from error
if not isinstance(value, dict):
raise ValueError(f"{label} root must be an object")
return value
def is_iso_date(value):
try:
date.fromisoformat(value)
except (TypeError, ValueError):
return False
return True
def is_decimal_greater_than(value, threshold):
try:
number = Decimal(value)
minimum = Decimal(threshold)
except (InvalidOperation, TypeError, ValueError):
return False
return number.is_finite() and number > minimum
def add_check(checks, name, passed, expected, actual):
checks.append(
{
"name": name,
"passed": bool(passed),
"expected": expected,
"actual": actual,
}
)
def validate(args):
manifest_path = Path(args.manifest)
data_path = Path(args.data_file)
policy_path = Path(args.policy)
manifest = load_json_object(manifest_path, "manifest")
policy = load_json_object(policy_path, "policy")
if policy.get("validator_id") != args.validator_id:
raise ValueError("policy validator_id does not match --validator-id")
if policy.get("delivery_type") != manifest.get("delivery_type"):
raise ValueError("policy delivery_type does not match the manifest")
if policy.get("schema_version") != manifest.get("schema_version"):
raise ValueError("policy schema_version does not match the manifest")
if policy.get("required_file") != data_path.name:
raise ValueError("policy required_file does not match the selected data file")
expected_columns = policy.get("columns")
required_columns = policy.get("required_columns")
column_rules = policy.get("column_rules")
if not isinstance(expected_columns, list) or not all(isinstance(item, str) for item in expected_columns):
raise ValueError("policy columns must be an array of strings")
if not isinstance(required_columns, list) or not all(isinstance(item, str) for item in required_columns):
raise ValueError("policy required_columns must be an array of strings")
if not isinstance(column_rules, dict):
raise ValueError("policy column_rules must be an object")
try:
with data_path.open("r", encoding="utf-8", newline="") as source:
reader = csv.DictReader(source)
actual_columns = reader.fieldnames or []
rows = list(reader)
except (OSError, UnicodeError, csv.Error) as error:
raise ValueError(f"data file is not readable CSV: {error}") from error
checks = []
add_check(checks, "columns", actual_columns == expected_columns, expected_columns, actual_columns)
add_check(checks, "rows", len(rows) == manifest.get("rows"), manifest.get("rows"), len(rows))
missing_values = []
for row_number, row in enumerate(rows, start=2):
missing = [column for column in required_columns if not (row.get(column) or "").strip()]
if missing:
missing_values.append({"row": row_number, "columns": missing})
add_check(checks, "required_values", not missing_values, "all required values are present", missing_values)
for column, rule in column_rules.items():
if not isinstance(rule, dict) or not isinstance(rule.get("check_name"), str):
raise ValueError(f"policy rule for {column} must contain check_name")
violations = []
expected = {}
if "type" in rule:
expected["type"] = rule["type"]
if "greater_than" in rule:
expected["greater_than"] = rule["greater_than"]
if "allowed_values" in rule:
expected["allowed_values"] = rule["allowed_values"]
for row_number, row in enumerate(rows, start=2):
value = row.get(column)
passed = True
if rule.get("type") == "iso_date":
passed = is_iso_date(value)
elif rule.get("type") == "decimal":
if "greater_than" not in rule:
raise ValueError(f"decimal rule for {column} must contain greater_than")
passed = is_decimal_greater_than(value, rule["greater_than"])
elif "type" in rule:
raise ValueError(f"unsupported policy type for {column}: {rule['type']}")
allowed_values = rule.get("allowed_values")
if allowed_values is not None:
if not isinstance(allowed_values, list) or not all(isinstance(item, str) for item in allowed_values):
raise ValueError(f"allowed_values for {column} must be an array of strings")
passed = passed and value in allowed_values
if not passed:
violations.append({"row": row_number, "value": value})
add_check(checks, rule["check_name"], not violations, expected, violations)
failed_checks = [item["name"] for item in checks if not item["passed"]]
return {
"validator_engine_version": VALIDATOR_ENGINE_VERSION,
"validator_id": args.validator_id,
"status": "PASS" if not failed_checks else "FAIL",
"failed_checks": failed_checks,
"checks": checks,
}
def parse_args(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--manifest", required=True)
parser.add_argument("--data-file", required=True)
parser.add_argument("--policy", required=True)
parser.add_argument("--validator-id", required=True)
return parser.parse_args(argv)
def main(argv=None):
args = parse_args(argv)
try:
result = validate(args)
except ValueError as error:
print(json.dumps({"status": "ERROR", "error": str(error)}, ensure_ascii=False))
return 2
print(json.dumps(result, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())