From dee06368ea9a1301e01d3b73b05ba2e8ba86ef5b Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 15 Oct 2026 05:22:00 +0000 Subject: [PATCH] Add real execution agent probe harness Why: verify configured Agent CLIs one profile at a time before declaring them usable Evidence: 53 focused tests, agent-exec validation, 397-test governance regression, and diff check Impact: adds opt-in source-external probes, acceptance checks, and current task records Refs: U5; docs/superpowers/plans/2026-10-15-project-agent-execution.md --- .task/feature-goal-first-delivery.md | 41 +- AGENT-EXECUTION.md | 6 +- tests/fixtures/agent-exec/README.md | 68 +++ tests/test_agent_exec.py | 698 ++++++++++++++++++++++++++- 4 files changed, 797 insertions(+), 16 deletions(-) create mode 100644 tests/fixtures/agent-exec/README.md diff --git a/.task/feature-goal-first-delivery.md b/.task/feature-goal-first-delivery.md index d8bbfb3e..85a48432 100644 --- a/.task/feature-goal-first-delivery.md +++ b/.task/feature-goal-first-delivery.md @@ -12,7 +12,7 @@ - 实施执行：agent-per-unit（配置：AGENT-EXECUTION.md） - 合入执行：agent（已授权交付范围内，Review 通过、冲突解决且当前 HEAD 的 CI 通过后合入） - 人工决策点：仅目标/范围/验收变更、未知外部副作用、高后果操作或无法自动判定的业务验收；计划完成后不停顿 -- 当前阶段：实施与自检；当前增量：U4（项目治理接线；U1–U3 已合入规范基线） +- 当前阶段：实施与自检；当前增量：U5（opt-in 真实 CLI 探针夹具；U4 已规范合入 `0f4c5c017d444369470443faede3283e101c3be9`；七个 profile 探针 NOT_RUN） - 流程依据：仓库根目录 policy/delivery.md（归档后仍按仓库根目录定位） ## 目标与非目标 @@ -23,7 +23,7 @@ ## Agent 实施收据 -现行执行权威：根目录 [AGENT-EXECUTION.md](../AGENT-EXECUTION.md) 的 agent-per-unit 路由与收据门控的 host fallback（`host-fallback-authorized`）。历史 I1/P2 Cursor 行仅为早期 cursor-per-unit 时期记录，不是现行权威，仓库检索不得将其当作活跃固定 Cursor 要求。U1–U3 早于活跃 agent-exec 路由，由 ce-work 固定 Cursor 控制器实施，非本配置角色路由选型。源码外 ce-work 运行根：`/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015`（`manifest.json`）。 +现行执行权威：根目录 [AGENT-EXECUTION.md](../AGENT-EXECUTION.md) 的 agent-per-unit 路由与收据门控的 host fallback（`host-fallback-authorized`）。历史 I1/P2 Cursor 行仅为早期 cursor-per-unit 时期记录，不是现行权威，仓库检索不得将其当作活跃固定 Cursor 要求。U1–U3 早于活跃 agent-exec 路由，由 ce-work 固定 Cursor 控制器实施，非本配置角色路由选型。源码外 ce-work 运行根：`/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015`（`manifest.json`）。U4 已由 host 集成：规范 `0f4c5c017d444369470443faede3283e101c3be9`（父 `8f11d002aec74ed21d3dae8ccbb760963b99e93a`，树 `687ac553730e12e67e6ff3540162a217b8e46e05`），accepted `attempt-2`，job `20261015T045650Z-ab6d6c41`，transport `2ef4f76887d5dfc8a069c74da03f32a8be12e88a`，`model_receipt_status=verified`（Cursor Agent / Auto）；attempt-1 已放弃。以上为该 manifest 的源码外事实，本卡不手抄命令输出。U5 七个 profile 探针保持 NOT_RUN。 | 单元 ID | 预期范围 | 配置摘要 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 | |---|---|---|---|---|---|---|---|---|---|---| @@ -33,8 +33,23 @@ | U1 | 配置合同 AGENT-EXECUTION.md / parser | 早于活跃 agent-exec 路由；ce-work 固定 Cursor 控制器；accepted `attempt-4`；job `20261015T021049Z-c63f0c9c`；见 `manifest.json` | implementation | 非角色路由；ce-work 固定 Cursor | attempts 1–4；accepted `attempt-4` | Cursor Agent / Auto（已核实） | 完成；规范 `bdad8ffc91beab7f0008a3382b650bfa15394065` | `5ce1c685ccf9b4da20f0ca7bb2449bf0ce8cef45` | 无 / 未授权 | `/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015`（manifest.json） | | U2 | adapter seam | 早于活跃 agent-exec 路由；ce-work 固定 Cursor 控制器；accepted `attempt-4`；job `20261015T025219Z-8edd087e`；见 `manifest.json` | implementation | 非角色路由；ce-work 固定 Cursor | attempts 1–4；accepted `attempt-4` | Cursor Agent / Auto（已核实） | 完成；规范 `92330a225f9d839aaaf279e6ae1ad96a4bdfebb4` | `cfdae5f9cf0dbec3ad991df77d39852236927255` | 无 / 未授权 | `/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015`（manifest.json） | | U3 | 可恢复执行控制器 | 早于活跃 agent-exec 路由；ce-work 固定 Cursor 控制器；accepted `attempt-9`；job `20261015T042945Z-e824725f`；见 `manifest.json` | implementation | 非角色路由；ce-work 固定 Cursor | attempts 1–9；accepted `attempt-9` | Cursor Agent / Auto（已核实） | 完成；规范 `8f11d002aec74ed21d3dae8ccbb760963b99e93a` | `6671fc2d6bdb7b64d8a83cc53a27d9443c98e61a` | 无 / 未授权 | `/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015`（manifest.json） | -| U4 | 项目治理接线（本单元） | NOT_RUN | implementation | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | `/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015` | -| U5 | 真实 CLI 探针与组合验收 | NOT_RUN | implementation | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — | +| U4 | 项目治理接线 | ce-work 固定 Cursor 控制器；accepted `attempt-2`；job `20261015T045650Z-ab6d6c41`；attempt-1 已放弃；见源码外 `manifest.json` | implementation | 非角色路由；ce-work 固定 Cursor | attempts 1–2；accepted `attempt-2` | Cursor Agent / Auto（manifest `model_receipt_status=verified`） | 完成；规范 `0f4c5c017d444369470443faede3283e101c3be9` | `2ef4f76887d5dfc8a069c74da03f32a8be12e88a` | 无 / 未授权 | `/tmp/compound-engineering-1000/ce-work/project-agent-exec-20261015` | +| U5 | 真实 CLI 探针与组合验收 | 夹具与验收说明已写入；七个 profile 探针 NOT_RUN | implementation | 每次恰好一个 `AGENT_EXEC_REAL_PROFILE` | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | 源码外收据待 host 执行后产生；不入库 | + +### U5 真实探针字段（host 稍后逐 profile 执行） + +门禁：`RUN_REAL_AGENT_EXEC=1`，且 `AGENT_EXEC_REAL_PROFILE` 每次恰好一个 profile。默认测试不接触 provider。缺失或不兼容的 OpenCode 记 `unavailable`，不得记 PASS。验收说明见 `tests/fixtures/agent-exec/README.md`。 + +| profile | 探针 | 模型收据 | transport | 变更路径 | 清理/保留 | 规范检出 | 能力结论 | +|---|---|---|---|---|---|---|---| +| cursor | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| antigravity | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| claude | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| codex | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| pi | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| kimi | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| opencode | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | + ## 关联方向与规划 @@ -58,7 +73,7 @@ - 方案引用：[2026-10-15 项目级执行 Agent 路由设计](../docs/superpowers/specs/2026-10-15-project-agent-execution-design.md)（已批准） - 实施计划引用：[2026-10-15 项目级执行 Agent 路由实施计划](../docs/superpowers/plans/2026-10-15-project-agent-execution.md)（用户明确免除计划复审） -- 测试计划引用：U1–U3 本地 unittest / agent-exec validate 为 SELF；U4 聚焦任务卡模板与 validate；U5 真实 CLI 探针与完整治理回归（GATE 待 PR HEAD） +- 测试计划引用：U1–U3 本地 unittest / agent-exec validate 为 SELF；U4 聚焦任务卡模板与 validate（SELF，源码外 manifest verification summary）；U5 真实 CLI 探针默认 NOT_RUN，仅 `RUN_REAL_AGENT_EXEC=1` 且单 profile 时由 host 执行；完整治理回归与 GATE 待 PR HEAD ## 阶段评审 @@ -69,7 +84,7 @@ | 需求表确认 | R1–R7 | 通过 | weichen | 2026-10-15 | 用户确认多 CLI 受控 adapter、`run-everything` 优先、主模型低消耗与全候选不可用后回退 | | 方案设计 | `docs/superpowers/specs/2026-10-15-project-agent-execution-design.md` | 通过 | weichen | 2026-10-15 | 用户明确“批准” | | 实施计划 | `docs/superpowers/plans/2026-10-15-project-agent-execution.md` | 授权连续执行 | weichen | 2026-10-15 | 用户明确“实施计划后不需要再让我审阅”，并指定 Cursor Agent 执行 | -| 实施与自检 | U1 `bdad8ffc` / U2 `92330a22` / U3 `8f11d002`；U4/U5 进行中 | 进行中 | Codex host + 外部 worker | 2026-10-15 | U1–U3 规范合入；U4 治理接线；U5 探针 NOT_RUN；历史 P2/`8b7e8ca` 仅作旧对象线索 | +| 实施与自检 | U1 `bdad8ffc` / U2 `92330a22` / U3 `8f11d002` / U4 `0f4c5c017d444369470443faede3283e101c3be9`；U5 探针 NOT_RUN | 进行中 | Codex host + 外部 worker | 2026-10-15 | U1–U4 规范合入；U5 夹具已写、真实探针 NOT_RUN；历史 P2/`8b7e8ca` 仅作旧对象线索 | | 正式测试与验收 | 待定（U1–U5 组合候选） | 待执行 | — | — | 旧 `cursor-per-unit` 候选 `cc03710` 已停止交付；不发明 GATE/REPLAY | | 代码 Review | 待定 | 未评审 | — | — | — | | 合入 | 待定 | 未评审 | — | — | — | @@ -103,16 +118,16 @@ ## 结果与验证（完成后填写） -- 结果：旧 cursor-per-unit 候选已停止交付并保留为历史；U0 设计已批准；U1–U3 已合入规范基线（`bdad8ffc…` / `92330a22…` / `8f11d002…`）；U4 将项目规则改为配置驱动 agent-per-unit；U5 真实探针 NOT_RUN。 -- 验证：U1–U3 本地 SELF 见各单元聚焦测试与 host 集成；U4 本地聚焦任务卡测试与 `agent_exec.py validate`（SELF）；完整 `make test-governance` / PR CI（GATE）由 host 权威事务执行，本卡不手抄。历史 P2 digest `d16726eb13aba7f5` 仅作旧对象线索。 -- 影响与剩余项：U5 探针、独立 Review、精确 HEAD CI、合入与任务卡闭合仍 NOT_RUN。 +- 结果：旧 cursor-per-unit 候选已停止交付并保留为历史；U0 设计已批准；U1–U4 已合入规范基线（`bdad8ffc…` / `92330a22…` / `8f11d002…` / `0f4c5c017d444369470443faede3283e101c3be9`）；U5 真实探针 NOT_RUN。 +- 验证：U1–U3 本地 SELF 见各单元聚焦测试与 host 集成；U4 本地聚焦任务卡测试、`agent_exec.py validate` 与 `make test-governance` 的结论见源码外 manifest verification summary（SELF，digest `8f277d49b61e57d97e39beb16d5a4cad702e7043d670094e3ebe0c9b1d26945f`），本卡不手抄该次输出。U5 默认测试不接触 provider，七个 profile 探针 NOT_RUN。完整 PR CI（GATE）由 host 权威事务执行。历史 P2 digest `d16726eb13aba7f5` 仅作旧对象线索。 +- 影响与剩余项：U5 七个 profile 探针、host 回退演练、独立 Review、精确 HEAD CI、合入与任务卡闭合仍 NOT_RUN。 ## 规划进展与下一步 -- 本次推进/剩余：U1–U3 已完成；U4 治理接线进行中；U5 组合验收待办。 -- 首选下一任务：U5 真实 CLI 探针与组合验收（新 agent-per-unit 进程）。 -- 推荐依据：遵循本次 goal-first 与不扩围原则；U4 完成后才具备日常任务路由权威。 -- 前置条件/待确认：U4 聚焦验证通过并由 host 权威集成；无待人工复审项。 +- 本次推进/剩余：U1–U4 已规范合入；U5 已添加 opt-in 探针夹具与验收说明，七个 profile 的真实探针仍 NOT_RUN。 +- 首选下一任务：host 按 profile 逐个执行 U5 真实探针（每次一个 `AGENT_EXEC_REAL_PROFILE`），再进入 Review 与精确 HEAD CI。 +- 推荐依据：U4 已提供 agent-per-unit 路由权威；探针未执行前不得把已安装 CLI 记为就位。 +- 前置条件/待确认：host 在源码外逐 profile 执行探针；无待人工复审项。 - 预期产物/验收条件：R1–R10 有代码/测试/证据映射；无活跃固定 Cursor 权威；U5 后决定是否声明已安装 CLI 就位。 - 重复任务核对：未发现需另建任务的重复范围。 - 启动授权：U1–U5 已授权连续执行；U5 之外新任务仍须明确授权 diff --git a/AGENT-EXECUTION.md b/AGENT-EXECUTION.md index a824e854..c936190c 100644 --- a/AGENT-EXECUTION.md +++ b/AGENT-EXECUTION.md @@ -5,8 +5,10 @@ authority for which Agent CLI profiles may implement, repair, test, or review bounded units. Human-readable notes may appear outside the machine block; the parser accepts exactly one `agent-execution` fenced block. -U1 delivers configuration parsing and validation only. Routing is not enabled -until later units wire adapters and the execution controller. +U3's execution controller and U4's project wiring are integrated. This file is +the execution-routing authority for implementation, repair, test, and review. +Installed CLI identity stays unproven until a host runs the U5 probes one +profile at a time. ```agent-execution schema: agent-execution/v1 diff --git a/tests/fixtures/agent-exec/README.md b/tests/fixtures/agent-exec/README.md new file mode 100644 index 00000000..63386177 --- /dev/null +++ b/tests/fixtures/agent-exec/README.md @@ -0,0 +1,68 @@ +# U5 真实 Agent CLI 探针 + +本夹具描述 host 稍后执行的 opt-in 探针。七个 profile 的真实结果都还是 `NOT_RUN`。这里不记录探针 `PASS`。 + +## 门禁 + +默认测试不设置下面两个变量，因此不会启动真实 CLI，也不会联系 provider。 + +- `RUN_REAL_AGENT_EXEC=1`：显式打开真实探针。其他值（包括 `true`）保持关闭。 +- `AGENT_EXEC_REAL_PROFILE`：每次恰好一个 profile。允许的名字是 `cursor`、`antigravity`、`claude`、`codex`、`pi`、`kimi`、`opencode`。空值、空白、逗号或空格列表都会拒绝，并且不会启动 CLI。 + +一次只跑一个 profile，不要对同一个目标并行开多个探针。 + +```bash +RUN_REAL_AGENT_EXEC=1 AGENT_EXEC_REAL_PROFILE=cursor \ + PYTHONDONTWRITEBYTECODE=1 python3 -m unittest \ + tests.test_agent_exec.RealAgentCliProbeTests.test_opt_in_real_profile_probe -v +``` + +把 `cursor` 换成要测的那一个 profile。建议顺序：`cursor`，然后 `antigravity`（可执行文件 `agy`），`claude`，`pi`，`codex`，`kimi`。`opencode` 单独跑。 + +## 夹具 + +探针在临时 Git 仓库里进行，不使用本仓的规范工作区。 + +- 分支是非默认分支 `feature/u5-real-probe`。 +- 配置里只启用被选中的那一个 profile，`model` 为 `auto`（省略模型参数，不把字符串 `auto` 传给 CLI）。 +- 单元只允许一个文件：`src/u5-marker.txt`。 +- 文件字节必须是 ASCII `U5-REAL-PROBE-MARKER` 再加一个换行 `0x0A`。 + +## 核对 + +Host 只在机械检查全部成立时把能力结论记为 `PASS`： + +- transport 树和工作区里的标记内容与上面的字节完全一致。 +- transport 提交只有一个父提交，且该父提交就是夹具基线 HEAD。 +- `changed_paths` 只有 `src/u5-marker.txt`。 +- 成功时保留隔离工作区，`transport_created` 为真，`exact_restore` 为假。 +- 模型收据只记录实际看得到的状态。原始输出里没有 `model_receipt_status` 时记 `not-observable`，不得编造成 `verified`。观察到 `mismatch` 时结论不是 `PASS`。 +- 规范检出的 HEAD、分支、状态和索引保持不变，标记文件不得出现在规范检出里。 + +## 不可用与回退 + +缺失或不兼容的 OpenCode 必须记 `unavailable`，不得记 `PASS`。不兼容指本地 help 缺少 adapter 要求的能力标志；这种情况下不得把 CLI 当成已经跑通。 + +唯一候选耗尽时，控制器的运行级终态可以是 `host-fallback-authorized`。那是回退授权，不是探针通过。夹具不会代替 CLI 去写标记文件。 + +Provider 配额、认证失败，或进程启动后没有留下标记且规范检出仍然完好，同样记 `unavailable`，不得记 `PASS`。范围外改动、规范检出被改写、标记内容不符，记 `failed`。 + +## 源码外收据 + +真实 opt-in 运行会保留临时仓库，并把收据复制到 `/tmp/agent-exec-/u5-probes/`。原始 CLI 日志留在控制器的私有 run 目录。这些文件不进入 Git。 + +默认单元测试使用隔离 PATH 上的假 CLI，只验证上述核对逻辑。 + +## 仍为 NOT_RUN + +| profile | 探针 | 模型收据 | transport | 变更路径 | 清理/保留 | 规范检出 | 能力结论 | +|---|---|---|---|---|---|---|---| +| cursor | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| antigravity | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| claude | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| codex | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| pi | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| kimi | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | +| opencode | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | + +Host 回退演练、完整治理回归、独立 Review、精确 HEAD CI 与合入也不在本夹具的已执行范围内。 diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py index 8827f69d..7453376e 100644 --- a/tests/test_agent_exec.py +++ b/tests/test_agent_exec.py @@ -1,8 +1,14 @@ -"""End-to-end execution controller contract for agent-exec (U3).""" +"""End-to-end execution controller contract for agent-exec (U3). + +U5 adds an opt-in real CLI probe. Default discovery does not contact providers. +Set RUN_REAL_AGENT_EXEC=1 and AGENT_EXEC_REAL_PROFILE to exactly one profile. +""" from __future__ import annotations import json import os +import re +import shlex import shutil import stat import subprocess @@ -1368,5 +1374,695 @@ class AgentExecControllerTests(unittest.TestCase): self._assert_canonical_untouched() +# Opt-in real CLI probes (U5). Default unittest must not contact providers. +U5_PROFILES = ( + "cursor", + "antigravity", + "claude", + "codex", + "pi", + "kimi", + "opencode", +) +U5_EXECUTABLE = { + "cursor": "cursor-agent", + "antigravity": "agy", + "claude": "claude", + "codex": "codex", + "pi": "pi", + "kimi": "kimi", + "opencode": "opencode", +} +U5_BRANCH = "feature/u5-real-probe" +U5_MARKER_REL = "src/u5-marker.txt" +U5_MARKER_BODY = "U5-REAL-PROBE-MARKER\n" +_MODEL_RECEIPT_RE = re.compile( + r'"model_receipt_status"\s*:\s*"(verified|not-observable|mismatch|unavailable)"' +) + + +def real_probe_opt_in(env: os._Environ[str] | dict[str, str] | None = None) -> bool: + """True only for the explicit host gate. Other truthy strings stay closed.""" + source = os.environ if env is None else env + return source.get("RUN_REAL_AGENT_EXEC") == "1" + + +def require_single_real_profile(env: dict[str, str]) -> str: + """Select exactly one installed-CLI profile. Reject lists and unknown names.""" + raw = env.get("AGENT_EXEC_REAL_PROFILE", "") + if not raw or raw.strip() != raw: + raise ValueError( + "AGENT_EXEC_REAL_PROFILE must name exactly one profile" + ) + if any(sep in raw for sep in (",", " ", "\t", "\n", ";", ":")): + raise ValueError( + "AGENT_EXEC_REAL_PROFILE must name exactly one profile" + ) + if raw not in U5_PROFILES: + raise ValueError(f"unknown real probe profile: {raw}") + return raw + + +def _observe_model_receipt(attempt: dict[str, object]) -> str: + """Record a model receipt only when the raw output actually carries one.""" + chunks = [str(attempt.get("summary") or "")] + raw_log = attempt.get("raw_log") + if isinstance(raw_log, str) and raw_log: + path = Path(raw_log) + if path.is_file(): + chunks.append(path.read_text(encoding="utf-8", errors="replace")) + found = _MODEL_RECEIPT_RE.findall("\n".join(chunks)) + if not found: + return "not-observable" + if "mismatch" in found: + return "mismatch" + return found[-1] + + +def _probe_tool_bindir(extra: Path | None = None) -> tempfile.TemporaryDirectory[str]: + """PATH containing only git, python3, and an optional fake CLI directory.""" + tool_dir = tempfile.TemporaryDirectory(prefix="u5-probe-path-") + bindir = Path(tool_dir.name) + if extra is not None: + for entry in extra.iterdir(): + target = bindir / entry.name + if entry.is_file(): + target.symlink_to(entry.resolve()) + for tool in ("git", "python3"): + found = shutil.which(tool) + if not found: + tool_dir.cleanup() + raise RuntimeError(f"required tool missing from PATH: {tool}") + link = bindir / tool + if not link.exists(): + link.symlink_to(found) + return tool_dir + + +def _write_probe_double( + bindir: Path, + name: str, + *, + help_text: str, + run_body: str, +) -> None: + path = bindir / name + # printf is a shell builtin. The offline PATH has no cat/mkdir. + help_lines = help_text.splitlines() or [""] + help_printf = "".join( + f" printf '%s\\n' {shlex.quote(line)}\n" for line in help_lines + ) + path.write_text( + "#!/bin/sh\n" + 'if [ "$1" = "--version" ] || [ "$1" = "-v" ] || [ "$1" = "-V" ]; then\n' + ' echo "0.0.1-probe"\n' + " exit 0\n" + "fi\n" + 'if [ "$1" = "--help" ] || [ "$1" = "-h" ]; then\n' + + help_printf + + " exit 0\n" + "fi\n" + + run_body, + encoding="utf-8", + ) + path.chmod(path.stat().st_mode | stat.S_IEXEC) + + +def _cursor_help() -> str: + from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS + + return "\n".join(_REQUIRED_HELP_MARKERS["cursor"]) + "\n" + + +def _marker_run_body(*, body: str, model_receipt: str | None) -> str: + payload: dict[str, object] = { + "terminal_status": "completed", + "summary": "wrote marker", + "changed_files": [U5_MARKER_REL], + "evidence": [], + "scope_expansion": None, + } + if model_receipt is not None: + payload["model_receipt_status"] = model_receipt + encoded = json.dumps(payload, separators=(",", ":")) + writer = ( + "from pathlib import Path; " + f"p=Path({json.dumps(U5_MARKER_REL)}); " + "p.parent.mkdir(parents=True); " + f"p.write_text({json.dumps(body)}, encoding='utf-8')" + ) + return ( + f"python3 -c {shlex.quote(writer)}\n" + f"printf '%s\\n' {shlex.quote(encoded)}\n" + "exit 0\n" + ) + + +def _write_probe_unit(path: Path, timeout_seconds: int) -> None: + packet = { + "unit_id": "u5-real-probe", + "goal": ( + "Create src/u5-marker.txt containing the ASCII bytes " + "U5-REAL-PROBE-MARKER followed by one newline (0x0A). " + "Do not create or modify any other path." + ), + "allowed_paths": [U5_MARKER_REL], + "verification": ["test -f src/u5-marker.txt"], + "constraints": [ + "Write only src/u5-marker.txt", + "Exact bytes are U5-REAL-PROBE-MARKER plus a single trailing newline", + "Leave the change uncommitted", + "Do not commit, push, open a PR, merge, or deploy", + ], + "network": True, + "external_cost": True, + "timeout_seconds": timeout_seconds, + } + path.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8") + + +def _cleanup_probe_runtime(repo: Path, receipt: dict[str, object] | None) -> None: + if not receipt: + return + run_dir = receipt.get("run_dir") + attempts = receipt.get("attempts") + if isinstance(attempts, list): + for attempt in attempts: + if not isinstance(attempt, dict): + continue + worktree = attempt.get("worktree") + if isinstance(worktree, str) and worktree: + _git(repo, "worktree", "remove", "--force", worktree, check=False) + shutil.rmtree(worktree, ignore_errors=True) + if isinstance(run_dir, str) and run_dir: + shutil.rmtree(run_dir, ignore_errors=True) + + +def execute_marker_probe( + profile: str, + *, + offline_bindir: Path | None, + timeout_seconds: int = 20, + retain_fixture: bool = False, +) -> dict[str, object]: + """Run one marker unit against a single profile. + + Offline fixtures pass ``offline_bindir`` and never use the ambient PATH. + A real profile requires RUN_REAL_AGENT_EXEC=1 and keeps its fixture. + """ + if profile not in U5_EXECUTABLE: + raise ValueError(f"unknown real probe profile: {profile}") + if offline_bindir is None and not real_probe_opt_in(): + raise RuntimeError( + "real Agent CLI probe refused without RUN_REAL_AGENT_EXEC=1" + ) + + tool_dir: tempfile.TemporaryDirectory[str] | None = None + repo_dir: tempfile.TemporaryDirectory[str] | None = None + repo: Path | None = None + receipt: dict[str, object] | None = None + try: + if offline_bindir is None: + probe_path = os.environ.get("PATH", "") + else: + tool_dir = _probe_tool_bindir(offline_bindir) + probe_path = tool_dir.name + if retain_fixture: + repo_root = Path(tempfile.mkdtemp(prefix="u5-real-probe-")) + else: + repo_dir = tempfile.TemporaryDirectory(prefix="u5-real-probe-") + repo_root = Path(repo_dir.name) + repo = repo_root / "repo" + repo.mkdir() + receipts = repo_root / "receipts" + receipts.mkdir() + init = _git(repo, "init", "-b", U5_BRANCH, check=False) + if init.returncode != 0: + _git(repo, "init") + _git(repo, "checkout", "-b", U5_BRANCH) + _git(repo, "config", "user.email", "agent-exec@example.com") + _git(repo, "config", "user.name", "Agent Exec") + (repo / "README.md").write_text("fixture\n", encoding="utf-8") + _git(repo, "add", "README.md") + _git(repo, "commit", "-m", "init") + branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip() + if branch != U5_BRANCH: + raise RuntimeError( + f"probe fixture must use non-default branch {U5_BRANCH}, got {branch}" + ) + _write_config( + repo, + profiles={ + profile: { + "adapter": profile, + "executable": U5_EXECUTABLE[profile], + "model": "auto", + } + }, + role_order=[profile], + timeout_seconds=timeout_seconds, + summary_bytes=4096, + raw_output_bytes=65536, + ) + before = _canonical_snapshot(repo) + unit = repo_root / "unit.json" + _write_probe_unit(unit, timeout_seconds) + receipt_path = receipts / "probe.json" + env = os.environ.copy() + env["PYTHONDONTWRITEBYTECODE"] = "1" + env["PATH"] = probe_path + workspace = Path(__file__).resolve().parents[1] + existing = env.get("PYTHONPATH", "") + env["PYTHONPATH"] = ( + f"{workspace}{os.pathsep}{existing}" if existing else str(workspace) + ) + proc = subprocess.run( + [ + "python3", + "-m", + "scripts.agent_exec", + "run", + "--repo", + str(repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt_path), + ], + cwd=str(repo_root), + capture_output=True, + text=True, + env=env, + timeout=timeout_seconds + 30, + ) + receipt = ( + json.loads(receipt_path.read_text(encoding="utf-8")) + if receipt_path.is_file() + else None + ) + outcome = _classify_marker_probe( + profile=profile, + repo=repo, + before=before, + receipt=receipt, + exit_code=proc.returncode, + stderr=proc.stderr, + ) + outcome["fixture_repo"] = str(repo) + if retain_fixture and receipt_path.is_file(): + durable = Path(f"/tmp/agent-exec-{os.getuid()}") / "u5-probes" + durable.mkdir(mode=0o700, parents=True, exist_ok=True) + os.chmod(durable, 0o700) + stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + copied = durable / f"{profile}-{stamp}.json" + shutil.copy2(receipt_path, copied) + outcome["source_external_receipt"] = str(copied) + return outcome + finally: + if not retain_fixture and repo is not None: + _cleanup_probe_runtime(repo, receipt if isinstance(receipt, dict) else None) + if tool_dir is not None: + tool_dir.cleanup() + if repo_dir is not None: + repo_dir.cleanup() + + +def _classify_marker_probe( + *, + profile: str, + repo: Path, + before: dict[str, str], + receipt: dict[str, object] | None, + exit_code: int, + stderr: str, +) -> dict[str, object]: + base: dict[str, object] = { + "profile": profile, + "exit_code": exit_code, + "canonical_preserved": _canonical_snapshot(repo) == before, + "marker_in_canonical": (repo / U5_MARKER_REL).exists(), + "configured_model": None, + "model_receipt_status": "not-observable", + "branch": None, + "changed_paths": [], + "transport_parent": None, + "transport_ref": None, + "attempt_status": None, + "run_terminal": None, + "cleanup": {}, + "host_fallback_authorizations": 0, + "base_head": before["head"], + } + if not base["canonical_preserved"]: + base["capability"] = "failed" + base["reason"] = "canonical checkout mutated" + return base + if base["marker_in_canonical"]: + base["capability"] = "failed" + base["reason"] = "canonical checkout contains the probe marker" + return base + if not isinstance(receipt, dict): + base["capability"] = "failed" + base["reason"] = f"missing receipt: {stderr[-500:]}" + return base + + base["run_terminal"] = receipt.get("terminal_status") + base["host_fallback_authorizations"] = receipt.get( + "host_fallback_authorizations", 0 + ) + summary = receipt.get("config_summary") + profiles = summary.get("profiles") if isinstance(summary, dict) else None + roles = summary.get("roles") if isinstance(summary, dict) else None + if not isinstance(profiles, dict) or set(profiles) != {profile}: + base["capability"] = "failed" + base["reason"] = "probe configured a profile other than the selection" + return base + selected = profiles[profile] + if isinstance(selected, dict): + base["configured_model"] = selected.get("model") + implementation = roles.get("implementation") if isinstance(roles, dict) else None + if implementation != [profile]: + base["capability"] = "failed" + base["reason"] = "implementation role is not exactly the selected profile" + return base + recorded_base = receipt.get("base") + if isinstance(recorded_base, dict): + base["branch"] = recorded_base.get("branch") + if base["branch"] in {"main", "master", "HEAD", None}: + base["capability"] = "failed" + base["reason"] = f"probe branch is not a non-default branch: {base['branch']}" + return base + + attempts = receipt.get("attempts") + if not isinstance(attempts, list) or len(attempts) != 1: + base["capability"] = "failed" + base["reason"] = "probe must record exactly one attempt" + return base + attempt = attempts[0] + if not isinstance(attempt, dict) or attempt.get("profile") != profile: + base["capability"] = "failed" + base["reason"] = "attempt profile does not match the selection" + return base + base["attempt_status"] = attempt.get("terminal_status") + base["changed_paths"] = list(attempt.get("changed_paths") or []) + base["transport_ref"] = attempt.get("transport_ref") + cleanup = attempt.get("cleanup") if isinstance(attempt.get("cleanup"), dict) else {} + base["cleanup"] = { + "exact_restore": bool(cleanup.get("exact_restore")), + "transport_created": bool(cleanup.get("transport_created")), + "workspace_preserved": bool(attempt.get("workspace_preserved")), + "skipped": bool(cleanup.get("skipped")), + } + base["model_receipt_status"] = _observe_model_receipt(attempt) + reason = attempt.get("reason") or receipt.get("reason") + + if attempt.get("terminal_status") == "unavailable": + base["capability"] = "unavailable" + base["reason"] = reason or "adapter unavailable" + if base["transport_ref"] or base["cleanup"]["transport_created"]: + base["capability"] = "failed" + base["reason"] = "unavailable probe still created transport" + return base + + if ( + receipt.get("terminal_status") == "success" + and attempt.get("terminal_status") == "success" + ): + return _classify_successful_marker( + repo=repo, + receipt=receipt, + attempt=attempt, + before=before, + base=base, + ) + + changed = list(attempt.get("changed_paths") or []) + if ( + base["canonical_preserved"] + and not base["marker_in_canonical"] + and not attempt.get("transport_ref") + and not changed + and attempt.get("terminal_status") + in {"failed-clean", "failed-unknown", "unavailable"} + ): + base["capability"] = "unavailable" + base["reason"] = reason or "probe produced no marker" + return base + + base["capability"] = "failed" + base["reason"] = reason or f"unclassified probe status {attempt.get('terminal_status')}" + return base + + +def _classify_successful_marker( + *, + repo: Path, + receipt: dict[str, object], + attempt: dict[str, object], + before: dict[str, str], + base: dict[str, object], +) -> dict[str, object]: + if base["model_receipt_status"] == "mismatch": + base["capability"] = "failed" + base["reason"] = "model receipt mismatch" + return base + if base["configured_model"] is not None: + base["capability"] = "failed" + base["reason"] = "auto model must omit the configured model value" + return base + changed = list(attempt.get("changed_paths") or []) + if changed != [U5_MARKER_REL]: + base["capability"] = "failed" + base["reason"] = f"changed paths are not exactly the marker: {changed}" + return base + cleanup = base["cleanup"] + if not isinstance(cleanup, dict): + base["capability"] = "failed" + base["reason"] = "missing cleanup record" + return base + if cleanup.get("exact_restore") or not cleanup.get("transport_created"): + base["capability"] = "failed" + base["reason"] = "success must retain transport and skip exact restore" + return base + if not cleanup.get("workspace_preserved"): + base["capability"] = "failed" + base["reason"] = "success must retain the isolated workspace until host review" + return base + transport = receipt.get("transport") + ref = attempt.get("transport_ref") + if not isinstance(transport, dict) or not isinstance(ref, str) or not ref: + base["capability"] = "failed" + base["reason"] = "success is missing a transport ref" + return base + if transport.get("ref") != ref: + base["capability"] = "failed" + base["reason"] = "receipt transport ref does not match the attempt" + return base + commit = _git(repo, "rev-parse", ref).stdout.strip() + parents = _git(repo, "rev-list", "--parents", "-n", "1", commit).stdout.split() + if len(parents) != 2: + base["capability"] = "failed" + base["reason"] = f"transport commit is not base-parented: {parents}" + return base + parent = parents[1] + base_head = before["head"] + recorded = receipt.get("base") + recorded_head = recorded.get("head") if isinstance(recorded, dict) else None + if parent != base_head or transport.get("parent") != base_head or recorded_head != base_head: + base["capability"] = "failed" + base["reason"] = "transport parent is not the fixture base HEAD" + return base + blob = _git(repo, "cat-file", "-p", f"{commit}:{U5_MARKER_REL}").stdout + worktree = attempt.get("worktree") + live = "" + if isinstance(worktree, str): + marker_path = Path(worktree) / U5_MARKER_REL + if marker_path.is_file(): + live = marker_path.read_text(encoding="utf-8") + if blob != U5_MARKER_BODY or live != U5_MARKER_BODY: + base["capability"] = "failed" + base["reason"] = "marker bytes differ from the required probe content" + base["marker_bytes"] = blob + return base + base["transport_parent"] = parent + base["marker_bytes"] = blob + base["capability"] = "PASS" + base["reason"] = None + return base + + +class RealAgentCliProbeTests(unittest.TestCase): + """U5 marker probe. Real providers run only behind RUN_REAL_AGENT_EXEC=1.""" + + def test_profile_selector_accepts_exactly_one_profile(self) -> None: + self.assertEqual(U5_PROFILES, tuple(U5_EXECUTABLE)) + self.assertFalse(real_probe_opt_in({"RUN_REAL_AGENT_EXEC": "true"})) + self.assertFalse(real_probe_opt_in({})) + self.assertTrue(real_probe_opt_in({"RUN_REAL_AGENT_EXEC": "1"})) + self.assertEqual( + require_single_real_profile({"AGENT_EXEC_REAL_PROFILE": "pi"}), + "pi", + ) + for raw in ("", " cursor", "cursor ", "cursor,codex", "cursor codex", "nope"): + with self.assertRaises(ValueError): + require_single_real_profile({"AGENT_EXEC_REAL_PROFILE": raw}) + + def test_real_probe_refuses_without_opt_in(self) -> None: + self.assertNotEqual(os.environ.get("RUN_REAL_AGENT_EXEC"), "1") + with self.assertRaises(RuntimeError) as caught: + execute_marker_probe("cursor", offline_bindir=None) + self.assertIn("RUN_REAL_AGENT_EXEC=1", str(caught.exception)) + + def test_offline_success_checks_marker_transport_and_canonical(self) -> None: + outcome = self._run_cursor( + _marker_run_body(body=U5_MARKER_BODY, model_receipt=None) + ) + self.assertEqual(outcome["capability"], "PASS", outcome) + self.assertEqual(outcome["marker_bytes"], U5_MARKER_BODY) + self.assertEqual(outcome["changed_paths"], [U5_MARKER_REL]) + self.assertEqual(outcome["transport_parent"], outcome["base_head"]) + self.assertTrue(outcome["transport_parent"]) + self.assertEqual(outcome["branch"], U5_BRANCH) + self.assertIsNone(outcome["configured_model"]) + self.assertEqual(outcome["model_receipt_status"], "not-observable") + self.assertEqual(outcome["run_terminal"], "success") + cleanup = outcome["cleanup"] + self.assertIsInstance(cleanup, dict) + assert isinstance(cleanup, dict) + self.assertTrue(cleanup["transport_created"]) + self.assertTrue(cleanup["workspace_preserved"]) + self.assertFalse(cleanup["exact_restore"]) + self.assertTrue(outcome["canonical_preserved"]) + self.assertFalse(outcome["marker_in_canonical"]) + self.assertEqual(outcome["host_fallback_authorizations"], 0) + + def test_observable_verified_model_receipt_stays_attached_to_success(self) -> None: + outcome = self._run_cursor( + _marker_run_body(body=U5_MARKER_BODY, model_receipt="verified") + ) + self.assertEqual(outcome["capability"], "PASS", outcome) + self.assertEqual(outcome["model_receipt_status"], "verified") + + def test_model_receipt_mismatch_is_not_pass(self) -> None: + outcome = self._run_cursor( + _marker_run_body(body=U5_MARKER_BODY, model_receipt="mismatch") + ) + self.assertEqual(outcome["capability"], "failed", outcome) + self.assertNotEqual(outcome["capability"], "PASS") + self.assertEqual(outcome["model_receipt_status"], "mismatch") + + def test_wrong_marker_is_not_pass(self) -> None: + outcome = self._run_cursor( + _marker_run_body(body="WRONG\n", model_receipt=None) + ) + self.assertEqual(outcome["capability"], "failed", outcome) + self.assertNotEqual(outcome["capability"], "PASS") + self.assertTrue(outcome["canonical_preserved"]) + + def test_missing_opencode_is_unavailable_not_pass(self) -> None: + outcome = self._run_offline("opencode", help_text="", run_body="exit 0\n", install=False) + self.assertEqual(outcome["capability"], "unavailable", outcome) + self.assertNotEqual(outcome["capability"], "PASS") + self.assertEqual(outcome["attempt_status"], "unavailable") + self.assertEqual(outcome["run_terminal"], "host-fallback-authorized") + self.assertEqual(outcome["host_fallback_authorizations"], 1) + self.assertFalse(outcome["marker_in_canonical"]) + self.assertTrue(outcome["canonical_preserved"]) + cleanup = outcome["cleanup"] + assert isinstance(cleanup, dict) + self.assertTrue(cleanup["skipped"]) + self.assertFalse(cleanup["transport_created"]) + + def test_incompatible_opencode_is_unavailable_not_pass(self) -> None: + outcome = self._run_offline( + "opencode", + help_text="opencode\nusage: opencode\n", + run_body=( + "mkdir -p src\n" + f"printf '%s\\n' 'SHOULD-NOT-RUN' > {U5_MARKER_REL}\n" + "exit 0\n" + ), + install=True, + ) + self.assertEqual(outcome["capability"], "unavailable", outcome) + self.assertNotEqual(outcome["capability"], "PASS") + self.assertEqual(outcome["attempt_status"], "unavailable") + self.assertIn("missing required capability", str(outcome["reason"])) + self.assertFalse(outcome["marker_in_canonical"]) + self.assertNotEqual(outcome["run_terminal"], "success") + + def test_launched_failure_without_marker_is_unavailable_not_pass(self) -> None: + outcome = self._run_cursor("echo 'auth failed' >&2\nexit 3\n") + self.assertEqual(outcome["capability"], "unavailable", outcome) + self.assertNotEqual(outcome["capability"], "PASS") + self.assertNotEqual(outcome["run_terminal"], "success") + self.assertEqual(outcome["host_fallback_authorizations"], 0) + self.assertTrue(outcome["canonical_preserved"]) + self.assertFalse(outcome["marker_in_canonical"]) + + @unittest.skipUnless( + real_probe_opt_in(), + "real CLI probe NOT_RUN: set RUN_REAL_AGENT_EXEC=1 and " + "AGENT_EXEC_REAL_PROFILE to exactly one profile", + ) + def test_opt_in_real_profile_probe(self) -> None: + profile = require_single_real_profile(dict(os.environ)) + outcome = execute_marker_probe( + profile, + offline_bindir=None, + timeout_seconds=1800, + retain_fixture=True, + ) + print("U5-PROBE " + json.dumps(outcome, sort_keys=True, default=str)) + if profile == "opencode" and outcome["capability"] == "unavailable": + self.assertNotEqual(outcome["capability"], "PASS") + self.assertEqual(outcome["attempt_status"], "unavailable") + self.assertFalse(outcome["marker_in_canonical"]) + self.assertTrue(outcome["canonical_preserved"]) + return + self.assertEqual(outcome["capability"], "PASS", outcome) + self.assertEqual(outcome["marker_bytes"], U5_MARKER_BODY) + self.assertEqual(outcome["changed_paths"], [U5_MARKER_REL]) + self.assertTrue(outcome["transport_parent"]) + self.assertEqual(outcome["branch"], U5_BRANCH) + self.assertNotEqual(outcome["model_receipt_status"], "mismatch") + self.assertTrue(outcome["canonical_preserved"]) + + def _run_cursor(self, run_body: str) -> dict[str, object]: + return self._run_offline( + "cursor", + help_text=_cursor_help(), + run_body=run_body, + install=True, + ) + + def _run_offline( + self, + profile: str, + *, + help_text: str, + run_body: str, + install: bool, + ) -> dict[str, object]: + fake = tempfile.TemporaryDirectory(prefix="u5-fake-cli-") + self.addCleanup(fake.cleanup) + bindir = Path(fake.name) + if install: + _write_probe_double( + bindir, + U5_EXECUTABLE[profile], + help_text=help_text, + run_body=run_body, + ) + return execute_marker_probe( + profile, + offline_bindir=bindir, + timeout_seconds=20, + retain_fixture=False, + ) + + if __name__ == "__main__": unittest.main()