From d4562a32e5261f49eaad20146770579e4adf0685 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E9=99=88?= Date: Thu, 20 Aug 2026 18:15:42 +0800 Subject: [PATCH] docs: add batch SDK migration workshop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 安陈 --- .../batch-sdk-migration-with-qca/index.md | 254 ++++++++++ .../batch-sdk-migration-with-qca/.env.example | 4 + demos/batch-sdk-migration-with-qca/.gitignore | 5 + .../FORWARD_ENVIRONMENT_SETUP.sh | 21 + .../MIGRATION_GUIDE.md | 35 ++ demos/batch-sdk-migration-with-qca/README.md | 172 +++++++ .../TEMPLATE_SYSTEM_PROMPT.md | 13 + .../check_baseline.py | 54 +++ .../projects/billing/app.py | 7 + .../projects/billing/check_manual_review.py | 17 + .../projects/billing/legacy_sdk.py | 7 + .../projects/billing/modern_sdk.py | 14 + .../projects/catalog/app.py | 8 + .../projects/catalog/legacy_sdk.py | 6 + .../projects/catalog/modern_sdk.py | 25 + .../projects/catalog/test_app.py | 17 + .../projects/inventory/app.py | 7 + .../projects/inventory/legacy_sdk.py | 6 + .../projects/inventory/modern_sdk.py | 14 + .../projects/inventory/test_app.py | 17 + .../projects/orders/app.py | 15 + .../projects/orders/legacy_sdk.py | 12 + .../projects/orders/modern_sdk.py | 23 + .../projects/orders/test_app.py | 21 + .../batch-sdk-migration-with-qca/qca_batch.py | 457 ++++++++++++++++++ demos/batch-sdk-migration-with-qca/tasks.json | 32 ++ .../tests/test_fixture_contract.py | 34 ++ .../tests/test_qca_batch.py | 251 ++++++++++ 28 files changed, 1548 insertions(+) create mode 100644 content/zh-CN/workshops/batch-sdk-migration-with-qca/index.md create mode 100644 demos/batch-sdk-migration-with-qca/.env.example create mode 100644 demos/batch-sdk-migration-with-qca/.gitignore create mode 100644 demos/batch-sdk-migration-with-qca/FORWARD_ENVIRONMENT_SETUP.sh create mode 100644 demos/batch-sdk-migration-with-qca/MIGRATION_GUIDE.md create mode 100644 demos/batch-sdk-migration-with-qca/README.md create mode 100644 demos/batch-sdk-migration-with-qca/TEMPLATE_SYSTEM_PROMPT.md create mode 100644 demos/batch-sdk-migration-with-qca/check_baseline.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/billing/app.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/billing/check_manual_review.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/billing/legacy_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/billing/modern_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/catalog/app.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/catalog/legacy_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/catalog/modern_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/catalog/test_app.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/inventory/app.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/inventory/legacy_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/inventory/modern_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/inventory/test_app.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/orders/app.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/orders/legacy_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/orders/modern_sdk.py create mode 100644 demos/batch-sdk-migration-with-qca/projects/orders/test_app.py create mode 100644 demos/batch-sdk-migration-with-qca/qca_batch.py create mode 100644 demos/batch-sdk-migration-with-qca/tasks.json create mode 100644 demos/batch-sdk-migration-with-qca/tests/test_fixture_contract.py create mode 100644 demos/batch-sdk-migration-with-qca/tests/test_qca_batch.py diff --git a/content/zh-CN/workshops/batch-sdk-migration-with-qca/index.md b/content/zh-CN/workshops/batch-sdk-migration-with-qca/index.md new file mode 100644 index 0000000..fe3c1a1 --- /dev/null +++ b/content/zh-CN/workshops/batch-sdk-migration-with-qca/index.md @@ -0,0 +1,254 @@ +--- +schema_version: 1 +slug: batch-sdk-migration-with-qca +title: 批量升级异构代码项目：用 QCA Batch Agent 完成迁移、测试与修复 +summary: 在 90 分钟 Workshop 中批量迁移四个结构不同的 Python 服务，让 Agent 在独立沙箱里修改代码、运行测试、迭代修复并交付可验证结果。 +type: workshop +category: build-deploy +tags: + - code-execution + - workflow-automation + - testing + - verification + - api +author: + name: 安陈 + github: anchenqlw +locale: zh-CN +--- + +## 学习目标 + +你将扮演一次 SDK 升级项目的负责人：四个 Python 服务都依赖 Legacy Shop SDK，但它们的调用方式、封装层和验收条件各不相同。你需要把四项迁移任务提交给 QCA Batch Agent，并用测试、Patch 和迁移报告验收结果。 + +完成 90 分钟练习后，你将能够： + +- 用 Forward Environment 的 `packages` 和 `setup_script` 准备练习工作区，并配置 Template、Identity 和明确的工具权限。 +- 把多个代码迁移任务写成 JSONL，上传文件、创建 Batch、轮询状态并下载结果。 +- 根据 `custom_id` 跟踪每项任务，用测试和修改范围检查验收迁移结果。 +- 读取 `error.jsonl`，修复输入问题并只重投失败任务。 +- 说明离峰调度、Credit、结果保留、凭据和人工复核边界。 + +### 你将完成的任务 + +Workshop 使用合成的 Legacy Shop SDK 和四个相互隔离的示例服务，不包含客户代码或生产凭据： + +| Batch Task | 迁移内容 | 预期交付物 | +|---|---|---| +| `migrate-catalog` | 适配新的 Product 和 Money 返回对象 | 通过测试的 Patch、迁移报告 | +| `migrate-orders` | 保留 Adapter，迁移幂等参数和异常映射 | 通过测试的 Patch、迁移报告 | +| `migrate-inventory` | 把同步库存接口迁移为异步接口 | 通过异步测试的 Patch、迁移报告 | +| `migrate-billing` | 识别缺失的业务舍入策略 | 人工复核说明、只新增该说明的 Patch | + +每个 Agent 都按照同一条工作链执行： + +```text +读取项目和迁移指南 +→ 修改代码 +→ 运行该项目的测试 +→ 读取新的堆栈和断言差异 +→ 调整实现 +→ 再次运行测试 +→ 通过，或带证据转人工复核 +``` + +课程结束时，你应得到两次 Batch 记录：第一批包含三个正常执行任务和一个故意构造的输入错误；第二批只重投首次失败的 Billing 任务。最终产物包括 Batch 结果文件、四份 Patch、三份迁移报告和一份人工复核说明；Billing 的 Patch 只能新增复核说明。 + +## 课程安排 + +### 0～10 分钟：认识任务与验收标准 + +先查看四个示例服务和 [迁移指南](https://github.com/QoderAI/cloud-agents-cookbook/blob/main/demos/batch-sdk-migration-with-qca/MIGRATION_GUIDE.md)，运行预迁移检查： + +```bash +cd demos/batch-sdk-migration-with-qca +python3 check_baseline.py +``` + +预期看到 Catalog、Orders、Inventory 因不同原因未通过迁移验收，Billing 被标记为缺少业务决策。这些结果构成四项任务的起点。随后查看 `tasks.json`，确认每个任务的项目目录、验收命令和预期交付物。 + +### 10～25 分钟：准备隔离环境 + +开始前需要： + +- Python 3.11 或更高版本。 +- QCA 账号和一个 Personal Access Token（PAT）。 +- Git，用于本地克隆公开的 Cookbook 仓库。 +- 能在 QCA 控制台创建 Environment、Forward Template 和 Identity 的权限。 + +先在本地克隆公开仓库： + +```bash +git clone https://github.com/QoderAI/cloud-agents-cookbook.git +cd cloud-agents-cookbook/demos/batch-sdk-migration-with-qca +``` + +在 QCA 控制台创建一个专用于本练习的 Forward Environment。Packages 中添加 `git`，再把 Demo 的 `FORWARD_ENVIRONMENT_SETUP.sh` 完整复制到 `setup_script`。脚本会在 Environment 中执行： + +```bash +git clone --depth 1 \ + https://github.com/QoderAI/cloud-agents-cookbook.git \ + /workspace/cloud-agents-cookbook +``` + +`setup_script` 由 `/bin/bash -lc` 执行，运行在 Packages 安装之后；非零退出会让 Environment 创建失败。脚本还会检查 Demo 是否存在，因此路径或网络不符合预期时会明确失败，而不是把不完整环境交给 Batch。[Create Forward Environment API](https://docs.qoder.com/cloud-agents/api/forward/environments/create) + +Environment 就绪后，创建 Forward Template： + +| 配置 | 本练习取值 | +|---|---| +| Environment | 刚创建的练习专用 Forward Environment | +| Model | `performance`；创建前可用 List Models API 确认账号当前可选模型 | +| System Prompt | 复制 Demo 中的 `TEMPLATE_SYSTEM_PROMPT.md` | +| 内置工具 | `Bash`、`Read`、`Write`、`Edit`、`Glob`、`Grep`、`DeliverArtifacts` | +| 工具权限 | 仅在这个无生产凭据的示例环境中设为 `always_allow` | + +Forward Mode 在本练习中不挂载 GitHub 仓库；Agent 使用 Environment 已准备好的 `/workspace/cloud-agents-cookbook`。客户也不需要、不能为仓库挂载选择路径。 + +Batch 是无人值守执行。根据当前接口规则，只要所用工具配置为 `always_ask` 或 `always_deny`，对应 JSONL 行就会校验失败。不要为了绕过审批把相同配置照搬到生产仓库；生产任务应先重新设计最小权限、网络、凭据和人工审批边界。 + +再创建一个本练习专用 Identity，记下 `tmpl_` 和 `idn_` 前缀的 ID。QCA 的 [Template API](https://docs.qoder.com/cloud-agents/api/forward/templates/create) 和 [Identity API](https://docs.qoder.com/cloud-agents/api/forward/identities/create) 给出了字段定义。 + +在当前终端设置变量，不要把真实值写进仓库： + +```bash +export QODER_PAT="<你的 QCA PAT>" +export QODER_TEMPLATE_ID="" +export QODER_IDENTITY_ID="" +``` + +### 25～40 分钟：读懂四个迁移任务 + +打开 `tasks.json`。每个任务只允许修改一个项目目录，并附带自己的验收命令。Agent 需要遵循三个共同约束： + +1. 先读迁移指南和目标项目，不能修改测试、Modern SDK 模拟实现或其它项目。 +2. 必须实际运行验收命令；不能用“代码看起来正确”代替测试结果。 +3. 测试通过时交付 `changes.patch` 和 `migration-report.md`；缺少业务决策时交付 `manual-review.md`，不能自行编造策略。 + +这种任务拆分让每个 Batch Task 都有明确作用域和可执行的验收条件。四行任务可由独立 Session 调度；一个项目失败不会阻止其它合法 JSONL 行继续执行，结果按 `custom_id` 分别跟踪。 + +### 40～55 分钟：生成并提交 Batch + +先运行 Demo 自带的离线单元测试。它使用本地 mock 服务，不访问 QCA，也不消耗 Credit： + +```bash +PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests -v +``` + +生成包含四行任务的 JSONL。为了演示“单行校验失败不阻塞其它行”，命令会故意删除 Billing 行的 `identity_id`： + +```bash +python3 qca_batch.py prepare \ + --tasks tasks.json \ + --output .work/batch-input.jsonl \ + --inject-invalid migrate-billing +``` + +检查文件时应只看到任务说明和资源 ID，不应出现任何 token： + +```bash +python3 qca_batch.py inspect --input .work/batch-input.jsonl +``` + +上传 JSONL 并创建 `24h` Batch： + +```bash +export BATCH_ID="$(python3 qca_batch.py submit \ + --input .work/batch-input.jsonl)" +printf '%s\n' "$BATCH_ID" +``` + +客户端先向 `/api/v1/cloud/files` 上传文件，并显式使用 `purpose=session_resource`，再调用 `POST /api/v1/forward/batches`。`custom_id` 用于把每一行与结果重新对应；一次 Batch 最多支持 10,000 行，全局并发上限为 50。[Create Batch API](https://docs.qoder.com/cloud-agents/api/forward/batches/create) + +> 现场与线上读者的执行时间不同：QCA Batch 当前默认只在服务端配置的离峰窗口 `22:00–08:00` 执行。线下 Workshop 可以在主办方明确确认临时开放即时执行后按课程时间操作；在线读者应预期 Batch 先停留在 `queued`，到下一离峰窗口才开始。`completion_window` 是最长完成期限，不代表立即执行。 + +### 55～70 分钟：观察任务并下载结果 + +轮询 Batch，直到进入 `completed`、`failed`、`cancelled` 或 `expired`： + +```bash +python3 qca_batch.py wait \ + --batch-id "$BATCH_ID" \ + --output-dir .work/results +``` + +等待过程中，客户端打印聚合状态和任务计数；终态后下载 `output.jsonl`，存在失败行时再下载 `error.jsonl`，并通过 List Batch Tasks API 保存 `tasks.json`。预签名下载 URL 含临时访问参数，客户端不会打印它；也不要在启用 `set -x` 的终端执行下载。 + +使用 `custom_id` 检查每个任务： + +```bash +python3 qca_batch.py summarize \ + --output .work/results/output.jsonl \ + --errors .work/results/error.jsonl \ + --tasks-report .work/results/tasks.json +``` + +任务列表中的 `usage.total_credits` 单位是 CAS Credit，不是 token 数或货币金额，应作为逐任务对账依据。Batch 详情也可能返回聚合 `usage.total_credits`，但客户端不应假定它始终存在。下载链接会过期，Batch 结果文件保留 30 天；需要审计的结果应及时保存到自己的受控存储中。[List Batch Tasks API](https://docs.qoder.com/cloud-agents/api/forward/batches/list-tasks) + +如果合法行返回 `session_error`，先查看 `tasks.json` 中的 `custom_id`、错误码、Credit 和 artifacts： + +- `All models failed` 或 `model queue recovery attempts exceeded` 属于执行层失败，不代表迁移测试失败。没有 artifacts 时，可以只重投这些行。 +- 如果测试提示相对目录不存在，检查任务 Prompt 是否保留了 `cd /workspace/cloud-agents-cookbook/demos/batch-sdk-migration-with-qca && ...`；不要靠 Agent 猜测当前工作目录。 +- 已经成功并交付 artifacts 的任务不要重复提交，避免重复消耗 Credit。 + +### 70～82 分钟：只重投失败任务 + +第一批的 Billing 行因为缺少 `identity_id` 应出现在 `error.jsonl`，其它合法行继续执行。根据失败行的 `custom_id`，从原任务清单重建有效 JSONL： + +```bash +python3 qca_batch.py retry \ + --tasks tasks.json \ + --errors .work/results/error.jsonl \ + --output .work/retry-input.jsonl +``` + +再次提交： + +```bash +export RETRY_BATCH_ID="$(python3 qca_batch.py submit \ + --input .work/retry-input.jsonl)" +python3 qca_batch.py wait \ + --batch-id "$RETRY_BATCH_ID" \ + --output-dir .work/retry-results +``` + +重投是一个新 Batch，可以复用上一批的 `custom_id`。不要把整批任务全部重跑，否则会重复消耗 Credit，也会模糊首次成功结果与重试结果的对应关系。 + +### 82～90 分钟：用确定性证据验收 + +下载每个完成任务交付的 Patch 和报告，在干净的本地 clone 分支上应用 Patch，再运行任务清单里的验收命令。至少检查： + +- Agent 是否只修改被分配的项目目录。 +- Catalog、Orders、Inventory 的验收测试是否真实通过。 +- 测试文件和 Modern SDK 模拟实现是否保持不变。 +- Billing 是否提交人工复核说明，而不是猜测舍入策略；它的 Patch 是否只新增该说明。 +- 报告是否记录执行过的命令、测试结果和剩余风险。 + +Agent 的回复、Patch 和报告都只是候选结果；测试通过与人工审查共同构成质量门禁。Workshop 不要求把修改自动推回远端，也不授权 Agent 接触生产仓库。 + +## 练习与总结 + +### 必做练习 + +完成一次包含可控校验错误的四任务 Batch，并只重投失败的 Billing 行。用下面的清单验收： + +- [ ] 已创建练习专用 Template 和 Identity，并记录自己的资源 ID。 +- [ ] Environment 通过 `setup_script` 准备公开示例仓库，Template 不绑定 GitHub Repository，也不包含生产凭据。 +- [ ] JSONL 每行有唯一 `custom_id`，PAT 未进入文件。 +- [ ] 第一批的合法行继续执行，缺字段行单独进入 `error.jsonl`。 +- [ ] 第二批只包含首次失败任务。 +- [ ] 三个代码迁移任务均已检查 Patch、修改范围和测试结果。 +- [ ] Billing 的业务歧义进入人工复核，没有被 Agent 静默猜测。 +- [ ] 能从 List Batch Tasks 读出单任务 Credit，并理解 Batch 聚合 `usage` 可能存在但不保证返回，以及输出文件的 30 天保留边界。 + +### 延伸练习 + +为第五个服务增加一项迁移任务，例如分页接口、流式返回或配置初始化方式变化。为它准备独立项目目录和验收测试，在 `tasks.json` 中填写唯一 `custom_id`、可写目录、验收命令和预期结果，再单独提交该任务。完成后检查它是否按约定交付 Patch、报告和测试证据。 + +完成后删除本地 `.work/`，归档需要保留的 Patch 和报告，并在控制台停用或删除本练习专用 Identity、Template、Environment 和凭据。清理动作和实际 Credit 以你的账号控制台为准。 + +## Demo 源码 + +Demo 提供四个合成迁移项目、Template 系统提示词、任务清单、纯 Python 标准库 Batch 客户端和 mock 单元测试。完整的运行、验证、失败重投、清理和成本说明见 README。 + +[查看 Demo 源码](https://github.com/QoderAI/cloud-agents-cookbook/tree/main/demos/batch-sdk-migration-with-qca) diff --git a/demos/batch-sdk-migration-with-qca/.env.example b/demos/batch-sdk-migration-with-qca/.env.example new file mode 100644 index 0000000..616649b --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/.env.example @@ -0,0 +1,4 @@ +QODER_PAT=replace_with_your_qoder_pat +QODER_TEMPLATE_ID=tmpl_replace_me +QODER_IDENTITY_ID=idn_replace_me +QODER_API_BASE=https://api.qoder.com diff --git a/demos/batch-sdk-migration-with-qca/.gitignore b/demos/batch-sdk-migration-with-qca/.gitignore new file mode 100644 index 0000000..ddd934a --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/.gitignore @@ -0,0 +1,5 @@ +/.work/ +/workshop.env +!/.env.example +/**/__pycache__/ +*.py[cod] diff --git a/demos/batch-sdk-migration-with-qca/FORWARD_ENVIRONMENT_SETUP.sh b/demos/batch-sdk-migration-with-qca/FORWARD_ENVIRONMENT_SETUP.sh new file mode 100644 index 0000000..4004912 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/FORWARD_ENVIRONMENT_SETUP.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +set -euo pipefail + +readonly cookbook_dir="/workspace/cloud-agents-cookbook" +readonly demo_dir="${cookbook_dir}/demos/batch-sdk-migration-with-qca" + +if [[ -e "${cookbook_dir}" ]]; then + printf 'setup failed: target already exists: %s\n' "${cookbook_dir}" >&2 + exit 1 +fi + +git clone --depth 1 \ + https://github.com/QoderAI/cloud-agents-cookbook.git \ + "${cookbook_dir}" + +if [[ ! -f "${demo_dir}/tasks.json" ]]; then + printf 'setup failed: workshop demo is missing: %s\n' "${demo_dir}" >&2 + exit 1 +fi + +printf 'workshop workspace ready: %s\n' "${demo_dir}" diff --git a/demos/batch-sdk-migration-with-qca/MIGRATION_GUIDE.md b/demos/batch-sdk-migration-with-qca/MIGRATION_GUIDE.md new file mode 100644 index 0000000..e3756d7 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/MIGRATION_GUIDE.md @@ -0,0 +1,35 @@ +# Legacy Shop SDK to Modern Shop SDK migration guide + +This guide describes the only supported migration rules for the synthetic Workshop projects. + +## Catalog API + +- Replace `legacy_sdk.Client().fetch(sku)` with `modern_sdk.Client(region="CN").get_product(product_id=sku)`. +- The new call returns a `Product` object. Read `product.sku`, `product.price.currency`, and `product.price.amount`. +- Preserve the public `display_price(sku)` function. Its result format is `: `. + +## Orders API + +- Replace `LegacyOrderClient.submit({"items": items})` with `ModernOrderClient().create_order(line_items=items, idempotency_key=request_id)`. +- The new result is an object. Use `result.order_id`, not dictionary indexing. +- Replace `LegacyOrderError` with `RequestError`. Map `RequestError.code` to the existing public error string `rejected:`. +- Preserve the `OrderGateway.place(items, request_id)` interface and forward `request_id` as the idempotency key. + +## Inventory API + +- Replace synchronous `Client.stock(sku)` with `await Client().get_inventory(sku=sku)`. +- The new result is an `InventoryResult`; use its `available` field. +- Change the public `available(sku)` function to `async def available(sku)` rather than creating an event loop inside the function. + +## Billing policy + +- Modern Billing requires an explicit `rounding_mode` of `half_even` or `half_up`. +- The supplied material does not say which mode the business has approved. +- Do not infer a mode from the Legacy implementation. Create `projects/billing/manual-review.md` with `status: blocked`, `decision: rounding_mode`, the two permitted choices, and the evidence that the policy is absent. +- Do not modify Billing source until a business owner chooses the policy. + +## Files that must not change + +- Any `test_*.py` file. +- Any `modern_sdk.py` file. +- Any project outside the task's assigned path. diff --git a/demos/batch-sdk-migration-with-qca/README.md b/demos/batch-sdk-migration-with-qca/README.md new file mode 100644 index 0000000..ac77e65 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/README.md @@ -0,0 +1,172 @@ +# QCA Batch Agent SDK 迁移 Workshop Demo + +这个 Demo 用四个合成 Python 服务演示 QCA Batch 的 Agent 执行闭环：每个任务读取同一份 SDK 迁移指南，但必须根据各自项目结构修改代码、运行测试、分析失败并继续修复。Catalog、Orders、Inventory 的目标是通过确定性测试；Billing 缺少业务舍入策略，目标是生成合格的人工复核材料而不是猜测。 + +Demo 的 Batch 客户端只使用 Python 标准库。本地单元测试用 mock HTTP 服务验证协议，不调用 QCA；真实 Batch 才会创建云端 Session 并消耗 Credit。 + +## 对应文章 + +- 标题：批量升级异构代码项目：用 QCA Batch Agent 完成迁移、测试与修复 +- Slug：`batch-sdk-migration-with-qca` +- 正文路径：`content/zh-CN/workshops/batch-sdk-migration-with-qca/index.md` + +## 前置条件 + +- Python 3.11 或更高版本。 +- Git，用于克隆公开的 [QoderAI/cloud-agents-cookbook](https://github.com/QoderAI/cloud-agents-cookbook) 仓库。 +- QCA 账号、Personal Access Token，以及创建 Environment、Forward Template、Identity 和 Batch 的权限。 +- 现场练习若要在 90 分钟内完成，主办方必须确认服务端已临时开放即时执行；公开环境默认在 `22:00–08:00` 离峰窗口执行。 + +本 Demo 不需要数据库、容器或第三方 Python 包。 + +## 安装与配置 + +克隆公开仓库并进入 Demo： + +```bash +git clone https://github.com/QoderAI/cloud-agents-cookbook.git +cd cloud-agents-cookbook/demos/batch-sdk-migration-with-qca +``` + +先运行完全离线的本地检查： + +```bash +PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests -v +python3 check_baseline.py +``` + +`check_baseline.py` 成功表示四个项目都处于设计好的“尚未迁移”状态；它不会修改项目。 + +在 QCA 控制台创建隔离的 Forward Environment： + +1. Packages 中添加 `git`。 +2. 把 `FORWARD_ENVIRONMENT_SETUP.sh` 的完整内容复制到 `setup_script`。 +3. 等待 Environment 创建成功。脚本会把公开仓库克隆到 `/workspace/cloud-agents-cookbook`，并检查当前 Demo 是否存在。 + +`setup_script` 在 Packages 安装之后通过 `/bin/bash -lc` 执行；脚本非零退出时 Environment 创建会失败。不要把 PAT 或其它凭据写进脚本。 + +Environment 就绪后创建 Forward Template： + +| 配置 | 取值 | +|---|---| +| Environment | 刚创建的练习专用 Forward Environment | +| Model | `performance`；创建前通过 List Models API 确认账号当前可用 | +| System Prompt | `TEMPLATE_SYSTEM_PROMPT.md` 的完整内容 | +| Built-in tools | `Bash`、`Read`、`Write`、`Edit`、`Glob`、`Grep`、`DeliverArtifacts` | +| Permission policy | 仅在本隔离 Demo 中为上述工具设置 `always_allow` | + +Forward Mode 在本练习中不挂载 GitHub Repository，也不需要 GitHub token。Batch 无人值守执行不接受 `always_ask` 或 `always_deny` 工具策略。不要把这个练习 Template 用于生产仓库。再创建一个练习专用 Identity，复制 Template ID 和 Identity ID。 + +可以复制占位文件帮助检查变量名，但只在未跟踪的本地文件中填值： + +```bash +cp .env.example workshop.env +``` + +编辑 `workshop.env` 后加载变量： + +```bash +set -a +. ./workshop.env +set +a +``` + +`workshop.env` 和 `.work/` 已由本目录的 `.gitignore` 排除，仍应在使用后主动删除。PAT 不得进入 Batch JSONL 或 Prompt。 + +## 运行 + +生成四行 Batch JSONL，并故意让 Billing 行缺少 `identity_id`： + +```bash +python3 qca_batch.py prepare \ + --tasks tasks.json \ + --output .work/batch-input.jsonl \ + --inject-invalid migrate-billing + +python3 qca_batch.py inspect --input .work/batch-input.jsonl +``` + +`inspect` 只打印任务 ID、Template ID、Identity 是否存在和输入长度，不打印完整 Prompt 或任何 token。 + +上传 JSONL 并创建 Batch： + +```bash +export BATCH_ID="$(python3 qca_batch.py submit \ + --input .work/batch-input.jsonl)" +printf '%s\n' "$BATCH_ID" +``` + +等待终态并下载结果： + +```bash +python3 qca_batch.py wait \ + --batch-id "$BATCH_ID" \ + --output-dir .work/results +``` + +如果 Batch 处于 `queued`，可以按 `Ctrl-C` 停止本地轮询；Batch 不会因此取消。进入离峰窗口后重新运行同一条 `wait` 命令即可。 + +汇总成功行、失败行、Credit 和交付物名称： + +```bash +python3 qca_batch.py summarize \ + --output .work/results/output.jsonl \ + --errors .work/results/error.jsonl \ + --tasks-report .work/results/tasks.json +``` + +`wait` 会额外调用 `GET /api/v1/forward/batches/{batch_id}/tasks` 并保存 `tasks.json`。任务级 `usage.total_credits`、错误摘要和 artifact 元数据以这个接口为准；Batch 详情也可能返回聚合 Credit，但客户端不应假定 `usage` 始终存在。 + +合法行若返回 `session_error`，先按 `custom_id` 查看任务记录。`All models failed`、`model queue recovery attempts exceeded` 等执行层错误不等于测试失败；没有 artifacts 时只重投受影响行。若测试报告 `projects/` 不存在，确认 JSONL Prompt 中的远程验收命令以 `cd /workspace/cloud-agents-cookbook/demos/batch-sdk-migration-with-qca &&` 开头。 + +根据 `error.jsonl` 只重建失败任务。被故意破坏的 Billing 行会恢复当前 Template 和 Identity： + +```bash +python3 qca_batch.py retry \ + --tasks tasks.json \ + --errors .work/results/error.jsonl \ + --output .work/retry-input.jsonl + +export RETRY_BATCH_ID="$(python3 qca_batch.py submit \ + --input .work/retry-input.jsonl)" + +python3 qca_batch.py wait \ + --batch-id "$RETRY_BATCH_ID" \ + --output-dir .work/retry-results +``` + +## 验证结果 + +本地测试应全部通过，`check_baseline.py` 应输出四行 `ready`。真实 Batch 的预期结果如下： + +| `custom_id` | 首批预期 | Agent 任务的验收证据 | +|---|---|---| +| `migrate-catalog` | 执行 | Catalog 测试通过；Patch 使用 Modern Product/Money 对象 | +| `migrate-orders` | 执行 | Orders 测试通过；Adapter 保留且转发幂等键、映射新异常 | +| `migrate-inventory` | 执行 | Inventory 异步测试通过；公开函数变为协程 | +| `migrate-billing` | JSONL 校验失败 | 重投后交付 `manual-review.md` 和只新增该文件的 `changes.patch`，源码保持不变 | + +对前三项，下载并人工检查 `changes.patch` 与 `migration-report.md`，在干净分支应用 Patch 后运行 `tasks.json` 中对应的验收命令。对 Billing，`manual-review.md` 必须包含 `status: blocked`、`decision: rounding_mode`、两个允许选项以及政策缺失证据。 + +不要把 Agent 的 `completed` 状态等同于代码正确。只有测试结果、修改范围检查和人工审查共同通过，迁移才算验收完成。 + +## 清理资源 + +保留需要审计的 Patch 和报告后，删除本地运行文件： + +```bash +rm -rf .work +rm -f workshop.env +``` + +在 QCA 控制台停用或删除练习专用 Identity、Template 和 Environment。不要删除仍被其它任务使用的共享资源。 + +## 成本与安全 + +- 本地单元测试和基线检查不访问网络、不创建云资源、不消耗 Credit。 +- 每个合法 Batch 行会创建独立 Agent Session 并产生 Credit。具体用量取决于模型、代码阅读、工具调用和修复轮次；以 List Batch Tasks 返回的任务级 `usage.total_credits` 为准，不把 Credit 换算成未经官方公布的货币价格。 +- QCA Batch 当前默认在服务端配置的 `22:00–08:00` 离峰窗口执行。现场即时执行是临时安排，在线读者不能据此假定自己的 Batch 会立即启动。 +- `QODER_PAT` 只用于 QCA API，不得进入 `setup_script`、JSONL、Prompt、日志、Patch、报告或提交记录。 +- `always_allow` 仅适用于这个无生产数据、无生产凭据、只包含公开示例仓库的隔离环境。生产接入必须重新评估最小权限、网络访问、审计和人工审批。 +- Batch 结果下载 URL 带临时访问参数。客户端不会打印 URL；不要在 `set -x` 下运行。结果文件保留 30 天，需要长期保存时应转移到自己的受控存储。 +- 仓库自动化只静态检查 Demo，不安装或运行源码。本目录的真实可运行性、产品事实和安全操作仍需 Maintainer 人工审核。 diff --git a/demos/batch-sdk-migration-with-qca/TEMPLATE_SYSTEM_PROMPT.md b/demos/batch-sdk-migration-with-qca/TEMPLATE_SYSTEM_PROMPT.md new file mode 100644 index 0000000..cf9ad37 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/TEMPLATE_SYSTEM_PROMPT.md @@ -0,0 +1,13 @@ +# SDK Migration Agent system prompt + +You migrate exactly one synthetic project per task in the Cookbook workspace prepared by the Forward Environment. + +Follow these rules: + +1. Read `MIGRATION_GUIDE.md`, the assigned project, and its tests before editing. +2. Treat the task's project path as the only writable source scope. Never edit tests, `modern_sdk.py`, another project, or repository configuration. Requested artifact files may be created at the paths specified by the task. +3. Use the exact acceptance command from the task. A written claim is not evidence that migration succeeded. +4. For an automated task, iterate until the acceptance command passes or a concrete blocker remains. Write `migration-report.md` inside the assigned project, then create the requested repository-relative `changes.patch` with `git diff --binary`. The report must list files changed, commands run, results, and remaining risks. +5. For a manual-review task, do not invent a missing business policy. Leave existing source code unchanged, create `manual-review.md` inside the assigned project with the missing decision and evidence, run its review checker, and create `changes.patch` with `git diff --binary`. The patch must add only `manual-review.md`; deliver both files. +6. Do not push commits, call external services, read credentials, or change permissions. Do not expose environment variables in output. +7. Use `DeliverArtifacts` only for the requested patch and report files. diff --git a/demos/batch-sdk-migration-with-qca/check_baseline.py b/demos/batch-sdk-migration-with-qca/check_baseline.py new file mode 100644 index 0000000..9dc16c2 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/check_baseline.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 + +"""Verify that every synthetic task starts in its intended unresolved state.""" + +from __future__ import annotations + +import json +import os +import shlex +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parent + + +def check_baseline() -> list[dict]: + tasks = json.loads((ROOT / "tasks.json").read_text(encoding="utf-8"))["tasks"] + env = dict(os.environ) + env["PYTHONDONTWRITEBYTECODE"] = "1" + results = [] + for task in tasks: + process = subprocess.run( + shlex.split(task["acceptance_command"]), + cwd=ROOT, + env=env, + capture_output=True, + text=True, + check=False, + ) + output = process.stdout + process.stderr + expected = task["baseline_error"] + if process.returncode == 0: + raise RuntimeError(f"{task['custom_id']} unexpectedly passes before migration") + if expected not in output: + raise RuntimeError(f"{task['custom_id']} failed without expected evidence {expected!r}") + results.append({"custom_id": task["custom_id"], "state": "ready", "evidence": expected}) + return results + + +def main() -> int: + try: + for result in check_baseline(): + print(f"{result['custom_id']}: {result['state']} ({result['evidence']})") + return 0 + except RuntimeError as exc: + print(f"baseline contract failed: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/demos/batch-sdk-migration-with-qca/projects/billing/app.py b/demos/batch-sdk-migration-with-qca/projects/billing/app.py new file mode 100644 index 0000000..1033a8d --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/billing/app.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 + +from legacy_sdk import calculate_total + + +def invoice_total(amount): + return calculate_total(amount) diff --git a/demos/batch-sdk-migration-with-qca/projects/billing/check_manual_review.py b/demos/batch-sdk-migration-with-qca/projects/billing/check_manual_review.py new file mode 100644 index 0000000..983945a --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/billing/check_manual_review.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 + +from pathlib import Path + + +review = Path(__file__).with_name("manual-review.md") +if not review.exists(): + raise SystemExit("manual-review.md is missing") + +text = review.read_text(encoding="utf-8") +required = ("status: blocked", "decision: rounding_mode", "half_even", "half_up") +missing = [value for value in required if value not in text] +if missing: + raise SystemExit("manual-review.md is missing required evidence: " + ", ".join(missing)) + +print("manual review package is complete") diff --git a/demos/batch-sdk-migration-with-qca/projects/billing/legacy_sdk.py b/demos/batch-sdk-migration-with-qca/projects/billing/legacy_sdk.py new file mode 100644 index 0000000..1249487 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/billing/legacy_sdk.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 + +from decimal import Decimal, ROUND_HALF_UP + + +def calculate_total(amount): + return Decimal(amount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) diff --git a/demos/batch-sdk-migration-with-qca/projects/billing/modern_sdk.py b/demos/batch-sdk-migration-with-qca/projects/billing/modern_sdk.py new file mode 100644 index 0000000..2c6585a --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/billing/modern_sdk.py @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: Apache-2.0 + +from decimal import Decimal, ROUND_HALF_EVEN, ROUND_HALF_UP + + +class Client: + def __init__(self, *, rounding_mode): + if rounding_mode not in {"half_even", "half_up"}: + raise ValueError("rounding_mode must be half_even or half_up") + self.rounding_mode = rounding_mode + + def calculate_total(self, amount): + rounding = ROUND_HALF_EVEN if self.rounding_mode == "half_even" else ROUND_HALF_UP + return Decimal(amount).quantize(Decimal("0.01"), rounding=rounding) diff --git a/demos/batch-sdk-migration-with-qca/projects/catalog/app.py b/demos/batch-sdk-migration-with-qca/projects/catalog/app.py new file mode 100644 index 0000000..6df0f18 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/catalog/app.py @@ -0,0 +1,8 @@ +# SPDX-License-Identifier: Apache-2.0 + +from legacy_sdk import Client + + +def display_price(sku): + product = Client().fetch(sku) + return f"{product['sku']}:{product['price_cents'] / 100:.2f}" diff --git a/demos/batch-sdk-migration-with-qca/projects/catalog/legacy_sdk.py b/demos/batch-sdk-migration-with-qca/projects/catalog/legacy_sdk.py new file mode 100644 index 0000000..9e914b1 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/catalog/legacy_sdk.py @@ -0,0 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 + + +class Client: + def fetch(self, sku): + return {"sku": sku, "price_cents": 2599} diff --git a/demos/batch-sdk-migration-with-qca/projects/catalog/modern_sdk.py b/demos/batch-sdk-migration-with-qca/projects/catalog/modern_sdk.py new file mode 100644 index 0000000..9dc22cb --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/catalog/modern_sdk.py @@ -0,0 +1,25 @@ +# SPDX-License-Identifier: Apache-2.0 + +from dataclasses import dataclass +from decimal import Decimal + + +@dataclass(frozen=True) +class Money: + currency: str + amount: Decimal + + +@dataclass(frozen=True) +class Product: + sku: str + price: Money + + +class Client: + def __init__(self, *, region): + if region != "CN": + raise ValueError("Workshop fixture supports only the CN region") + + def get_product(self, *, product_id): + return Product(product_id, Money("CNY", Decimal("25.99"))) diff --git a/demos/batch-sdk-migration-with-qca/projects/catalog/test_app.py b/demos/batch-sdk-migration-with-qca/projects/catalog/test_app.py new file mode 100644 index 0000000..9fa6401 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/catalog/test_app.py @@ -0,0 +1,17 @@ +# SPDX-License-Identifier: Apache-2.0 + +import inspect +import unittest + +import app + + +class CatalogMigrationTest(unittest.TestCase): + def test_uses_modern_sdk_and_formats_money(self): + self.assertIn("modern_sdk", inspect.getsource(app)) + self.assertNotIn("legacy_sdk", inspect.getsource(app)) + self.assertEqual(app.display_price("SKU-42"), "SKU-42:CNY 25.99") + + +if __name__ == "__main__": + unittest.main() diff --git a/demos/batch-sdk-migration-with-qca/projects/inventory/app.py b/demos/batch-sdk-migration-with-qca/projects/inventory/app.py new file mode 100644 index 0000000..8bc4a9c --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/inventory/app.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 + +from legacy_sdk import Client + + +def available(sku): + return Client().stock(sku) diff --git a/demos/batch-sdk-migration-with-qca/projects/inventory/legacy_sdk.py b/demos/batch-sdk-migration-with-qca/projects/inventory/legacy_sdk.py new file mode 100644 index 0000000..08aca1f --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/inventory/legacy_sdk.py @@ -0,0 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 + + +class Client: + def stock(self, sku): + return 7 if sku else 0 diff --git a/demos/batch-sdk-migration-with-qca/projects/inventory/modern_sdk.py b/demos/batch-sdk-migration-with-qca/projects/inventory/modern_sdk.py new file mode 100644 index 0000000..5e5f5fe --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/inventory/modern_sdk.py @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: Apache-2.0 + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class InventoryResult: + sku: str + available: int + + +class Client: + async def get_inventory(self, *, sku): + return InventoryResult(sku=sku, available=11 if sku else 0) diff --git a/demos/batch-sdk-migration-with-qca/projects/inventory/test_app.py b/demos/batch-sdk-migration-with-qca/projects/inventory/test_app.py new file mode 100644 index 0000000..24bcc3e --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/inventory/test_app.py @@ -0,0 +1,17 @@ +# SPDX-License-Identifier: Apache-2.0 + +import inspect +import unittest + +import app + + +class InventoryMigrationTest(unittest.IsolatedAsyncioTestCase): + async def test_public_api_becomes_async_and_uses_result_object(self): + self.assertIn("modern_sdk", inspect.getsource(app)) + self.assertTrue(inspect.iscoroutinefunction(app.available)) + self.assertEqual(await app.available("SKU-42"), 11) + + +if __name__ == "__main__": + unittest.main() diff --git a/demos/batch-sdk-migration-with-qca/projects/orders/app.py b/demos/batch-sdk-migration-with-qca/projects/orders/app.py new file mode 100644 index 0000000..1c6a4a3 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/orders/app.py @@ -0,0 +1,15 @@ +# SPDX-License-Identifier: Apache-2.0 + +from legacy_sdk import LegacyOrderClient, LegacyOrderError + + +class OrderGateway: + def __init__(self): + self.client = LegacyOrderClient() + + def place(self, items, request_id): + try: + result = self.client.submit({"items": items}) + except LegacyOrderError: + return "rejected:legacy_error" + return f"{result['id']}:{request_id}" diff --git a/demos/batch-sdk-migration-with-qca/projects/orders/legacy_sdk.py b/demos/batch-sdk-migration-with-qca/projects/orders/legacy_sdk.py new file mode 100644 index 0000000..bf5b7eb --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/orders/legacy_sdk.py @@ -0,0 +1,12 @@ +# SPDX-License-Identifier: Apache-2.0 + + +class LegacyOrderError(Exception): + pass + + +class LegacyOrderClient: + def submit(self, payload): + if not payload.get("items"): + raise LegacyOrderError("empty order") + return {"id": "ord-legacy", "status": "created"} diff --git a/demos/batch-sdk-migration-with-qca/projects/orders/modern_sdk.py b/demos/batch-sdk-migration-with-qca/projects/orders/modern_sdk.py new file mode 100644 index 0000000..1fa1bc1 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/orders/modern_sdk.py @@ -0,0 +1,23 @@ +# SPDX-License-Identifier: Apache-2.0 + +from dataclasses import dataclass + + +class RequestError(Exception): + def __init__(self, code): + super().__init__(code) + self.code = code + + +@dataclass(frozen=True) +class OrderResult: + order_id: str + status: str + idempotency_key: str + + +class ModernOrderClient: + def create_order(self, *, line_items, idempotency_key): + if not line_items: + raise RequestError("empty_order") + return OrderResult("ord-modern", "created", idempotency_key) diff --git a/demos/batch-sdk-migration-with-qca/projects/orders/test_app.py b/demos/batch-sdk-migration-with-qca/projects/orders/test_app.py new file mode 100644 index 0000000..8190047 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/projects/orders/test_app.py @@ -0,0 +1,21 @@ +# SPDX-License-Identifier: Apache-2.0 + +import inspect +import unittest + +import app + + +class OrdersMigrationTest(unittest.TestCase): + def test_adapter_maps_result_and_idempotency_key(self): + self.assertIn("modern_sdk", inspect.getsource(app)) + gateway = app.OrderGateway() + self.assertEqual(gateway.place([{"sku": "SKU-42", "quantity": 1}], "request-7"), "ord-modern:request-7") + + def test_adapter_maps_new_error_code(self): + gateway = app.OrderGateway() + self.assertEqual(gateway.place([], "request-8"), "rejected:empty_order") + + +if __name__ == "__main__": + unittest.main() diff --git a/demos/batch-sdk-migration-with-qca/qca_batch.py b/demos/batch-sdk-migration-with-qca/qca_batch.py new file mode 100644 index 0000000..f7b059f --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/qca_batch.py @@ -0,0 +1,457 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 + +"""Minimal QCA Forward Batch client used by the Workshop. + +The module intentionally uses only the Python standard library. It never logs +authorization headers or pre-signed result URLs. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +import uuid +from pathlib import Path +from typing import Callable, Iterable +from urllib import error, request + + +DEFAULT_BASE_URL = "https://api.qoder.com" +TERMINAL_STATUSES = {"completed", "failed", "cancelled", "expired"} + + +class QoderApiError(RuntimeError): + """A sanitized API error that does not expose credentials or signed URLs.""" + + +def read_jsonl(path: Path) -> list[dict]: + rows = [] + for number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + if not raw_line.strip(): + continue + try: + value = json.loads(raw_line) + except json.JSONDecodeError as exc: + raise ValueError(f"{path}: line {number} is not valid JSON: {exc.msg}") from exc + if not isinstance(value, dict): + raise ValueError(f"{path}: line {number} must contain a JSON object") + rows.append(value) + return rows + + +def write_jsonl(path: Path, rows: Iterable[dict]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + content = "".join(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n" for row in rows) + path.write_text(content, encoding="utf-8") + + +def load_tasks(path: Path) -> list[dict]: + data = json.loads(path.read_text(encoding="utf-8")) + tasks = data.get("tasks") if isinstance(data, dict) else None + if not isinstance(tasks, list) or not tasks: + raise ValueError(f"{path}: expected a non-empty 'tasks' array") + required = {"custom_id", "project_path", "acceptance_command", "outcome"} + seen = set() + for index, task in enumerate(tasks, start=1): + if not isinstance(task, dict) or not required.issubset(task): + missing = sorted(required - set(task) if isinstance(task, dict) else required) + raise ValueError(f"{path}: task {index} is missing {', '.join(missing)}") + custom_id = task["custom_id"] + if custom_id in seen: + raise ValueError(f"{path}: duplicate custom_id {custom_id!r}") + seen.add(custom_id) + return tasks + + +def task_prompt(task: dict) -> str: + target = task["project_path"] + command = task["acceptance_command"] + workspace = "/workspace/cloud-agents-cookbook/demos/batch-sdk-migration-with-qca" + remote_command = f"cd {workspace} && {command}" + if task["outcome"] == "manual-review": + outcome = ( + "This task is expected to require a business decision. Do not guess. " + f"If MIGRATION_GUIDE.md does not define the required policy, create {target}/manual-review.md " + f"with the missing decision, evidence, and a safe next step. Run `{remote_command}`, " + f"then run `git diff --binary -- {target} > changes.patch` and deliver changes.patch plus " + f"{target}/manual-review.md with DeliverArtifacts. The patch must add only manual-review.md." + ) + else: + outcome = ( + f"Complete the migration and run `{remote_command}` until it passes. Then write " + f"{target}/migration-report.md, run `git diff --binary -- {target} > changes.patch`, " + f"and deliver changes.patch plus {target}/migration-report.md with DeliverArtifacts." + ) + return "\n".join( + [ + f"Before every relative Bash command, change directory to {workspace}.", + "Read TEMPLATE_SYSTEM_PROMPT.md and MIGRATION_GUIDE.md before changing code.", + f"Your only writable project scope is {target}.", + f"Local acceptance command: {command}", + outcome, + ] + ) + + +def build_batch_lines( + tasks: list[dict], + template_id: str, + identity_id: str, + inject_invalid: str | None = None, +) -> list[dict]: + if inject_invalid and inject_invalid not in {task["custom_id"] for task in tasks}: + raise ValueError(f"unknown custom_id for --inject-invalid: {inject_invalid}") + lines = [] + for task in tasks: + line = { + "custom_id": task["custom_id"], + "template_id": template_id, + "identity_id": identity_id, + "body": {"input": task_prompt(task)}, + } + if task["custom_id"] == inject_invalid: + del line["identity_id"] + lines.append(line) + return lines + + +def build_retry_lines( + tasks: list[dict], + error_rows: list[dict], + template_id: str, + identity_id: str, +) -> list[dict]: + failed_ids = { + row.get("custom_id") + for row in error_rows + if row.get("status") == "failed" and isinstance(row.get("custom_id"), str) + } + task_ids = {task["custom_id"] for task in tasks} + unknown = sorted(failed_ids - task_ids) + if unknown: + raise ValueError(f"error file contains custom_id values absent from tasks: {', '.join(unknown)}") + selected = [task for task in tasks if task["custom_id"] in failed_ids] + if not selected: + raise ValueError("error file contains no failed tasks to retry") + return build_batch_lines(selected, template_id, identity_id) + + +def merge_result_rows(output_rows: list[dict], error_rows: list[dict]) -> list[dict]: + """Return one row per custom_id, overlaying errors without losing task usage.""" + order = [] + by_custom_id = {} + for row in [*output_rows, *error_rows]: + custom_id = row.get("custom_id") + if not isinstance(custom_id, str): + continue + if custom_id not in by_custom_id: + order.append(custom_id) + previous = by_custom_id.get(custom_id, {}) + by_custom_id[custom_id] = {**previous, **row} + return [by_custom_id[custom_id] for custom_id in order] + + +class QoderBatchClient: + def __init__(self, token: str, base_url: str = DEFAULT_BASE_URL, timeout: float = 30): + if not token: + raise ValueError("Qoder PAT is required") + self.token = token + self.base_url = base_url.rstrip("/") + self.timeout = timeout + + def _api_headers(self, content_type: str | None = None) -> dict[str, str]: + headers = {"Authorization": f"Bearer {self.token}", "Accept": "application/json"} + if content_type: + headers["Content-Type"] = content_type + return headers + + def _json_request(self, method: str, path: str, payload: dict | None = None) -> dict: + body = None if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8") + req = request.Request( + f"{self.base_url}{path}", + data=body, + method=method, + headers=self._api_headers("application/json" if body is not None else None), + ) + try: + with request.urlopen(req, timeout=self.timeout) as response: + return json.loads(response.read().decode("utf-8")) + except error.HTTPError as exc: + try: + detail = json.loads(exc.read().decode("utf-8")) + message = detail.get("error", {}).get("message") or detail.get("message") + except (json.JSONDecodeError, UnicodeDecodeError): + message = None + suffix = f": {message}" if message else "" + raise QoderApiError(f"QCA API {method} {path} returned HTTP {exc.code}{suffix}") from exc + except error.URLError as exc: + raise QoderApiError(f"QCA API {method} {path} could not be reached") from exc + + def upload_input(self, input_path: Path) -> str: + boundary = f"qca-workshop-{uuid.uuid4().hex}" + file_bytes = input_path.read_bytes() + parts = [ + f"--{boundary}\r\nContent-Disposition: form-data; name=\"purpose\"\r\n\r\nsession_resource\r\n".encode(), + ( + f"--{boundary}\r\n" + f"Content-Disposition: form-data; name=\"file\"; filename=\"{input_path.name}\"\r\n" + "Content-Type: application/jsonl\r\n\r\n" + ).encode(), + file_bytes, + f"\r\n--{boundary}--\r\n".encode(), + ] + req = request.Request( + f"{self.base_url}/api/v1/cloud/files", + data=b"".join(parts), + method="POST", + headers=self._api_headers(f"multipart/form-data; boundary={boundary}"), + ) + try: + with request.urlopen(req, timeout=self.timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + except error.HTTPError as exc: + raise QoderApiError(f"QCA Files upload returned HTTP {exc.code}") from exc + except error.URLError as exc: + raise QoderApiError("QCA Files upload could not be reached") from exc + file_id = payload.get("id") + if not isinstance(file_id, str) or not file_id: + raise QoderApiError("QCA Files upload response did not contain an id") + return file_id + + def create_batch(self, input_file_id: str, completion_window: str = "24h") -> dict: + return self._json_request( + "POST", + "/api/v1/forward/batches", + {"input_file_id": input_file_id, "completion_window": completion_window}, + ) + + def submit(self, input_path: Path, completion_window: str = "24h") -> dict: + return self.create_batch(self.upload_input(input_path), completion_window) + + def get_batch(self, batch_id: str) -> dict: + return self._json_request("GET", f"/api/v1/forward/batches/{batch_id}") + + def list_tasks(self, batch_id: str) -> list[dict]: + tasks = [] + after_id = None + while True: + suffix = "?limit=100" + if after_id: + from urllib.parse import quote + + suffix += f"&after_id={quote(after_id, safe='')}" + page = self._json_request("GET", f"/api/v1/forward/batches/{batch_id}/tasks{suffix}") + rows = page.get("data") + if not isinstance(rows, list): + raise QoderApiError("List Batch Tasks response did not contain a data array") + tasks.extend(row for row in rows if isinstance(row, dict)) + if not page.get("has_more"): + return tasks + after_id = page.get("last_id") + if not isinstance(after_id, str) or not after_id: + raise QoderApiError("List Batch Tasks response omitted last_id while has_more=true") + + def _download_result(self, batch_id: str, kind: str, destination: Path) -> None: + result = self._json_request("GET", f"/api/v1/forward/batches/{batch_id}/{kind}") + signed_url = result.get("url") + if not isinstance(signed_url, str) or not signed_url: + raise QoderApiError(f"Batch {kind} response did not contain a download URL") + try: + with request.urlopen(signed_url, timeout=self.timeout) as response: + content = response.read() + except (error.HTTPError, error.URLError) as exc: + raise QoderApiError(f"Batch {kind} file download failed") from exc + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(content) + + def wait( + self, + batch_id: str, + output_dir: Path, + poll_interval: float = 15, + on_status: Callable[[dict], None] | None = None, + ) -> dict: + while True: + batch = self.get_batch(batch_id) + if on_status: + on_status(batch) + if batch.get("status") in TERMINAL_STATUSES: + break + time.sleep(poll_interval) + if batch.get("output_file_id"): + self._download_result(batch_id, "output", output_dir / "output.jsonl") + if batch.get("error_file_id"): + self._download_result(batch_id, "error", output_dir / "error.jsonl") + tasks = self.list_tasks(batch_id) + (output_dir / "tasks.json").write_text( + json.dumps({"data": tasks}, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + return batch + + +def require_env(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise ValueError(f"environment variable {name} is required") + return value + + +def client_from_env() -> QoderBatchClient: + return QoderBatchClient( + require_env("QODER_PAT"), + base_url=os.environ.get("QODER_API_BASE", DEFAULT_BASE_URL), + ) + + +def status_line(batch: dict) -> str: + counts = batch.get("request_counts") or {} + usage = batch.get("usage") or {} + values = [ + f"status={batch.get('status', 'unknown')}", + f"total={counts.get('total', 0)}", + f"pending={counts.get('pending', 0)}", + f"running={counts.get('running', 0)}", + f"completed={counts.get('completed', 0)}", + f"failed={counts.get('failed', 0)}", + ] + if "total_credits" in usage: + values.append(f"credits={usage['total_credits']}") + return " ".join(values) + + +def command_prepare(args: argparse.Namespace) -> None: + tasks = load_tasks(args.tasks) + lines = build_batch_lines( + tasks, + require_env("QODER_TEMPLATE_ID"), + require_env("QODER_IDENTITY_ID"), + inject_invalid=args.inject_invalid, + ) + write_jsonl(args.output, lines) + print(f"wrote {len(lines)} task(s) to {args.output}", file=sys.stderr) + + +def command_inspect(args: argparse.Namespace) -> None: + rows = read_jsonl(args.input) + for row in rows: + summary = { + "custom_id": row.get("custom_id"), + "template_id": row.get("template_id"), + "has_identity_id": bool(row.get("identity_id")), + "input_characters": len((row.get("body") or {}).get("input", "")), + } + print(json.dumps(summary, ensure_ascii=False)) + + +def command_submit(args: argparse.Namespace) -> None: + result = client_from_env().submit(args.input, completion_window=args.completion_window) + batch_id = result.get("id") + if not isinstance(batch_id, str) or not batch_id: + raise QoderApiError("Create Batch response did not contain an id") + print(batch_id) + + +def command_wait(args: argparse.Namespace) -> None: + final = client_from_env().wait( + args.batch_id, + output_dir=args.output_dir, + poll_interval=args.poll_interval, + on_status=lambda batch: print(status_line(batch), file=sys.stderr), + ) + print(json.dumps({"id": final.get("id"), "status": final.get("status")}, ensure_ascii=False)) + + +def command_retry(args: argparse.Namespace) -> None: + lines = build_retry_lines( + load_tasks(args.tasks), + read_jsonl(args.errors), + require_env("QODER_TEMPLATE_ID"), + require_env("QODER_IDENTITY_ID"), + ) + write_jsonl(args.output, lines) + print(f"wrote {len(lines)} retry task(s) to {args.output}", file=sys.stderr) + + +def command_summarize(args: argparse.Namespace) -> None: + output_rows = read_jsonl(args.output) if args.output.exists() else [] + error_rows = read_jsonl(args.errors) if args.errors and args.errors.exists() else [] + task_rows = [] + if args.tasks_report and args.tasks_report.exists(): + report = json.loads(args.tasks_report.read_text(encoding="utf-8")) + if not isinstance(report, dict) or not isinstance(report.get("data"), list): + raise ValueError(f"{args.tasks_report}: expected an object with a data array") + task_rows = [row for row in report["data"] if isinstance(row, dict)] + rows = merge_result_rows(task_rows or output_rows, error_rows) + for row in rows: + usage = row.get("usage") or {} + error_value = row.get("error") or {} + print( + json.dumps( + { + "custom_id": row.get("custom_id"), + "status": row.get("status"), + "credits": usage.get("total_credits"), + "error_code": error_value.get("code"), + "artifacts": [item.get("name") for item in row.get("artifacts", [])], + }, + ensure_ascii=False, + ) + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="QCA Batch Workshop client") + subparsers = parser.add_subparsers(dest="command", required=True) + + prepare = subparsers.add_parser("prepare", help="build Batch JSONL from tasks.json") + prepare.add_argument("--tasks", type=Path, required=True) + prepare.add_argument("--output", type=Path, required=True) + prepare.add_argument("--inject-invalid") + prepare.set_defaults(func=command_prepare) + + inspect = subparsers.add_parser("inspect", help="print a credential-safe JSONL summary") + inspect.add_argument("--input", type=Path, required=True) + inspect.set_defaults(func=command_inspect) + + submit = subparsers.add_parser("submit", help="upload JSONL and create a Batch") + submit.add_argument("--input", type=Path, required=True) + submit.add_argument("--completion-window", choices=("24h", "48h", "72h"), default="24h") + submit.set_defaults(func=command_submit) + + wait_parser = subparsers.add_parser("wait", help="poll a Batch and download terminal results") + wait_parser.add_argument("--batch-id", required=True) + wait_parser.add_argument("--output-dir", type=Path, required=True) + wait_parser.add_argument("--poll-interval", type=float, default=15) + wait_parser.set_defaults(func=command_wait) + + retry = subparsers.add_parser("retry", help="rebuild valid JSONL for failed custom_id values") + retry.add_argument("--tasks", type=Path, required=True) + retry.add_argument("--errors", type=Path, required=True) + retry.add_argument("--output", type=Path, required=True) + retry.set_defaults(func=command_retry) + + summarize = subparsers.add_parser("summarize", help="summarize output and error JSONL") + summarize.add_argument("--output", type=Path, required=True) + summarize.add_argument("--errors", type=Path) + summarize.add_argument("--tasks-report", type=Path) + summarize.set_defaults(func=command_summarize) + return parser + + +def main(argv: list[str] | None = None) -> int: + try: + args = build_parser().parse_args(argv) + args.func(args) + return 0 + except (OSError, ValueError, QoderApiError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/demos/batch-sdk-migration-with-qca/tasks.json b/demos/batch-sdk-migration-with-qca/tasks.json new file mode 100644 index 0000000..74112ba --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/tasks.json @@ -0,0 +1,32 @@ +{ + "tasks": [ + { + "custom_id": "migrate-catalog", + "project_path": "projects/catalog", + "acceptance_command": "python3 -B -m unittest discover -s projects/catalog -p 'test_*.py' -v", + "outcome": "automated", + "baseline_error": "modern_sdk" + }, + { + "custom_id": "migrate-orders", + "project_path": "projects/orders", + "acceptance_command": "python3 -B -m unittest discover -s projects/orders -p 'test_*.py' -v", + "outcome": "automated", + "baseline_error": "modern_sdk" + }, + { + "custom_id": "migrate-inventory", + "project_path": "projects/inventory", + "acceptance_command": "python3 -B -m unittest discover -s projects/inventory -p 'test_*.py' -v", + "outcome": "automated", + "baseline_error": "modern_sdk" + }, + { + "custom_id": "migrate-billing", + "project_path": "projects/billing", + "acceptance_command": "python3 -B projects/billing/check_manual_review.py", + "outcome": "manual-review", + "baseline_error": "manual-review.md is missing" + } + ] +} diff --git a/demos/batch-sdk-migration-with-qca/tests/test_fixture_contract.py b/demos/batch-sdk-migration-with-qca/tests/test_fixture_contract.py new file mode 100644 index 0000000..f653137 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/tests/test_fixture_contract.py @@ -0,0 +1,34 @@ +# SPDX-License-Identifier: Apache-2.0 + +import json +import unittest +from pathlib import Path + +from check_baseline import ROOT, check_baseline + + +class FixtureContractTests(unittest.TestCase): + def test_tasks_have_isolated_existing_projects_and_commands(self): + tasks = json.loads((ROOT / "tasks.json").read_text(encoding="utf-8"))["tasks"] + self.assertEqual(len(tasks), 4) + self.assertEqual(len({task["custom_id"] for task in tasks}), 4) + self.assertEqual(len({task["project_path"] for task in tasks}), 4) + for task in tasks: + project = ROOT / task["project_path"] + self.assertTrue(project.is_dir(), project) + self.assertIn(task["project_path"], task["acceptance_command"]) + self.assertIn(task["outcome"], {"automated", "manual-review"}) + + def test_every_task_starts_with_the_documented_failure_evidence(self): + results = check_baseline() + self.assertEqual([item["custom_id"] for item in results], [ + "migrate-catalog", + "migrate-orders", + "migrate-inventory", + "migrate-billing", + ]) + self.assertTrue(all(item["state"] == "ready" for item in results)) + + +if __name__ == "__main__": + unittest.main() diff --git a/demos/batch-sdk-migration-with-qca/tests/test_qca_batch.py b/demos/batch-sdk-migration-with-qca/tests/test_qca_batch.py new file mode 100644 index 0000000..fb51888 --- /dev/null +++ b/demos/batch-sdk-migration-with-qca/tests/test_qca_batch.py @@ -0,0 +1,251 @@ +# SPDX-License-Identifier: Apache-2.0 + +import json +import socket +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from qca_batch import ( + QoderBatchClient, + build_batch_lines, + build_retry_lines, + merge_result_rows, + read_jsonl, + write_jsonl, +) + + +TASKS = [ + { + "custom_id": "migrate-catalog", + "project_path": "projects/catalog", + "acceptance_command": "python3 -B -m unittest discover -s projects/catalog -p 'test_*.py' -v", + "outcome": "automated", + }, + { + "custom_id": "migrate-billing", + "project_path": "projects/billing", + "acceptance_command": "python3 -B projects/billing/check_manual_review.py", + "outcome": "manual-review", + }, +] + + +class BatchLineTests(unittest.TestCase): + def test_build_lines_can_inject_one_validation_error(self): + lines = build_batch_lines( + TASKS, + template_id="tmpl_example", + identity_id="idn_example", + inject_invalid="migrate-billing", + ) + + self.assertEqual([line["custom_id"] for line in lines], ["migrate-catalog", "migrate-billing"]) + self.assertEqual(lines[0]["identity_id"], "idn_example") + self.assertNotIn("identity_id", lines[1]) + self.assertIn("projects/catalog", lines[0]["body"]["input"]) + self.assertIn("/workspace/cloud-agents-cookbook", lines[0]["body"]["input"]) + self.assertIn("cd /workspace/cloud-agents-cookbook/demos/batch-sdk-migration-with-qca && python3", lines[0]["body"]["input"]) + self.assertIn("git diff --binary", lines[0]["body"]["input"]) + self.assertIn("projects/billing/manual-review.md", lines[1]["body"]["input"]) + self.assertNotIn("replace_with", json.dumps(lines)) + + def test_retry_rebuilds_only_failed_custom_ids(self): + errors = [ + { + "custom_id": "migrate-billing", + "status": "failed", + "error": {"code": "invalid_request", "message": "identity_id is required"}, + } + ] + + lines = build_retry_lines(TASKS, errors, "tmpl_example", "idn_example") + + self.assertEqual(len(lines), 1) + self.assertEqual(lines[0]["custom_id"], "migrate-billing") + self.assertEqual(lines[0]["identity_id"], "idn_example") + + def test_jsonl_round_trip_preserves_unicode(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "input.jsonl" + write_jsonl(path, [{"custom_id": "任务-1", "body": {"input": "升级服务"}}]) + + self.assertEqual(read_jsonl(path)[0]["body"]["input"], "升级服务") + + def test_result_summary_deduplicates_failed_rows(self): + output = [ + {"custom_id": "migrate-catalog", "status": "completed"}, + { + "custom_id": "migrate-billing", + "status": "failed", + "error": {"code": "generic"}, + "usage": {"total_credits": 1.25}, + }, + ] + errors = [ + {"custom_id": "migrate-billing", "status": "failed", "error": {"code": "invalid_request"}} + ] + + rows = merge_result_rows(output, errors) + + self.assertEqual([row["custom_id"] for row in rows], ["migrate-catalog", "migrate-billing"]) + self.assertEqual(rows[1]["error"]["code"], "invalid_request") + self.assertEqual(rows[1]["usage"]["total_credits"], 1.25) + + +class RecordingHandler(BaseHTTPRequestHandler): + requests = [] + batch_reads = 0 + + def log_message(self, *_args): + return + + def _body(self): + length = int(self.headers.get("Content-Length", "0")) + return self.rfile.read(length) + + def _json(self, status, payload): + body = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self): + body = self._body() + type(self).requests.append(("POST", self.path, self.headers, body)) + if self.path == "/api/v1/cloud/files": + self._json(200, {"id": "file_input001"}) + return + if self.path == "/api/v1/forward/batches": + self._json(200, {"id": "batch_example001", "status": "validating"}) + return + self._json(404, {"error": {"message": "not found"}}) + + def do_GET(self): + type(self).requests.append(("GET", self.path, self.headers, b"")) + if self.path == "/api/v1/forward/batches/batch_example001": + type(self).batch_reads += 1 + if type(self).batch_reads == 1: + self._json( + 200, + { + "id": "batch_example001", + "status": "processing", + "request_counts": {"total": 2, "pending": 0, "running": 2, "completed": 0, "failed": 0, "cancelled": 0, "expired": 0}, + "usage": {"total_credits": 1.25}, + }, + ) + else: + self._json( + 200, + { + "id": "batch_example001", + "status": "completed", + "output_file_id": "file_output001", + "error_file_id": "file_error001", + "request_counts": {"total": 2, "pending": 0, "running": 0, "completed": 1, "failed": 1, "cancelled": 0, "expired": 0}, + "usage": {"total_credits": 2.5}, + }, + ) + return + if self.path in { + "/api/v1/forward/batches/batch_example001/output", + "/api/v1/forward/batches/batch_example001/error", + }: + kind = self.path.rsplit("/", 1)[-1] + self._json(200, {"url": f"{self.server.base_url}/downloads/{kind}", "expires_at": "2030-01-01T00:00:00Z"}) + return + if self.path == "/api/v1/forward/batches/batch_example001/tasks?limit=100": + self._json( + 200, + { + "data": [ + { + "custom_id": "migrate-catalog", + "status": "completed", + "usage": {"total_credits": 2.5}, + "artifacts": [{"file_id": "file_patch", "name": "changes.patch"}], + }, + {"custom_id": "migrate-billing", "status": "failed", "artifacts": []}, + ], + "has_more": False, + "first_id": "migrate-catalog", + "last_id": "migrate-billing", + }, + ) + return + if self.path == "/downloads/output": + body = b'{"custom_id":"migrate-catalog","status":"completed"}\n' + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + if self.path == "/downloads/error": + body = b'{"custom_id":"migrate-billing","status":"failed"}\n' + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + self._json(404, {"error": {"message": "not found"}}) + + +class ClientTests(unittest.TestCase): + def setUp(self): + RecordingHandler.requests = [] + RecordingHandler.batch_reads = 0 + self.server = ThreadingHTTPServer(("", 0), RecordingHandler) + self.server.base_url = f"http://{socket.gethostname()}:{self.server.server_port}" + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=2) + + def test_submit_uploads_session_resource_then_creates_batch(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "input.jsonl" + path.write_text('{"custom_id":"task-1"}\n', encoding="utf-8") + client = QoderBatchClient("placeholder-token", base_url=self.server.base_url) + + response = client.submit(path) + + self.assertEqual(response["id"], "batch_example001") + upload = RecordingHandler.requests[0] + self.assertEqual(upload[:2], ("POST", "/api/v1/cloud/files")) + self.assertIn(b'name="purpose"', upload[3]) + self.assertIn(b"session_resource", upload[3]) + create = RecordingHandler.requests[1] + self.assertEqual(create[:2], ("POST", "/api/v1/forward/batches")) + self.assertEqual(json.loads(create[3]), {"input_file_id": "file_input001", "completion_window": "24h"}) + + def test_wait_downloads_output_and_optional_error_without_exposing_url(self): + statuses = [] + with tempfile.TemporaryDirectory() as directory: + client = QoderBatchClient("placeholder-token", base_url=self.server.base_url) + final = client.wait( + "batch_example001", + output_dir=Path(directory), + poll_interval=0, + on_status=statuses.append, + ) + + self.assertEqual(final["status"], "completed") + self.assertEqual((Path(directory) / "output.jsonl").read_text(encoding="utf-8").strip(), '{"custom_id":"migrate-catalog","status":"completed"}') + self.assertEqual((Path(directory) / "error.jsonl").read_text(encoding="utf-8").strip(), '{"custom_id":"migrate-billing","status":"failed"}') + tasks = json.loads((Path(directory) / "tasks.json").read_text(encoding="utf-8")) + self.assertEqual(tasks["data"][0]["usage"]["total_credits"], 2.5) + + self.assertEqual([item["status"] for item in statuses], ["processing", "completed"]) + + +if __name__ == "__main__": + unittest.main()