unit-test-python-generate-run

name	unit-test-python-generate-run
description	Python 单测生成流水线的生成-执行阶段。手动启动。

unit-test-python-generate-run

本 skill 是 Python 单测流水线的第二阶段。上游（unit-test-gen-init）扫描仓库产出 test_cases.json 基线，记录每个函数的 dimensions / mocks_needed / func_md5 等元数据。本 skill 负责：

1. 初始化调度状态（generate_process.json）
2. 按文件批量派发 Claude sub-agent
3. 每个 sub-agent 独立完成：生成测试 → 执行 → 采集覆盖率 → 失败修复 → 迭代补测
4. 收集结果，生成最终报告

支持 Python（pytest + coverage.py）。

硬规则

1. 基线只读：test_cases.json 是 init 阶段的产物，本 skill 永远不修改它。
2. 所有生成文件必须在 test/ 或 .test/ 下：本 skill（含所有子 agent、临时文件、中间产物）只能在 test/ 和 .test/ 目录下创建/修改文件，不得修改这两个目录以外的任何文件（含 /tmp/）。
3. 尽量少生成空文件：不创建仅含 {}/[]/pass 的占位文件。heartbeat 用 touch 创建空文件属例外。

主 agent 行为边界（严禁违反）

1. 严禁在 sub-agent 产物未校验通过时标 completed — 必须验证三个 shard 文件全部存在（见步骤 6）
2. 严禁在 sub-agent 返回含 rate_limit / 429 / timeout 时立即重试 — 先标 pending，等至少 30 秒再进入下一轮 claim
3. 严禁编写测试代码 — 主 agent 只负责 dispatch / verify / merge，测试代码由 sub-agent 生成
4. 严禁在所有 sub-agent 连续失败时继续 dispatch — 熔断后立即终止，向用户报告
5. 严禁跳过产物校验直接读 sub-agent 返回的 JSON 当结果 — sub-agent 返回的文本仅供参考

职责分工

• 主 agent（Claude）：调度编排、状态管理、结果收集、生成报告。
• 子 agent（Claude sub-agent）：生成测试代码、执行测试、分类失败、修复测试、迭代补测、返回覆盖率结果。
• 脚本：机械性数据处理（执行测试、采集覆盖率、解析失败、合并 cases），不写测试代码也不判断失败原因。

前置条件

• test/generated_unit/test_cases.json 已由 init 阶段产生，如果没有生成则直接终止

关键文件

路径	谁写	说明
test/generated_unit/test_cases.json	init（只读）	函数元数据基线
test/generated_unit/generate_process.json	dispatch.py（claim 子命令原子写）	调度状态：文件级 status + claimed_at + 子 agent 结果
test/generated_unit/test_run_state.json	主 agent（dispatch merge-state 产出）	合并后的 cases 内容 / 状态
test/generated_unit/<src>/test_<name>.py	子 agent	生成的 Python 测试
.test/run_results/<slug>.json	子 agent（runner.py run）	per-file 测试结果 + 覆盖率 shard
.test/heartbeats/<slug>.txt	子 agent（touch）	判活信号，mtime 超过 stale-seconds 未更新 → stale
.test/state_shards/<slug>.json	子 agent（LLM 直接写入）	per-file cases shard
.test/bug_shards/<slug>.json	子 agent（LLM 直接写入）	per-file 源码 bug shard
.test/source_bugs.json	主 agent（dispatch merge-bugs 产出）	合并后的源码疑似 bug
.test/claim_batch/<timestamp>.json	dispatch.py（claim 子命令）	每次 claim 的输出快照，保留所有批次历史

并行正确性的关键约束：.test/run_results、.test/state_shards、.test/bug_shards是每个 sub-agent 自己的 shard 目录，互不冲突。 全局单文件（test_run_state.json、.test/source_bugs.json）只在步骤 8 由主 agent 调用 dispatch.py merge-* 一次性生成。

脚本结构

所有脚本位于 scripts/ 下，按职责分为 3 个脚本：

脚本	子命令	使用者	说明
dispatch.py	init / claim / prepare-shard / verify-artifacts / merge-state / merge-bugs / report	主 agent	调度编排 + 产物校验 + shard 合并 + 报告生成
runner.py	run	子 agent	测试执行 + 覆盖率采集（支持单文件作用域、增量模式）
validate_shard.py	—	子 agent	shard 文件格式验证（state_shard + bug_shard）

标准工作流

执行前说明

本 skill 启动后，将按以下流程为你的仓库自动生成 Python 单元测试：

步骤 1  环境检查（pytest / pytest-cov / pytest-xdist）
步骤 2  初始化调度状态（生成 generate_process.json）
步骤 3  确认覆盖率阈值（可与用户交互修改 generate_process.json）
步骤 4─7 循环调度（每个源文件独立派发 sub-agent）：
         ┌─ claim 待处理文件（AIMD 自适应并发）
         ├─ 派发 sub-agent（每个文件一个独立的 Claude 子 agent）
         │   └─ sub-agent 内部循环：
         │       生成测试代码 → 执行 → 采集覆盖率
         │       → 有失败？ → 硬编码分类 → 修复 → 重跑
         │       → 覆盖率达标？ → 返回结果
         ├─ 校验 sub-agent 产物（三个 shard 文件必须齐全）
         └─ 全部终态？ → 退出循环
步骤 8  合并所有 shard → 生成最终报告

整个过程对源码零侵入：只在 test/generated_unit/ 和 .test/ 目录下创建文件。

最终输出：

产出物	路径	说明
测试代码	test/generated_unit/<src>/test_<name>.py	每个源文件对应的 pytest 测试
覆盖率报告	test/generated_unit/per_file_report.md	按文件的分析报告：覆盖率、失败分类、疑似 bug、未覆盖代码、改进建议
调度状态	test/generated_unit/generate_process.json	每个文件的完成状态（completed / unmet / abandoned）
疑似 bug	.test/source_bugs.json	被 sub-agent 识别的源码疑似 bug 列表
运行状态	test/generated_unit/test_run_state.json	所有 case 的详细状态、断言结果

注意事项：

• 覆盖率阈值可在步骤 3 修改（默认：语句 90%、分支 80%、函数 100%）
• 单个文件最多迭代 5 轮，每轮包含生成→执行→修复循环
• 并发数由 AIMD 算法自动调节，遇到 rate limit 会自动降速
• 可随时中断，下次启动会自动回收未完成的任务

步骤 1：环境探测与安装

在派发子 agent 之前，检查测试环境是否就绪：

python3 -c "import pytest, pytest_cov, coverage, xdist; print('OK')"

如果缺少依赖，按需安装：

pip install pytest pytest-cov coverage pytest-xdist psutil -q

如果 xdist 安装失败，不影响功能，可以继续；如果其他库安装失败，则停止，告知用户原因。

步骤 2：初始化调度状态

如果 generate_process.json 已存在，可以直接进入步骤 3。

python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py init

说明：

1. 生成 generate_process.json，每个源文件一条记录，初始 status 为 "pending"。
2. dispatch.py默认读取 test/generated_unit/test_cases.json，输出到 test/generated_unit/generate_process.json。程序可通过 --baseline / --output 配置。

步骤 3：确认覆盖率阈值与并发数

读取 generate_process.json 的 coverage_config：

cat test/generated_unit/generate_process.json | python3 -c "import json,sys; d=json.load(sys.stdin); print(json.dumps(d.get('coverage_config',{}), indent=2))"

然后使用 AskUserQuestion 工具向用户确认覆盖率和并发数，示例：

AskUserQuestion:
  questions:
    - question: "是否需要修改覆盖率阈值配置？当前值：语句 90%、分支 80%、函数 100%"
      header: "覆盖率阈值"
      options:
        - label: "使用当前值"
          description: "不做修改，直接进入下一步"
        - label: "修改阈值"
          description: "修改 generate_process.json 中的 coverage_config"
      multiSelect: false
    - question: "并发数设置？API 限流推荐 3，不限流推荐 10"
      header: "并发数"
      options:
        - label: "3（限流模式）"
          description: "适合 API 有限流的场景"
        - label: "10（不限流）"
          description: "适合 API 无限流的场景"
        - label: "自定义"
          description: "手动输入并发数数值"
      multiSelect: false

如果用户选择修改覆盖率，用 Edit 工具修改 generate_process.json 的 coverage_config 字段。不修改 test_cases.json。后续所有脚本和子 agent 均从 generate_process.json 读取阈值。

用户选择的并发数用于步骤 4 的 --number 参数。

步骤 4：原子 claim 一批文件

python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py claim --number <步骤3的并发数>

说明：

1. 默认读取 test/generated_unit/generate_process.json。可通过 --process 配置。
2. --number 使用步骤 3 确认的并发数。
3. 输出同时落盘到 .test/claim_batch/ 目录，保留所有批次历史。

这一步同时完成选中 N 个文件并将其标注为 "running"，无需再用 Edit 改 JSON。返回的 JSON：

• batch_size：本轮实际 claim 的文件数
• claimed_at：本轮 claim 的 ISO 时间戳
• claim_batch_path：claim batch 文件的实际路径（步骤 5a 的 --from-claim 必须使用此字段，不要用 claimed_at 拼接）
• files[]：与 batch 同构，每个文件带 paths.{run_result,state_shard,bug_shard,slug} —— 这些 shard 路径必须原样传给子 agent，否则并行时会互相覆盖
• status_counts：各状态计数
• running_count：当前处于 "running" 状态的文件数（含本轮之前已在跑的）
• reclaimed_stale：把超过 --stale-seconds 的 "running" 任务（子 agent 崩溃 / 超时）重新回收到本批
• auto_abandoned：attempt_count >= 3 且产物始终缺失，自动标记为 abandoned
• recommended_concurrency：AIMD 建议的下一轮并发数（1/2/3）
• circuit_break：是否触发熔断（连续 2 轮全 abandoned）
• circuit_break_reason：熔断原因字符串，未熔断时为 null
• all_done：全部进入终态时为 true

AIMD 节流规则

主 agent 不需要手工计算并发数。dispatch.py claim 内置 AIMD 算法自动调整：

条件	并发数
无历史记录（冷启动）	1
最近 3 个完成的 sub-agent 中 ≥2 个 effective_attempt_count > 1	降到 1（MD）
最近 5 个连续 completed 且 effective_attempt_count == 1	恢复到 --max-number（AI）
最近 5 个中有 last_error_category == "rate_limit"	立即降到 1（429 触发 MD）
其他	1

如果 batch_size == 0 且 all_done == true，跳到步骤 8。

步骤 5：派发子 agent

用 dispatch.py prepare-shard 从 claim batch 批量生成 task_envelope，再并发派发 python-test-gen-agent。

5a. 批量生成 task_envelope

从步骤 4 的 claim batch 文件一次性生成所有 sub-agent 所需的 task_envelope：

python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py prepare-shard \
  --from-claim <步骤4返回的claim_batch_path> \
  --blind

说明：

1. --from-claim 读取 claim batch 中的 files[]，从 generate_process.json 自动推导 round 和 shard 路径。
2. 每个 envelope 输出到 .test/task_envelopes/<slug>.json。
3. stdout 输出 JSON，generated[] 数组中每个元素带 source_path 和 envelope_path，供步骤 5b 使用。
4. prepare-shard 会读取现有 shard（run_result、state_shard），收集源码片段（每个函数 ±20 行）。sub-agent 只需读这一个文件即可开始工作。
5. 单文件兜底：prepare-shard --file <source_path> --round <n> --output <path> 仍可用。
6. --blind 启用盲测模式：前两轮 source_snippets 只含签名+docstring，第三轮起子 agent 自动转为阅读源码补测。

5b. 并发派发

在同一条回复里，对 generated[] 中每个 envelope 用 Agent 工具启动 python-test-gen-agent，提示词模板：

读取 <envelope_path>，按其 JSON 内容中定义的任务执行。

所有子 agent 并发启动（run_in_background: false），等待全部完成后进入步骤 6。

5c. 硬性约束

• 每个文件对应一个子 agent，不可合并多个文件到一个 agent
• paths.* 中的 shard 路径必须原样传入（来自 task_envelope），不可修改——并行时子 agent 依赖各自的 shard 隔离
• 某个子 agent 失败不阻塞其他子 agent，继续等待剩余完成
• 不在此步重试失败——stale 回收由步骤 4 的 dispatch.py claim --stale-seconds 在下一轮自动处理
• 子 agent 返回的文本仅供参考，步骤 6 以产物校验为准

步骤 6：收集结果 + 产物校验

产物校验（硬性，不可跳过）：sub-agent 结束后，调用脚本校验产物：

python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py verify-artifacts \
  --process test/generated_unit/generate_process.json \
  --file <source_path> \
  --on-missing pending

脚本会原子检查三个 shard 文件：

1. .test/run_results/<slug>.json
2. .test/state_shards/<slug>.json
3. .test/bug_shards/<slug>.json（空文件也算）

返回 verified == true → 产物齐全，attempt_count 自动归零 返回 verified == false → 产物缺失，status 回退到 --on-missing（pending 或 abandoned），last_error_category 写入 "no_artifact"

校验通过后，对每个文件：

1. 读取 generate_process.json
2. 将结果写入对应文件的 result 字段
3. 根据 result 将 status 改为：
   • "completed"：unmet_reasons == []（达标）或 unmet_reasons 非空但 objective_blocker == true（dead code / 抽象方法 / 不可达分支等客观原因，备注原因）
   • "unmet"：unmet_reasons 非空且 objective_blocker == false——未达标原因是测试能力不足（迭代不够、测试方案问题等）
   • "abandoned"：所有 gap 函数都被 source_bug 阻塞，同时写入 abandon_reason = "all_source_bugs"

步骤 7：循环

• circuit_break == true → 立即停止 dispatch，跳到步骤 8 生成报告并终止。报告需说明限流导致哪些文件 abandoned
• recommended_concurrency == 1 且本轮唯一的 sub-agent 也失败 → 同样跳到步骤 8，不要继续尝试
• 仍有 "pending" 或 stale "running" → 回到步骤 4（下一批，claim 会自动回收 stale + AIMD 调节并发数）
• 仍有 "running" 但未超时 → 等待子 agent 完成
• 全部终态 → 步骤 8

步骤 8：合并 shards → 生成报告

并行 shards 由主 agent 统一合并：

# 8.1 合并所有 per-file state shards
python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py merge-state \
  --shards-dir .test/state_shards \
  --process test/generated_unit/generate_process.json \
  --output test/generated_unit/test_run_state.json

# 8.2 合并所有 per-file bug shards
python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py merge-bugs \
  --shards-dir .test/bug_shards \
  --output .test/source_bugs.json

# 8.3 生成按文件的分析报告（从 .test/run_results 目录聚合覆盖率）
python .claude/skills/unit-test-python-generate-run/scripts/dispatch.py report \
  --process test/generated_unit/generate_process.json \
  --run-state test/generated_unit/test_run_state.json \
  --run-results-dir .test/run_results \
  --source-bugs .test/source_bugs.json \
  --output test/generated_unit/per_file_report.md \
  --format markdown

读取报告文件并呈现给用户。同时从 generate_process.json 汇总：

• 达标文件数（status == "completed"）vs 未达标（"unmet"）vs 已放弃（"abandoned"）
• 每个文件的迭代次数（result.iterations_used）
• 未达标原因（result.unmet_reasons）和客观阻碍标记（result.objective_blocker）

generate_process.json 结构

{
  "version": "1.0",
  "generated_at": "2026-04-21T12:00:00",
  "baseline_ref": "test/generated_unit/test_cases.json",
  "max_iterations": 3,
  "shards_root": ".test",
  "coverage_config": {
    "statement_threshold": 90,
    "branch_threshold": 80,
    "function_threshold": 100,
    "no_progress_rounds": 2,
    "per_function_max_iterations": 3
  },
  "files": {
    "core/parser.py": {
      "file_md5": "abc123",
      "test_path": "test/generated_unit/core/test_parser.py",
      "functions": {
        "parse_header": {
          "dimensions": ["functional", "boundary", "exception"],
          "line_range": [10, 50],
          "signature": "def parse_header(data: bytes) -> dict",
          "mocks_needed": [],
          "test_optional": false,
          "func_md5": "def123"
        }
      },
      "status": "pending",
      "claim_round": 0,
      "attempt_count": 0,
      "effective_attempt_count": 0,
      "last_error_category": null,
      "last_attempt_at": null,
      "abandon_reason": null,
      "result": null
    }
  }
}

字段	说明
claim_round	已被 claim 的次数（含 stale reclaim），首次 claim 时从 0 变 1
attempt_count	有效尝试次数（stale reclaim 不计入）；attempt_count >= 3 且无产物 → 自动 abandoned；产物校验通过后归零
effective_attempt_count	每次 claim 都 +1（含 stale reclaim），用于 AIMD 节流判断重试频率和自动 abandoned 阈值（≥3）
last_error_category	产物校验失败时由 verify-artifacts 写入：no_artifact
last_attempt_at	最近一次 claim 的 ISO 时间戳
abandon_reason	"exhausted_attempts"（限流/崩溃耗尽尝试）或 "all_source_bugs"（所有 gap 被源码 bug 阻塞）

注意：coverage_config 中的 no_progress_rounds 和 per_function_max_iterations 可能在 test_cases.json 中不存在，此时 sub-agent 使用默认值（2 和 3）。claimed_at 和 claim_round 在初始状态不存在，首次 dispatch claim 时写入。

status 状态机：

pending ──(dispatch claim)──▶ running ──(子 agent 返回 + 产物校验通过)──▶ completed / unmet / abandoned
                                 │
                                 ├─(产物缺失 + stale 超时)──▶ pending（回收重试）
                                 │
                                 └─(effective_attempt_count >= 3 且无产物)──▶ abandoned（熔断放弃）

status	含义
pending	还没派发过，或已被回收等待重试
running	dispatch claim 已写入 claimed_at
completed	子 agent 返回 unmet_reasons == []，达到阈值
unmet	未达标且原因非客观（迭代不够 / 测试方案不足），objective_blocker == false
abandoned	effective_attempt_count >= 3 无产物（abandon_reason = "exhausted_attempts"），或所有 gap 函数都被 source_bug 阻塞（abandon_reason = "all_source_bugs"）

子 agent 返回结果结构

{
  "source_path": "core/parser.py",
  "test_path": "test/generated_unit/core/test_parser.py",
  "functions": {
    "parse_header": {
      "dimensions": ["functional", "boundary", "exception"],
      "coverage": {
        "statement": { "target": 90, "actual": 95.5 },
        "branch": { "target": 80, "actual": 88.0 },
        "function": { "target": 100, "actual": 100 }
      }
    }
  },
  "unmet_reasons": [],
  "objective_blocker": true,
  "iterations_used": 3
}

故障排查出口

• 测试全挂（导入错误）：子 agent 在 test/generated_unit/conftest.py 新建最小实现
• 覆盖率为 0：pytest-cov 未装；runner.py 会标记 tool_status
• 源码 md5 漂移：runner.py 检测到漂移写入 md5_drifts，提醒用户重跑 init
• 子 agent 超时 / 崩溃：不用手工清理。下一次 dispatch claim --stale-seconds N 会把 claimed_at 早于 now-N 秒的"running"任务自动回收进本批，reclaimed_stale 字段会列出被回收的文件路径。

依赖

• Python 3.9+
• pip install pytest pytest-cov coverage

参考文档

• references/run-state-schema.md：test_run_state.json 结构 + case 字段定义
• references/run-result-schema.md：.test/run_result.json 结构 + 覆盖率配置
• references/failure-classification.md：LLM 判定 test_code_bug / source_code_bug 的规则