Merge pull request 'Enforce agent CLI execution routing' (#572) from feature/agent-cli-enforcement into main

AGENT-EXECUTION.md

@@ -7,11 +7,19 @@ parser accepts exactly one `agent-execution` fenced block.
 
 U3's execution controller and U4's project wiring are integrated. This file is
 the execution-routing authority for implementation, repair, test, and review.
-Active routes are intentionally limited to Cursor, Antigravity (`agy`), and
+Normal implementation, repair, and test routes stay Cursor, Antigravity
-PI. `model: auto` lets each CLI use its configured default model; PI therefore
+(`agy`), and PI. Each of those profiles may be attempted up to
-uses the operator-selected MiniMax default. A provider that makes no successful
+`max_attempts_per_profile`, which is exactly 3 whenever `emergency_profile`
-stdin write or stdout/stderr read for `idle_timeout_seconds` is stopped before
+is configured. Only after those candidates safely exhaust their attempts does
-`timeout_seconds`.
+the enabled emergency `codex` profile run once. Codex remains out of the three
+normal role lists and may also appear in review. `fallback_to_host` is
+`never`: the controller does not emit `host-fallback-authorized` and does not
+increment host fallback authorizations. Safe exhaustion without an emergency
+candidate ends `unavailable` or `blocked`, and that result does not authorize
+the host to write the unit. The legacy value `all-unavailable` remains
+parseable and likewise grants no host write authority. Cursor uses
+`cursor-agent-stable`. `model: auto` lets each CLI use its configured default
+model; PI therefore uses the operator-selected MiniMax default.
 
 ```agent-execution
 schema: agent-execution/v1
@@ -19,23 +27,25 @@ schema: agent-execution/v1
 execution:
   fresh_process_per_unit: true
   permission: run-everything
-  fallback_to_host: all-unavailable
+  fallback_to_host: never
   max_parallel: 3
   timeout_seconds: 7200
   idle_timeout_seconds: 600
   summary_bytes: 16384
   raw_output_bytes: 10485760
+  max_attempts_per_profile: 3
+  emergency_profile: codex
 
 roles:
   implementation: [cursor, antigravity, pi]
   repair: [cursor, antigravity, pi]
   test: [pi, antigravity, cursor]
-  review: [pi, antigravity, cursor]
+  review: [pi, antigravity, cursor, codex]
 
 profiles:
   cursor:
     adapter: cursor
-    executable: cursor-agent
+    executable: cursor-agent-stable
     model: auto
     enabled: true
 
@@ -50,4 +60,10 @@ profiles:
     executable: pi
     model: auto
     enabled: true
+
+  codex:
+    adapter: codex
+    executable: codex
+    model: auto
+    enabled: true
 ```

AGENTS.md

@@ -20,9 +20,10 @@ policy/delivery.md。
   低影响噪音直接忽略；Review/迭代循环只围绕阻塞项，阻塞项闭合即结束。
 - 测试与风险成比例：实施中只跑最小直接检查；完整/长耗时套件仅在最终候选、规范明确要求或
   实际影响面确有必要时运行一次。禁止在每个小文档、配置或 adapter 修改后重复全量测试。
-- 编排成本不得超过任务本身。平凡文档/配置单元最多使用一个有界 CLI 进程；除非用户明确要求
+- 编排成本不得超过任务本身。实施、修复与测试按 AGENT-EXECUTION.md 的候选和尝试上限调度；
-  或已有证据证明是当前目标的直接阻塞，否则不得新增嵌套控制器、额外 worktree、复杂收据体系
+  不限制单次 CLI 的推理时长、工具次数、规划方式或首次写文件时间。除非用户明确要求或已有证据
-  或新架构。执行路由是手段，不能成为推迟用户产物的理由。
+  证明是当前目标的直接阻塞，否则不得新增嵌套控制器、额外 worktree、复杂收据体系或新架构。
+  执行路由是手段，不能成为推迟用户产物的理由。
 
 ## 任务目标与范围（强制）
 
@@ -41,11 +42,27 @@ policy/delivery.md。
 - 产生仓库变更的实施按可独立实施的单元拆分；任务卡记 `实施执行：agent-per-unit（配置：AGENT-EXECUTION.md）`。
 - 外部 Agent 在新进程/隔离工作区执行一个有界单元；host 负责计划、拆解、范围审阅、权威验证、规范提交、
   PR/CI/合入、证据与任务元数据及安全清理。Worker 不得提交、推送、开或合 PR、部署、改路由或派遣同伴。
-- 当前 host 仅在 `scripts/agent_exec.py run` 对该精确单元发出有效 `host-fallback-authorized` 收据后，
+- 实施、修复与测试的正常候选只有 Cursor、Antigravity（`agy`）和 PI，顺序以配置为准。每个候选对
-  方可撰写该单元的仓库变更；`failed-unknown`、`scope-expansion`、无效配置、未对账 Git 状态与安全失败
+ 安全可恢复失败（`unavailable`，或外部无副作用已被机制证明的 `failed-clean`）最多 3 次全新尝试，然后才换下一候选。
-  均不授权回退。适配器、终态与回退细则见配置及 [policy/delivery.md](policy/delivery.md)、
+ 单元包的 `network=false`/`external_cost=false` 只是声明，不是证明；没有强制隔离或可信对账时，
-  [policy/governance.md](policy/governance.md)。不削弱 goal-first、严格范围、先可用、独立 Review、
+ 已启动的失败尝试即使私有克隆可精确恢复也记 `failed-unknown`。成功还须从该 CLI 实际事件信封解出
-  证据、平台准入与高后果双钥匙规则。
+ 有效的 `completed` 终态记录（schema 写在 worker 提示词中）；缺失或无效即 fail closed。
+  三者都安全耗尽后，才允许一次全新的 Codex CLI 紧急尝试。Codex 平时只用于规划、方案设计、实施计划
+  或独立 Review，不进入这三个正常角色。`fallback_to_host` 为 `never`。安全耗尽不发出
+  host-fallback-authorized，终态为 unavailable 或 blocked，不授权 host 撰写该单元。旧值
+  `all-unavailable` 仍可解析，同样不产生 host 撰写授权。
+- 显式 fail-stop 立即停止，不重试、不换候选、不启动紧急尝试：`failed-unknown`、`scope-expansion`、
+  `blocked`、`cancelled`、无效配置、未对账 Git 状态与安全失败。总超时，以及 stdin/stdout/stderr
+  都没有活动的空闲超时，归入 `failed-unknown`。
+- 一次尝试只在显式终态失败、进程退出、总超时或上述空闲超时时结束。CLI 可在目标、范围与安全边界内
+  自主检查、计划、测试并迭代。
+- 实施变更只经成功 transport 由 `scripts/agent_exec.py apply` 集成，或由 pre-commit `admit` 按精确内容
+ 做本地完整性核对。本地 HMAC 历史与该钩子只是本地完整性/篡改证据（SELF），不是 GATE、可信身份，
+ 也不防同一主机 OS 用户；本地记录缺失或无效仍拒绝，交付信任只来自精确 HEAD 的 CI 或独立 REPLAY。
+ host 直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**` 与
+  `docs/superpowers/plans/**`。host 不直接撰写实施单元。细则见
+  [policy/delivery.md](policy/delivery.md) 与 [policy/governance.md](policy/governance.md)。
+  不削弱 goal-first、严格范围、先可用、独立 Review、证据、平台准入与高后果双钥匙规则。
 
 ### 阶段与会话隔离（强制）
 

Makefile

@@ -13,7 +13,7 @@ PLUGINS    := plugins
 COMBO      := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null)
 VERSION    := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO)
 
-.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate
+.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit
 
 version:
 	@echo "agent-system $(VERSION)"
@@ -78,10 +78,19 @@ help:
 	@echo "make pin         推进 submodule 指针到各仓远端 main（提交前须 make test 通过）"
 	@echo "make status      submodule 状态"
 	@echo "make agent-exec-validate  校验 AGENT-EXECUTION.md（agent_exec validate）"
+	@echo "make agent-exec-apply RECEIPT=<path>  快进当前分支到成功收据的 transport"
+	@echo "make agent-exec-admit  校验暂存路径具备 transport 来源或属于 host 豁免"
 
 agent-exec-validate:
 	PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo .
 
+agent-exec-apply:
+	@if [ -z "$(RECEIPT)" ]; then echo "usage: make agent-exec-apply RECEIPT=<path>"; exit 2; fi
+	PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py apply --repo . --receipt "$(RECEIPT)"
+
+agent-exec-admit:
+	PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py admit --repo .
+
 sync:
 	git submodule update --init --recursive
 

policy/adoption.md

@@ -18,7 +18,7 @@
 | standards/verification.md | 验收 | 本地 make test-governance + 组合测试；当前证据与历史 scratch 记录分开 |
 | standards/multi-repo.md | 跨仓/gitlink | manual；受控历史迁移通道已于 2026-09-30 行使一次（documents/records/2026-09-30-history-migration.md） |
 | 工作区回收 | 交付终点 | advisory（task-finish）；负例：合入前 finish 拒绝输出 |
-| policy/delivery.md（项目收紧） | 需求、长期方向、试验、实施迭代及交付 | goal-first：目标/范围/终点一次确认，范围内连续推进验证、Review、合入和安全收口；实施写入按 agent-per-unit（配置 AGENT-EXECUTION.md；外部 Agent 一单元一新进程；host 仅在有效 host-fallback-authorized 收据后可写实施变更）；仅真实决策点需人工；任务卡提供记录，post-merge 默认只提示，行为回归见 tests/test_task.py |
+| policy/delivery.md（项目收紧） | 需求、长期方向、试验、实施迭代及交付 | goal-first：目标/范围/终点一次确认，范围内连续推进验证、Review、合入和安全收口；实施写入按 agent-per-unit（配置 AGENT-EXECUTION.md；实施/修复/测试正常候选仅为 Cursor、agy、PI，每候选对安全可恢复失败最多 3 次全新尝试，全部安全耗尽后一次 Codex CLI 紧急尝试；Codex 平时只用于规划、设计、计划或独立 Review；显式 fail-stop 立即停止；不限制推理时长、工具次数、规划方式或首次写文件时间；集成须成功 transport 的 agent-exec apply，或 pre-commit admit 精确内容来源；host 豁免仅为 .task/** 与 docs/superpowers/specs/**、plans/**；host 不直接实施；fallback_to_host 为 never；安全耗尽不发出 host-fallback-authorized，终态为 unavailable 或 blocked）；仅真实决策点需人工；任务卡提供记录，post-merge 默认只提示，行为回归见 tests/test_task.py |
 | AGENT-EXECUTION.md + agent-exec | 项目执行路由 | 本地 SELF：parser/adapter/controller 假 CLI 契约由 tests/test_agent_exec*.py 与 make agent-exec-validate 强制；已安装 provider/model 真实身份待 U5 探针，此前不接受为已就位；远端平台准入仍为 GATE |
 | standards/tool-adaptation.md（draft.8） | 为工具创建规则/钩子实例 | 本组合 CLI/git 级实例已登记：policy/tool-adaptation-matrix.md（A1–A7 逐行定级+正负例证据） |
 | 同上（zcode 实例） | zcode 客户端写路径 | policy/tool-adaptation-zcode.md：PreToolUse 写守卫；脚本负例与客户端接线分开，客户端 pending-live |

policy/delivery.md

@@ -20,8 +20,9 @@
 - Markdown、配置、提示词、链接和说明等简单修改默认走最小闭环：定位 → 修改 → 最小直接验证
   → 报告。不得为此新建平台、框架、自动化系统、嵌套控制器、额外 worktree、复杂收据体系、
   新 PR 模板或新的阶段审批。
-- Agent 路由、任务卡和证据工具都是交付手段，不是交付目标。候选 CLI 迟迟不产出时按配置及时
+- Agent 路由、任务卡和证据工具都是交付手段，不是交付目标。推理时间、工具次数、规划方式和尚未写文件
-  切换或按既有回退规则处理；不得转而长期建设或加固路由系统，导致用户要求的简单产物一直未交付。
+  都不结束一次尝试。尝试只在显式终态失败、进程退出、总超时，或 stdin/stdout/stderr 都没有活动的
+  空闲超时时结束。不得转而长期建设或加固路由系统，导致用户要求的简单产物一直未交付。
 
 ### 最小可用即停，问题按是否阻塞分流
 
@@ -38,8 +39,9 @@
   本仓规范明确要求、真实影响面覆盖广，或最小测试不足以判断回归的情况。
 - 禁止在每个小文档、配置、提示词或 adapter 修改后重复完整套件；已经针对同一候选取得且对象
   未变化的有效结果不得因记录、格式或非阻塞意见重复执行。
-- 编排成本不得超过任务本身。平凡单元最多启动一个有界 CLI 进程；主模型负责计划、范围与审查，
+- 编排成本不得超过任务本身。实施、修复与测试按 AGENT-EXECUTION.md 的候选和每候选最多 3 次全新尝试
-  但角色分工不得推迟简单产物。检测到实现粒度与任务不匹配时，立即降级到最小充分做法。
+  调度；主模型负责计划、范围与审查，角色分工不得推迟简单产物。检测到实现粒度与任务不匹配时，
+  立即降级到最小充分做法。不限制单次 CLI 的推理时长、工具次数、规划方式或首次写文件时间。
 - 移除机械人工停顿不等于增加机械 Agent 停顿。真实业务取舍、授权变化、高后果操作和必须由用户
   判定的验收仍请求人工；除此之外，不为计划、工具、自审或非阻塞完善反复等待。
 
@@ -69,7 +71,31 @@
 产生仓库变更的实施工作拆成可独立实施的单元。任务卡默认记
 `实施执行：agent-per-unit（配置：AGENT-EXECUTION.md）`。执行路由以根目录
 [AGENT-EXECUTION.md](../AGENT-EXECUTION.md) 为唯一权威；按角色声明的候选顺序在新进程与
-隔离工作区中尝试外部 Agent。不得跨单元复用同一外部进程。
+隔离工作区中尝试外部 Agent。不得跨单元复用同一外部进程，也不得 resume 同一次尝试。
+
+实施、修复与测试的正常候选只有 Cursor、Antigravity（`agy`）和 PI。每个候选对安全可恢复失败
+最多 3 次全新尝试：`unavailable`，或外部无副作用已被机制证明的 `failed-clean`。三次仍属这类失败后才进入
+下一候选。单元包的 `network=false` 与 `external_cost=false` 只是声明，不证明外部干净；当前没有强制的
+网络/外部费用隔离或可信对账，所以已启动的失败尝试即使私有克隆可精确恢复，也记为 `failed-unknown`
+并保留现场。三个正常候选都安全耗尽后，才允许一次全新的 Codex CLI 紧急尝试；这次不按每候选 3 次计算。
+紧急尝试若仍是安全可恢复失败，运行停在 `blocked`。host 不接过该单元继续撰写。
+
+Codex 平时只用于规划、方案设计、实施计划，或独立 Review。执行配置没有单独的 plan 角色；
+上述规划与设计工作可以直接使用 Codex CLI。`review` 角色的候选可以包含 Codex，并按该角色自己的
+顺序和尝试上限执行。Codex 不出现在实施、修复、测试的正常候选里；这三个角色上的一次紧急尝试是唯一例外。
+
+显式 fail-stop 立即停止整次运行，不重试、不换候选、不启动紧急尝试：`failed-unknown`、
+`scope-expansion`、`blocked`、`cancelled`、无效配置、未对账的 Git 状态与安全失败。总超时，
+以及 stdin/stdout/stderr 都没有活动的空闲超时，归入 `failed-unknown`。
+
+一次尝试只在显式终态失败、进程退出、总超时或该空闲超时时结束。推理时长、工具次数、规划方式和
+首次写文件时间都不是结束条件。CLI 可在该单元的目标、范围与安全边界内自主检查、计划、测试并迭代。
+
+worker 提示词写明终态记录的 JSON Schema（`terminal_status`、`summary`、`changed_files`、`evidence`、
+`scope_expansion`）。控制器只从该 CLI 实际事件信封解码：Cursor `stream-json` 的 `result` 事件、
+agy `stream-json` 的 `result` 事件或 `json` 输出对象、PI `json` 模式最后一条 assistant `message_end`。
+退出码 0 且有改动时，只有有效的 `completed` 记录才可能成功；记录缺失、无效或仅有散文均记
+`failed-unknown`。`blocked`、`cancelled`、`failed-unknown` 记录直接按该终态停止，不得成功。
 
 主模型（host）负责计划、拆解、有界单元包、强制范围、检查与审阅实际 diff、运行权威验证、
 规范提交、管理 PR/CI/合入、写入证据与任务管理元数据，以及安全清理。权威验证、集成、合入与
@@ -82,11 +108,23 @@
 `run-everything` 仅在隔离单元内取消逐步命令批准提示；不扩大用户授权范围，也不授予 worker
 规范 Git、推送、PR、合入、部署或清理权威。
 
-当前 host **仅当** `scripts/agent_exec.py run` 对该精确单元发出有效
+实施结果进入当前分支，必须来自一次成功 transport，由 host 执行
-`host-fallback-authorized` 收据后，方可直接撰写该单元的仓库变更。全部候选须已安全耗尽
+`scripts/agent_exec.py apply` 在干净分支上快进；或者本次暂存内容通过 pre-commit 调用的
-（`unavailable` 或已精确恢复的 `failed-clean`）。`failed-unknown`、`scope-expansion`、无效配置、
+`admit`：非豁免路径与该成功 transport 的路径和精确内容一致，且 transport 的父提交是当前 HEAD。
-未对账 Git 状态与安全失败均不授权回退。规划、只读调查、证据捕获、控制器/任务卡元数据与集成
+apply/admit 依赖的本地 HMAC 执行历史与 checkout 钩子只是本地完整性/篡改证据（SELF）：同一主机
-操作属于 host 管理工作，不递归创建另一实施单元。
+OS 用户可读取密钥或绕过钩子，因此它不是 GATE、可信身份或防恶意主体的保护。本地记录缺失或无效
+仍拒绝；交付信任只来自精确 HEAD 的必需 CI（GATE）或独立 REPLAY。
+host 可以直接编写并提交的豁免只有 `.task/**`、`docs/superpowers/specs/**` 和
+`docs/superpowers/plans/**`。除此之外，host 不直接撰写实施单元。
+
+安全耗尽不发出 `host-fallback-authorized`，`host_fallback_authorizations` 保持为 0。
+未配置紧急候选时终态为 unavailable 或 blocked，不授权 host 撰写实施单元。现行
+[AGENT-EXECUTION.md](../AGENT-EXECUTION.md) 的 `fallback_to_host` 是 `never`。旧值
+`all-unavailable` 仍可解析，但同样不产生 host 撰写授权。已归档任务卡里的旧表述保持原样，
+不改写成现行规则。
+
+规划、只读调查、证据捕获、任务卡与 superpowers 规格/计划，以及 apply 或 admit 通过之后的集成、
+验证、提交和 PR/CI/合入，属于 host 管理工作，不递归创建另一实施单元。
 
 本条不重新引入逐阶段人工门禁；goal-first 的决策点、严格范围、先可用再迭代、独立 Review、
 证据、平台准入与高后果双钥匙规则继续有效。

policy/governance.md

@@ -8,17 +8,29 @@
 智能体审阅者及对象版本；权威验证与合入仍由 host/主模型执行，worker 局部自检不替代。
 本地 evidence 仅记录 SELF 命令，不能证明授权/人工决策或提升为 CI/独立审阅证据。
 
-### Agent 实施收据与回退
+### Agent 实施收据与来源
 
 每次外部实施尝试须在任务卡「Agent 实施收据」记录：配置摘要（`AGENT-EXECUTION.md` digest）、
-角色、声明候选顺序、attempt、CLI/模型、终态、transport、fallback，以及源码外原始收据路径。
+角色、声明候选顺序、attempt、CLI/模型、终态、transport、fallback（只表示该行是否为 Codex
-初始生成状态为 `NOT_RUN`。原始事件流与超限输出保留在源码外；未知终态（`failed-unknown`）
+紧急尝试），以及源码外原始收据路径。初始生成状态为 `NOT_RUN`。原始事件流与超限输出保留在
-与 `scope-expansion` 的隔离工作区须保留待对账，不得静默丢弃。清理须有精确恢复或保留现场的
+源码外；未知终态（`failed-unknown`）与 `scope-expansion` 的隔离工作区须保留待对账，不得静默丢弃。
-可复核证据。
+清理须有精确恢复或保留现场的可复核证据。
 
-当前 host 仅在 `scripts/agent_exec.py run` 对该精确单元发出有效 `host-fallback-authorized`
+实施、修复与测试的正常候选只有 Cursor、Antigravity（`agy`）和 PI。每个候选对安全可恢复失败
-收据后，方可撰写该单元的仓库变更；该收据不得由叙述或局部失败推断。`failed-unknown`、
+最多 3 次全新尝试，全部安全耗尽后才允许一次 Codex CLI 紧急尝试。Codex 平时只用于规划、方案设计、
-`scope-expansion`、无效配置、未对账 Git 状态与安全失败均不授权回退。
+实施计划或独立 Review。`fallback_to_host` 为 `never`。安全耗尽不发出 host-fallback-authorized，
+终态为 unavailable 或 blocked，不授权 host 撰写实施单元。显式 fail-stop 立即停止。一次尝试只在显式终态失败、进程退出、总超时，
+或 stdin/stdout/stderr 都没有活动的空闲超时时结束；推理时长、工具次数、规划方式和首次写文件时间
+都不是结束条件。CLI 可在目标、范围与安全边界内自主检查、计划、测试并迭代。路由细则见
+[迭代交付](delivery.md)。
+
+实施变更进入仓库，只通过成功 transport 的 `scripts/agent_exec.py apply`，或 pre-commit `admit`
+对暂存内容的本地完整性核对。本地 HMAC 执行历史与该钩子只记作本地完整性/篡改证据（SELF），
+不是 GATE、可信身份，也不防同一主机 OS 用户；本地记录缺失或无效仍拒绝，交付信任来自精确 HEAD 的
+CI 或独立 REPLAY。host 可以直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**`
+与 `docs/superpowers/plans/**`。host 不直接撰写实施单元。`failed-unknown`、`scope-expansion`、
+无效配置、未对账 Git 状态与安全失败都不授权继续路由或代写。`network=false`/`external_cost=false`
+声明不证明外部干净，已启动的失败尝试在没有强制隔离或可信对账时记 `failed-unknown`。
 
 ## 启动与恢复
 

policy/tool-adaptation-matrix.md

@@ -14,7 +14,7 @@
 | A5 回收 | 本地命令检查；写者退出须可证明，无法证明时才请求 owner 决策 | 登记根目录、锁/操作状态、父子仓与唯一文件、合入核验 | `tests/test_task.py` |
 | A6 秘密 | 本地脱敏及 CI 获取边界；权限与宿主隔离另验 | 命令/输出/摘要脱敏，凭据不持久写 Git 配置 | `tests/test_evidence.py`、`tests/test_ci_credentials.py` |
 | A7 证据 | 本地 SELF 收据；外部来源独立核验 | 首次捕获、完整脱敏输出/摘要、对象记录、原样导入 | `tests/test_evidence.py`；CI/审阅链接另核 |
-| A8 阶段记录与迭代 | goal-first + agent-per-unit（AGENT-EXECUTION.md）；模板默认连续执行，仅真实决策点需人工，没有全流程授权强制；实施写入经配置候选在新进程/隔离工作区按单元执行；host 回退仅凭有效 host-fallback-authorized 收据 | policy/delivery.md、AGENT-EXECUTION.md；任务卡分别记录阶段状态、决策点、Agent 实施收据（配置摘要/角色/候选/attempt/CLI-模型/终态/transport/fallback/源码外路径）、Review、冲突、当前 HEAD 的 CI 和合入；post-merge 默认仅提示 | `tests/test_task.py` 核对 goal-first/agent-per-unit 模板及钩子不自动写入；`python3 scripts/agent_exec.py validate` / parser·adapter·controller 假测为本地 SELF 强制行为；已安装 provider/model 身份待 U5，远端平台准入为 GATE；智能体核对授权范围、实施单元边界、Review、冲突及匹配 HEAD 的 CI/平台准入 |
+| A8 阶段记录与迭代 | goal-first + agent-per-unit（AGENT-EXECUTION.md）；模板默认连续执行，仅真实决策点需人工，没有全流程授权强制；实施/修复/测试仅 Cursor、agy、PI，每候选对安全可恢复失败最多 3 次全新尝试，全部安全耗尽后一次 Codex CLI 紧急尝试；Codex 平时仅规划、设计、计划或独立 Review；显式 fail-stop 立即停止；不限制推理时长、工具次数、规划方式或首次写文件时间；集成经成功 transport 的 agent-exec apply，或 pre-commit admit 精确内容来源；host 豁免仅为 .task/** 与 docs/superpowers/specs/**、plans/**；host 不直接实施；fallback_to_host 为 never；安全耗尽不发出 host-fallback-authorized，终态为 unavailable 或 blocked | policy/delivery.md、AGENT-EXECUTION.md；任务卡分别记录阶段状态、决策点、Agent 实施收据（配置摘要/角色/候选/attempt/CLI-模型/终态/transport/紧急尝试/源码外路径）、Review、冲突、当前 HEAD 的 CI 和合入；post-merge 默认仅提示 | `tests/test_task.py` 核对 goal-first/agent-per-unit 模板、正常路由、尝试上限、fail-stop、空闲超时与 apply/admit 来源表述，及钩子不自动写入；`python3 scripts/agent_exec.py validate` / parser·adapter·controller 假测为本地 SELF 强制行为；已安装 provider/model 身份待 U5，远端平台准入为 GATE；智能体核对授权范围、实施单元边界、Review、冲突及匹配 HEAD 的 CI/平台准入 |
 | A9 长期方向与后续任务 | manual；模板不校验规划有效性或自动调度任务 | policy/delivery.md；PROJECT.md 提供规划入口，任务卡关联规划/里程碑并记录下一步建议 | `tests/test_task.py` 核对生成字段与未授权默认值；智能体核对当前规划、进展、重复任务和建议依据，新任务仍不自动获得授权 |
 | A10 交付窗口 | manual；无自动互斥或接管 | [合入调度](merge-queue.md)与 #518：同一目标一个最终窗口，等待分支不追 main；轮到后同步、固定、验证并由智能体合入 | 每个实际 PR 核对窗口记录、精确 HEAD 三项 GATE、Review 和合入后 main；不要求额外试行或全写者审计 |
 | A11 固定后证据 | manual 选择源码外位置；工具捕获绝对未用路径/原始输出；无 OS 冻结 | policy/governance.md；RECEIPT 捕获，后续授权闭合再原样导入 | `tests/test_evidence.py` 固定对象、权限、覆盖拒绝、原样导入与幂等 |

scripts/agent_exec_adapters.py

@@ -5,11 +5,14 @@ does not contact providers or launch real agent runs.
 """
 from __future__ import annotations
 
+import json
 import re
 import shutil
 import subprocess
+from collections.abc import Sequence
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Any
 
 from scripts.agent_exec_config import (
     ALLOWED_PLACEHOLDERS,
@@ -23,6 +26,34 @@ _PROBE_TIMEOUT_SECONDS = 5
 MAX_UNIT_PACKET_BYTES = 262_144
 UNIT_PACKET_DISPLAY_MARKER = "<unit-packet>"
 
+TERMINAL_STATUSES = (
+    "completed",
+    "failed",
+    "blocked",
+    "cancelled",
+    "failed-unknown",
+    "scope-expansion",
+)
+TERMINAL_RECORD_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "required": [
+        "terminal_status",
+        "summary",
+        "changed_files",
+        "evidence",
+        "scope_expansion",
+    ],
+    "properties": {
+        "terminal_status": {"enum": list(TERMINAL_STATUSES)},
+        "summary": {"type": "string"},
+        "changed_files": {"type": "array", "items": {"type": "string"}},
+        "evidence": {"type": "array", "items": {"type": "string"}},
+        "scope_expansion": {"type": ["string", "null"]},
+    },
+}
+# Bounds the backwards brace scan over one final message.
+_MAX_TERMINAL_TEXT_CHARS = 262_144
+
 _PLACEHOLDER_MAP = {
     "${workspace}": "workspace",
     "${unit_packet}": "unit_packet",
@@ -378,17 +409,194 @@ def _read_unit_packet_text(path: Path) -> str:
 
 
 def _build_bounded_unit_prompt(packet_text: str) -> str:
+    schema = json.dumps(TERMINAL_RECORD_SCHEMA, separators=(",", ":"))
     return (
         "Implement exactly the supplied bounded implementation unit. "
         "Work only inside the current workspace. "
         "Leave all changes uncommitted. "
         "Do not expand scope beyond the unit packet. "
         "Return a structured final result matching the unit's required schema.\n\n"
+        "--- TERMINAL RECORD ---\n\n"
+        "End your final message with exactly one terminal record: a single-line "
+        "JSON object that validates against this JSON Schema:\n"
+        f"{schema}\n"
+        "Use terminal_status completed only when the unit is fully implemented. "
+        "Use blocked, cancelled, failed-unknown, failed, or scope-expansion "
+        "otherwise. A missing or invalid terminal record fails the attempt.\n\n"
         "--- BOUNDED IMPLEMENTATION UNIT PACKET ---\n\n"
         f"{packet_text}"
     )
 
 
+def extract_terminal_record(text: str | None) -> dict[str, Any] | None:
+    """Return the sole terminal record that ends one final message.
+
+    Only whitespace may follow the record, and no other valid record may
+    precede it; anything else fails closed.
+    """
+    if not isinstance(text, str) or not text:
+        return None
+    text = text[-_MAX_TERMINAL_TEXT_CHARS:].rstrip()
+    decoder = json.JSONDecoder()
+
+    def decode_at(pos: int) -> tuple[object, int]:
+        try:
+            return decoder.raw_decode(text, pos)
+        except json.JSONDecodeError:
+            return None, -1
+
+    pos = text.rfind("{")
+    while pos != -1:
+        candidate, end = decode_at(pos)
+        if end == len(text) and isinstance(candidate, dict):
+            break
+        pos = text.rfind("{", 0, pos)
+    else:
+        return None
+    record = _valid_terminal_record(candidate)
+    if record is None:
+        return None
+    earlier = text.rfind("{", 0, pos)
+    while earlier != -1:
+        if _valid_terminal_record(decode_at(earlier)[0]) is not None:
+            return None
+        earlier = text.rfind("{", 0, earlier)
+    return record
+
+
+def _valid_terminal_record(obj: object) -> dict[str, Any] | None:
+    if not isinstance(obj, dict):
+        return None
+    status = obj.get("terminal_status")
+    if not isinstance(status, str):
+        return None
+    status = status.strip().replace("_", "-")
+    if status not in TERMINAL_STATUSES:
+        return None
+    if not isinstance(obj.get("summary"), str):
+        return None
+    for key in ("changed_files", "evidence"):
+        items = obj.get(key)
+        if not isinstance(items, list) or any(not isinstance(i, str) for i in items):
+            return None
+    if "scope_expansion" not in obj or not (
+        obj["scope_expansion"] is None or isinstance(obj["scope_expansion"], str)
+    ):
+        return None
+    record = dict(obj)
+    record["terminal_status"] = status
+    return record
+
+
+def decode_terminal_record(
+    adapter: str,
+    *,
+    event_lines: Sequence[str | None],
+    stdout_tail: str,
+    result_text: str | None,
+) -> dict[str, Any] | None:
+    """Decode the worker terminal record from the adapter's real output.
+
+    ``event_lines`` are recent complete stdout lines that may carry a terminal
+    envelope, oldest first; ``None`` marks an envelope line too large to keep.
+    Event-stream adapters are read only from their final assistant/result
+    envelope, so prompt echoes and tool output never count as a record.
+    """
+    if adapter in ("cursor", "claude"):
+        return extract_terminal_record(_cursor_final_text(event_lines))
+    if adapter == "antigravity":
+        return extract_terminal_record(_agy_final_text(event_lines, stdout_tail))
+    if adapter == "pi":
+        return extract_terminal_record(_pi_final_text(event_lines))
+    record = extract_terminal_record(result_text)
+    if record is None:
+        record = extract_terminal_record(stdout_tail)
+    return record
+
+
+def _event_objects(event_lines: Sequence[str | None]) -> list[dict[str, Any] | None]:
+    """Newest-first parsed events; ``None`` keeps the oversized-envelope marker."""
+    out: list[dict[str, Any] | None] = []
+    for line in reversed(event_lines):
+        if line is None:
+            out.append(None)
+            continue
+        try:
+            obj = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(obj, dict):
+            out.append(obj)
+    return out
+
+
+def _cursor_final_text(event_lines: Sequence[str | None]) -> str | None:
+    for obj in _event_objects(event_lines):
+        if obj is None:
+            return None
+        if obj.get("type") != "result":
+            continue
+        if obj.get("is_error") is not False or obj.get("subtype") != "success":
+            return None
+        text = obj.get("result")
+        return text if isinstance(text, str) else None
+    return None
+
+
+def _agy_final_text(
+    event_lines: Sequence[str | None], stdout_tail: str
+) -> str | None:
+    candidates = _event_objects(event_lines)
+    try:
+        whole = json.loads(stdout_tail.strip())
+    except json.JSONDecodeError:
+        whole = None
+    if isinstance(whole, dict):
+        candidates.append(whole)
+    for obj in candidates:
+        if obj is None:
+            return None
+        if obj.get("event") == "result":
+            payload = obj.get("result")
+        elif "event" not in obj and "response" in obj:
+            payload = obj
+        else:
+            continue
+        if not isinstance(payload, dict):
+            return None
+        if payload.get("is_error") or payload.get("error"):
+            return None
+        text = payload.get("response")
+        return text if isinstance(text, str) else None
+    return None
+
+
+def _pi_final_text(event_lines: Sequence[str | None]) -> str | None:
+    for obj in _event_objects(event_lines):
+        if obj is None:
+            return None
+        if obj.get("type") != "message_end":
+            continue
+        message = obj.get("message")
+        if not isinstance(message, dict) or message.get("role") != "assistant":
+            continue
+        if message.get("stopReason") in ("error", "aborted"):
+            return None
+        content = message.get("content")
+        if isinstance(content, str):
+            return content
+        if not isinstance(content, list):
+            return None
+        return "".join(
+            block["text"]
+            for block in content
+            if isinstance(block, dict)
+            and block.get("type") == "text"
+            and isinstance(block.get("text"), str)
+        )
+    return None
+
+
 def _expand_argv(
     items: tuple[str, ...],
     values: dict[str, str],
@@ -480,11 +688,15 @@ def _sanitize_display(
 
 __all__ = [
     "MAX_UNIT_PACKET_BYTES",
+    "TERMINAL_RECORD_SCHEMA",
+    "TERMINAL_STATUSES",
     "UNIT_PACKET_DISPLAY_MARKER",
     "AdapterProbe",
     "AdapterSpec",
     "BUILTIN_SPECS",
     "Invocation",
     "build_invocation",
+    "decode_terminal_record",
+    "extract_terminal_record",
     "probe_adapter",
 ]

scripts/agent_exec_config.py

@@ -23,7 +23,11 @@ ALLOWED_PLACEHOLDERS = frozenset(
 PLACEHOLDER_RE = re.compile(r"\$\{[A-Za-z0-9_]+\}")
 
 PERMISSION_VALUES = frozenset({"run-everything"})
-FALLBACK_VALUES = frozenset({"all-unavailable"})
+FALLBACK_VALUES = frozenset({"all-unavailable", "never"})
+# Implementation/repair/test roles each need exactly one enabled profile per
+# adapter here, with or without emergency_profile; review is unrestricted.
+NORMAL_ROLE_ADAPTERS = frozenset({"cursor", "antigravity", "pi"})
+MAX_ATTEMPTS_PER_PROFILE = 3
 PROMPT_TRANSPORTS = frozenset({"stdin", "positional", "flag"})
 OUTPUT_FORMATS = frozenset({"json", "stream-json", "jsonl", "text"})
 
@@ -54,9 +58,13 @@ EXECUTION_KEYS = frozenset(
         "idle_timeout_seconds",
         "summary_bytes",
         "raw_output_bytes",
+        "max_attempts_per_profile",
+        "emergency_profile",
     }
 )
-EXECUTION_REQUIRED = EXECUTION_KEYS - frozenset({"idle_timeout_seconds"})
+EXECUTION_REQUIRED = EXECUTION_KEYS - frozenset(
+    {"idle_timeout_seconds", "max_attempts_per_profile", "emergency_profile"}
+)
 PROFILE_KEYS = frozenset({"adapter", "executable", "model", "enabled"})
 CUSTOM_ADAPTER_KEYS = frozenset(
     {
@@ -84,6 +92,8 @@ class ExecutionPolicy:
     summary_bytes: int
     raw_output_bytes: int
     idle_timeout_seconds: int = DEFAULT_IDLE_TIMEOUT_SECONDS
+    max_attempts_per_profile: int = 1
+    emergency_profile: str | None = None
 
 
 @dataclass(frozen=True)
@@ -337,6 +347,10 @@ def _build_config(raw: dict[str, Any]) -> ExecutionConfig:
             if profile.adapter not in custom_adapters:
                 raise ConfigError(f"unknown adapter: {profile.adapter!r}")
 
+    _validate_emergency_profile(execution, profiles, roles)
+    _validate_normal_role_adapters(profiles, roles)
+    _validate_normal_role_exactness(profiles, roles)
+
     return ExecutionConfig(
         schema=schema,
         execution=execution,
@@ -391,6 +405,23 @@ def _build_execution(raw: Any) -> ExecutionPolicy:
     if raw_output_bytes < summary_bytes:
         raise ConfigError("raw_output_bytes must be >= summary_bytes")
 
+    if "max_attempts_per_profile" in raw:
+        max_attempts_per_profile = _positive_bounded(
+            raw["max_attempts_per_profile"],
+            "max_attempts_per_profile",
+            1,
+            MAX_ATTEMPTS_PER_PROFILE,
+        )
+    else:
+        max_attempts_per_profile = 1
+
+    if "emergency_profile" in raw:
+        emergency_profile = raw["emergency_profile"]
+        if not isinstance(emergency_profile, str) or not emergency_profile:
+            raise ConfigError("emergency_profile must be a non-empty string")
+    else:
+        emergency_profile = None
+
     return ExecutionPolicy(
         fresh_process_per_unit=True,
         permission=permission,
@@ -400,9 +431,79 @@ def _build_execution(raw: Any) -> ExecutionPolicy:
         summary_bytes=summary_bytes,
         raw_output_bytes=raw_output_bytes,
         idle_timeout_seconds=idle_timeout_seconds,
+        max_attempts_per_profile=max_attempts_per_profile,
+        emergency_profile=emergency_profile,
     )
 
 
+def _validate_emergency_profile(
+    execution: ExecutionPolicy,
+    profiles: Mapping[str, Profile],
+    roles: Mapping[str, tuple[str, ...]],
+) -> None:
+    name = execution.emergency_profile
+    if name is None:
+        return
+    if execution.max_attempts_per_profile != 3:
+        raise ConfigError(
+            "max_attempts_per_profile must equal 3 when emergency_profile is configured"
+        )
+    profile = profiles.get(name)
+    if profile is None:
+        raise ConfigError(f"emergency_profile {name!r} does not exist")
+    if not profile.enabled:
+        raise ConfigError(f"emergency_profile {name!r} must be enabled")
+    if profile.adapter != "codex":
+        raise ConfigError(f"emergency_profile {name!r} must use adapter codex")
+    for role in ("implementation", "repair", "test"):
+        if name in roles[role]:
+            raise ConfigError(
+                f"emergency_profile {name!r} must not appear in role {role!r}"
+            )
+
+
+def _validate_normal_role_adapters(
+    profiles: Mapping[str, Profile],
+    roles: Mapping[str, tuple[str, ...]],
+) -> None:
+    """Reject adapters outside ``NORMAL_ROLE_ADAPTERS`` on implementation, repair, and test.
+
+    Applies to every profile in those roles, enabled or not, in every mode.
+    """
+    for role in ("implementation", "repair", "test"):
+        for name in roles[role]:
+            adapter = profiles[name].adapter
+            if adapter not in NORMAL_ROLE_ADAPTERS:
+                raise ConfigError(
+                    f"profile {name!r} in role {role!r} must use adapter "
+                    "cursor, antigravity, or pi"
+                )
+
+
+def _validate_normal_role_exactness(
+    profiles: Mapping[str, Profile],
+    roles: Mapping[str, tuple[str, ...]],
+) -> None:
+    """Each normal role needs exactly one enabled profile per normal adapter.
+
+    Disabled profiles do not count and may appear alongside the enabled one.
+    Review is exempt.
+    """
+    for role in ("implementation", "repair", "test"):
+        enabled: dict[str, list[str]] = {}
+        for name in roles[role]:
+            profile = profiles[name]
+            if profile.enabled:
+                enabled.setdefault(profile.adapter, []).append(name)
+        for adapter in sorted(NORMAL_ROLE_ADAPTERS):
+            names = enabled.get(adapter, [])
+            if len(names) != 1:
+                raise ConfigError(
+                    f"role {role!r} must reference exactly one enabled "
+                    f"{adapter} profile; found {names or 'none'}"
+                )
+
+
 def _positive_bounded(value: Any, name: str, lo: int, hi: int) -> int:
     if type(value) is not int or isinstance(value, bool):
         raise ConfigError(f"{name} must be an integer")

scripts/hooks/pre-commit

@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
-# 本地硬闸（pre-commit）：集成分支直提拒绝 + 分支命名。权威准入在 PR（baseline.md:26）。
+# 本地检查（pre-commit）：集成分支直提拒绝 + 分支命名 + 本地完整性核对。权威准入在 PR（baseline.md:26）。
 branch=$(git rev-parse --abbrev-ref HEAD)
 case "$branch" in
   main|master)
@@ -10,4 +10,7 @@ echo "$branch" | grep -qE '^(feature|fix|hotfix|release)/[A-Za-z0-9._-]+$' || {
   echo "pre-commit: 拒绝——分支名须为 feature|fix|hotfix|release/<slug>，当前：${branch}" >&2
   exit 1
 }
-exit 0
+root=$(git rev-parse --show-toplevel)
+# admit 只是本地完整性/篡改证据（SELF）：同一 OS 用户可读取本地 HMAC 密钥并绕过本钩子，
+# 它不是 GATE，也不是可信身份。缺失或无效的本地记录仍拒绝；交付信任来自精确 HEAD 的 CI 或独立 REPLAY。
+PYTHONDONTWRITEBYTECODE=1 python3 "$root/scripts/agent_exec.py" admit --repo "$root"

scripts/task.sh

@@ -213,7 +213,7 @@ ${parent_note}
 
 ## Agent 实施收据
 
-按根目录 AGENT-EXECUTION.md 路由；外部 Agent 一单元一新进程/工作区。仅当 scripts/agent_exec.py run 对该精确单元发出有效 host-fallback-authorized 收据后，当前 host 方可实施该单元。failed-unknown、scope-expansion、无效配置、未对账 Git 状态与安全失败均不授权回退。原始日志保留在源码外收据路径。
+按根目录 AGENT-EXECUTION.md 路由；外部 Agent 一单元一新进程/工作区。实施、修复与测试的正常候选仅为 Cursor、Antigravity（agy）和 PI；每个候选对安全可恢复失败最多 3 次全新尝试，然后才换下一候选。三者都安全耗尽后，才允许一次全新的 Codex CLI 紧急尝试。Codex 平时只用于规划、方案设计、实施计划或独立 Review。显式 fail-stop（failed-unknown、scope-expansion、blocked、cancelled、无效配置、未对账 Git 状态与安全失败）立即停止。不因推理时长、工具次数、规划方式或首次写文件时间判定失败；尝试只在显式终态失败、进程退出、总超时，或 stdin/stdout/stderr 均无活动的空闲超时时失败。CLI 可在目标、范围与安全边界内自主检查、计划、测试并迭代。实施集成必须是成功 transport 上的 scripts/agent_exec.py apply，或 pre-commit admit 的精确内容来源。host 直接管理的豁免仅为 .task/**、docs/superpowers/specs/** 与 docs/superpowers/plans/**。fallback 列只记录该行是否为 Codex 紧急尝试。安全耗尽不发出 host-fallback-authorized，终态为 unavailable 或 blocked。不授权 host 直接撰写该单元。原始日志保留在源码外收据路径。
 
 | 单元 ID | 预期范围 | 配置摘要 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 |
 |---|---|---|---|---|---|---|---|---|---|---|

tests/test_agent_exec_admission.py

@@ -0,0 +1,739 @@
+"""Provenance admission: apply a success transport, or admit exact staged paths."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+import uuid
+from pathlib import Path
+
+from scripts import agent_exec
+
+
+SOURCE = Path(__file__).resolve().parents[1]
+
+
+def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["git", "-C", str(repo), *args],
+        check=check,
+        capture_output=True,
+        text=True,
+    )
+
+
+class AgentExecAdmissionTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-admit-")
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.repo = self.root / "repo"
+        self.repo.mkdir()
+        _git(self.repo, "init", "-b", "main")
+        _git(self.repo, "config", "user.email", "agent-exec@example.com")
+        _git(self.repo, "config", "user.name", "Agent Exec")
+        (self.repo / "README.md").write_text("fixture\n", encoding="utf-8")
+        _git(self.repo, "add", "README.md")
+        _git(self.repo, "commit", "-m", "init")
+        _git(self.repo, "checkout", "-b", "feature/admission-test")
+        self.head = _git(self.repo, "rev-parse", "HEAD").stdout.strip()
+
+    def _cli(self, *args: str) -> subprocess.CompletedProcess[str]:
+        env = os.environ.copy()
+        env["PYTHONDONTWRITEBYTECODE"] = "1"
+        env["PYTHONPATH"] = str(SOURCE)
+        return subprocess.run(
+            ["python3", "-m", "scripts.agent_exec", *args],
+            cwd=str(self.root),
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+
+    def _state(self) -> dict[str, str]:
+        return {
+            "head": _git(self.repo, "rev-parse", "HEAD").stdout.strip(),
+            "branch": _git(self.repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip(),
+            "status": _git(
+                self.repo, "status", "--porcelain", "--untracked-files=all"
+            ).stdout,
+            "index": _git(self.repo, "ls-files", "--stage").stdout,
+            "readme": (self.repo / "README.md").read_text(encoding="utf-8"),
+        }
+
+    def _transport_commit(
+        self,
+        files: dict[str, str],
+        *,
+        parent: str | None = None,
+        ref: str | None = None,
+    ) -> tuple[str, str]:
+        parent = parent or _git(self.repo, "rev-parse", "HEAD").stdout.strip()
+        ref = ref or f"refs/agent-exec/{uuid.uuid4().hex}/attempt-1"
+        index = self.root / f"index-{uuid.uuid4().hex}"
+        env = os.environ.copy()
+        env["GIT_INDEX_FILE"] = str(index)
+        subprocess.run(
+            ["git", "-C", str(self.repo), "read-tree", parent],
+            check=True,
+            env=env,
+            capture_output=True,
+            text=True,
+        )
+        for rel, content in files.items():
+            blob = subprocess.run(
+                ["git", "-C", str(self.repo), "hash-object", "-w", "--stdin"],
+                input=content,
+                text=True,
+                check=True,
+                capture_output=True,
+            ).stdout.strip()
+            subprocess.run(
+                [
+                    "git",
+                    "-C",
+                    str(self.repo),
+                    "update-index",
+                    "--add",
+                    "--cacheinfo",
+                    f"100644,{blob},{rel}",
+                ],
+                check=True,
+                env=env,
+                capture_output=True,
+                text=True,
+            )
+        tree = subprocess.run(
+            ["git", "-C", str(self.repo), "write-tree"],
+            check=True,
+            env=env,
+            capture_output=True,
+            text=True,
+        ).stdout.strip()
+        commit = subprocess.run(
+            [
+                "git",
+                "-C",
+                str(self.repo),
+                "-c",
+                "user.name=agent-exec-transport",
+                "-c",
+                "user.email=agent-exec-transport@localhost.invalid",
+                "commit-tree",
+                tree,
+                "-p",
+                parent,
+                "-m",
+                "agent-exec transport",
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.strip()
+        _git(self.repo, "update-ref", ref, commit)
+        index.unlink(missing_ok=True)
+        return commit, ref
+
+    def _receipt(
+        self,
+        *,
+        parent: str,
+        ref: str | None,
+        paths: list[str],
+        status: str = "success",
+        repo: str | None = None,
+        reason: str | None = None,
+        transport_parent: str | None = None,
+        omit_transport: bool = False,
+        run_id: str | None = None,
+        commit: str | None = None,
+    ) -> Path:
+        data: dict[str, object] = {
+            "schema": agent_exec.RECEIPT_SCHEMA,
+            "repo": repo if repo is not None else str(self.repo.resolve()),
+            "terminal_status": status,
+            "reason": reason,
+            "base": {
+                "head": parent,
+                "tree": _git(self.repo, "rev-parse", f"{parent}^{{tree}}").stdout.strip(),
+                "branch": "feature/admission-test",
+            },
+            "changed_paths": paths,
+        }
+        if run_id is not None:
+            data["run_id"] = run_id
+        if not omit_transport:
+            transport: dict[str, object] = {
+                "ref": ref,
+                "parent": parent if transport_parent is None else transport_parent,
+                "attempt_id": "attempt-1",
+            }
+            if commit is not None:
+                transport["commit"] = commit
+            data["transport"] = transport
+        path = self.root / f"receipt-{uuid.uuid4().hex}.json"
+        path.write_text(json.dumps(data), encoding="utf-8")
+        return path
+
+    def _common_dir(self) -> Path:
+        return Path(
+            _git(
+                self.repo,
+                "rev-parse",
+                "--path-format=absolute",
+                "--git-common-dir",
+            ).stdout.strip()
+        )
+
+    def _history_path(self) -> Path:
+        return self._common_dir() / "agent-exec" / "execution-history.jsonl"
+
+    def _attestation_key(self) -> bytes:
+        dest = self._common_dir() / "agent-exec"
+        dest.mkdir(mode=0o700, exist_ok=True)
+        os.chmod(dest, 0o700)
+        path = dest / "attestation.key"
+        if path.is_symlink():
+            raise RuntimeError("test attestation key must not be a symlink")
+        if path.exists():
+            key = path.read_bytes()
+            if len(key) != 32:
+                raise RuntimeError("test attestation key must be 32 bytes")
+            return key
+        key = os.urandom(32)
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        try:
+            os.write(fd, key)
+            os.fchmod(fd, 0o600)
+        finally:
+            os.close(fd)
+        return key
+
+    @staticmethod
+    def _sign_history_record(record: dict[str, object], key: bytes) -> dict[str, object]:
+        body = {field: value for field, value in record.items() if field != "attestation"}
+        payload = json.dumps(
+            body, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        signed = dict(body)
+        signed["attestation"] = hmac.new(key, payload, hashlib.sha256).hexdigest()
+        return signed
+
+    def _history(
+        self,
+        *,
+        parent: str,
+        ref: str,
+        paths: list[str],
+        status: str = "success",
+        repo: str | None = None,
+        commit: str | None = None,
+        run_id: str | None = None,
+        sign: bool | None = None,
+    ) -> dict[str, object]:
+        if commit is None:
+            commit = _git(self.repo, "rev-parse", f"{ref}^{{commit}}").stdout.strip()
+        identity = run_id or uuid.uuid4().hex
+        record: dict[str, object] = {
+            "schema": agent_exec.HISTORY_SCHEMA,
+            "terminal_status": status,
+            "record_id": identity,
+            "run_id": identity,
+            "repo": repo if repo is not None else str(self.repo.resolve()),
+            "changed_paths": paths,
+            "transport": {"ref": ref, "parent": parent, "commit": commit},
+        }
+        if sign is None:
+            sign = status == "success"
+        if sign:
+            record = self._sign_history_record(record, self._attestation_key())
+        dest = self._history_path().parent
+        dest.mkdir(mode=0o700, exist_ok=True)
+        os.chmod(dest, 0o700)
+        line = json.dumps(
+            record, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        )
+        with self._history_path().open("a", encoding="utf-8") as handle:
+            handle.write(line + "\n")
+        return record
+
+    def _reset_branch(self) -> None:
+        _git(self.repo, "reset", "--hard", self.head)
+        _git(self.repo, "clean", "-fd")
+
+    def _stage(self, rel: str, content: str) -> None:
+        path = self.repo / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        _git(self.repo, "add", "--", rel)
+
+    def test_apply_fast_forwards_clean_matching_transport(self) -> None:
+        commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        run_id = uuid.uuid4().hex
+        signed = self._history(
+            parent=self.head,
+            ref=ref,
+            paths=["src/marker.txt"],
+            commit=commit,
+            run_id=run_id,
+        )
+        receipt = self._receipt(
+            parent=self.head,
+            ref=ref,
+            paths=["src/marker.txt"],
+            run_id=run_id,
+            commit=commit,
+        )
+        self.assertEqual(signed["transport"]["commit"], commit)
+        self.assertIn("attestation", signed)
+        proc = self._cli(
+            "apply", "--repo", str(self.repo), "--receipt", str(receipt)
+        )
+        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
+        self.assertIn("apply: fast-forwarded", proc.stdout)
+        self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), commit)
+        self.assertEqual(
+            _git(self.repo, "rev-parse", "feature/admission-test").stdout.strip(),
+            commit,
+        )
+        self.assertEqual((self.repo / "src/marker.txt").read_text(encoding="utf-8"), "shipped\n")
+        self.assertEqual(_git(self.repo, "status", "--porcelain").stdout, "")
+        parents = _git(self.repo, "rev-list", "--parents", "-n", "1", "HEAD").stdout.split()
+        self.assertEqual(parents, [commit, self.head])
+
+    def test_apply_rejects_without_changing_state(self) -> None:
+        _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        valid = dict(parent=self.head, ref=ref, paths=["src/marker.txt"])
+
+        def reject(label: str, receipt: Path) -> None:
+            before = self._state()
+            proc = self._cli(
+                "apply", "--repo", str(self.repo), "--receipt", str(receipt)
+            )
+            self.assertEqual(
+                proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout
+            )
+            self.assertIn(label, proc.stderr)
+            self.assertNotIn("apply: fast-forwarded", proc.stdout)
+            self.assertEqual(self._state(), before)
+
+        with self.subTest(label="stale HEAD"):
+            (self.repo / "advance.txt").write_text("advance\n", encoding="utf-8")
+            _git(self.repo, "add", "advance.txt")
+            _git(self.repo, "commit", "-m", "advance")
+            reject("stale HEAD", self._receipt(**valid))
+            _git(self.repo, "reset", "--hard", self.head)
+
+        with self.subTest(label="dirty worktree"):
+            (self.repo / "README.md").write_text("dirty\n", encoding="utf-8")
+            reject("dirty worktree", self._receipt(**valid))
+            _git(self.repo, "checkout", "--", "README.md")
+
+        with self.subTest(label="non-success receipt"):
+            reject(
+                "non-success receipt",
+                self._receipt(**valid, status="failed-unknown"),
+            )
+
+        with self.subTest(label="missing transport ref"):
+            reject(
+                "missing transport",
+                self._receipt(
+                    parent=self.head,
+                    ref="refs/agent-exec/missing/attempt",
+                    paths=["src/marker.txt"],
+                ),
+            )
+
+        with self.subTest(label="omitted transport"):
+            reject(
+                "missing transport",
+                self._receipt(
+                    parent=self.head,
+                    ref=None,
+                    paths=["src/marker.txt"],
+                    omit_transport=True,
+                ),
+            )
+
+        with self.subTest(label="mismatched transport"):
+            side = _git(
+                self.repo,
+                "commit-tree",
+                _git(self.repo, "rev-parse", "HEAD^{tree}").stdout.strip(),
+                "-p",
+                self.head,
+                "-m",
+                "side",
+            ).stdout.strip()
+            _mismatch_commit, mismatch_ref = self._transport_commit(
+                {"src/marker.txt": "other\n"}, parent=side
+            )
+            reject(
+                "mismatched transport",
+                self._receipt(
+                    parent=self.head, ref=mismatch_ref, paths=["src/marker.txt"]
+                ),
+            )
+
+        with self.subTest(label="receipt repo mismatch"):
+            reject(
+                "receipt repo mismatch",
+                self._receipt(
+                    **valid, repo=str((self.root / "other-repo").resolve())
+                ),
+            )
+
+    def test_apply_rejects_fabricated_unsigned_or_tampered_history(self) -> None:
+        commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        run_id = uuid.uuid4().hex
+        paths = ["src/marker.txt"]
+
+        def reject(label: str) -> None:
+            self._reset_branch()
+            before = self._state()
+            receipt = self._receipt(
+                parent=self.head,
+                ref=ref,
+                paths=paths,
+                run_id=run_id,
+                commit=commit,
+            )
+            proc = self._cli(
+                "apply", "--repo", str(self.repo), "--receipt", str(receipt)
+            )
+            combined = proc.stdout + proc.stderr
+            self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+            self.assertIn("apply: no local integrity record", proc.stderr)
+            self.assertNotIn("apply: fast-forwarded", proc.stdout)
+            self.assertEqual(self._state(), before, label)
+
+        with self.subTest(label="fabricated receipt"):
+            reject("fabricated receipt")
+
+        with self.subTest(label="unsigned history"):
+            self._history(
+                parent=self.head,
+                ref=ref,
+                paths=paths,
+                commit=commit,
+                run_id=run_id,
+                sign=False,
+            )
+            reject("unsigned history")
+
+        with self.subTest(label="tampered history"):
+            self._reset_branch()
+            signed = self._history(
+                parent=self.head,
+                ref=ref,
+                paths=paths,
+                commit=commit,
+                run_id=run_id,
+            )
+            history_path = self._history_path()
+            lines = history_path.read_text(encoding="utf-8").splitlines()
+            tampered = json.loads(lines[-1])
+            self.assertEqual(tampered["attestation"], signed["attestation"])
+            tampered["record_id"] = "tampered-record"
+            lines[-1] = json.dumps(
+                tampered, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+            )
+            history_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+            reject("tampered history")
+
+    def test_apply_redacts_secret_shaped_receipt_text(self) -> None:
+        _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        secret = "sk-admissionsecret"
+        receipt = self._receipt(
+            parent=self.head,
+            ref=ref,
+            paths=["src/marker.txt"],
+            status="failed-unknown",
+            reason=secret,
+        )
+        before = self._state()
+        proc = self._cli("apply", "--repo", str(self.repo), "--receipt", str(receipt))
+        combined = proc.stdout + proc.stderr
+        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+        self.assertIn("non-success receipt", proc.stderr)
+        self.assertNotIn(secret, combined)
+        self.assertIn("***", combined)
+        self.assertEqual(self._state(), before)
+
+    def test_admit_rejects_staged_paths_without_provenance(self) -> None:
+        self._stage("src/extra.txt", "unproven\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
+        self.assertIn("admit: no local integrity record", proc.stderr)
+        self.assertEqual(self._state(), before)
+
+    def test_admit_accepts_exact_transport_paths(self) -> None:
+        _commit, ref = self._transport_commit(
+            {"src/marker.txt": "shipped\n", "src/other.txt": "also\n"}
+        )
+        self._history(
+            parent=self.head,
+            ref=ref,
+            paths=["src/marker.txt", "src/other.txt"],
+        )
+        self._stage("src/marker.txt", "shipped\n")
+        self._stage("src/other.txt", "also\n")
+        self._stage(".task/feature-admission.md", "host card\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
+        self.assertIn("admit: ok", proc.stdout)
+        self.assertEqual(self._state(), before)
+
+        self._git_reset_index()
+        self._stage("src/marker.txt", "shipped\n")
+        partial = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(partial.returncode, agent_exec.EXIT_BLOCKED, partial.stderr)
+        self.assertIn("admit: no local integrity record", partial.stderr)
+
+    def test_admit_rejects_tampered_record_and_moved_ref(self) -> None:
+        commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        signed = self._history(parent=self.head, ref=ref, paths=["src/marker.txt"], commit=commit)
+        history_path = self._history_path()
+        lines = history_path.read_text(encoding="utf-8").splitlines()
+        tampered = json.loads(lines[-1])
+        self.assertEqual(tampered["attestation"], signed["attestation"])
+        tampered["record_id"] = "tampered-record"
+        lines[-1] = json.dumps(
+            tampered, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        )
+        history_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+        self._stage("src/marker.txt", "shipped\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        combined = proc.stdout + proc.stderr
+        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+        self.assertIn("admit: no local integrity record", proc.stderr)
+        self.assertNotIn("admit: ok", proc.stdout)
+        self.assertEqual(self._state(), before)
+
+        self._reset_branch()
+        tree = _git(self.repo, "rev-parse", f"{commit}^{{tree}}").stdout.strip()
+        moved = _git(
+            self.repo,
+            "commit-tree",
+            tree,
+            "-p",
+            self.head,
+            "-m",
+            "moved ref",
+        ).stdout.strip()
+        self.assertNotEqual(moved, commit)
+        restored = self._sign_history_record(
+            {
+                "schema": agent_exec.HISTORY_SCHEMA,
+                "terminal_status": "success",
+                "record_id": signed["record_id"],
+                "run_id": signed["run_id"],
+                "repo": str(self.repo.resolve()),
+                "changed_paths": ["src/marker.txt"],
+                "transport": {"commit": commit, "parent": self.head, "ref": ref},
+            },
+            self._attestation_key(),
+        )
+        history_path.write_text(
+            json.dumps(
+                restored, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+        _git(self.repo, "update-ref", ref, moved)
+        self._stage("src/marker.txt", "shipped\n")
+        moved_before = self._state()
+        moved_proc = self._cli("admit", "--repo", str(self.repo))
+        moved_text = moved_proc.stdout + moved_proc.stderr
+        self.assertEqual(moved_proc.returncode, agent_exec.EXIT_BLOCKED, moved_text)
+        self.assertIn("admit: no local integrity record", moved_proc.stderr)
+        self.assertNotIn("admit: ok", moved_proc.stdout)
+        self.assertEqual(self._state(), moved_before)
+        self.assertEqual(_git(self.repo, "rev-parse", ref).stdout.strip(), moved)
+
+    def test_admit_rejects_different_content_at_transport_path(self) -> None:
+        _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        self._history(parent=self.head, ref=ref, paths=["src/marker.txt"])
+        self._stage("src/marker.txt", "rewritten\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
+        self.assertIn("admit: no local integrity record", proc.stderr)
+        self.assertEqual(self._state(), before)
+
+    def test_rename_and_copy_records_keep_non_exempt_source(self) -> None:
+        """Copy/rename records must not drop a non-exempt source behind an exempt dest."""
+        paths = agent_exec._name_status_paths(
+            "R100\tsrc/secret.txt\t.task/from-rename.md\n"
+            "C100\tlib/outside.txt\tdocs/superpowers/plans/from-copy.md\n"
+        )
+        self.assertEqual(
+            agent_exec._non_exempt(paths),
+            {"src/secret.txt", "lib/outside.txt"},
+        )
+
+    def test_admit_rejects_rename_of_non_exempt_source_into_exempt_path(self) -> None:
+        (self.repo / "src").mkdir()
+        (self.repo / "src" / "secret.txt").write_text("secret\n", encoding="utf-8")
+        _git(self.repo, "add", "src/secret.txt")
+        _git(self.repo, "commit", "-m", "add secret")
+        (self.repo / ".task").mkdir()
+        _git(self.repo, "mv", "src/secret.txt", ".task/hidden.md")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        combined = proc.stdout + proc.stderr
+        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+        self.assertIn("src/secret.txt", proc.stderr)
+        self.assertNotIn("admit: ok", proc.stdout)
+        self.assertEqual(self._state(), before)
+
+    def test_admit_accepts_host_managed_exemptions(self) -> None:
+        self._stage(".task/feature-admission.md", "card\n")
+        self._stage("docs/superpowers/specs/admission.md", "spec\n")
+        self._stage("docs/superpowers/plans/admission.md", "plan\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
+        self.assertIn("admit: ok", proc.stdout)
+        self.assertEqual(self._state(), before)
+
+        self._stage("docs/superpowers/specs-extra/not-exempt.md", "nope\n")
+        rejected = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(rejected.returncode, agent_exec.EXIT_BLOCKED, rejected.stderr)
+        self.assertIn("admit: no local integrity record", rejected.stderr)
+
+    def test_whitespace_paths_are_classified_exactly(self) -> None:
+        non_exempt = (
+            " .task/card.md",
+            "\t.task/card.md",
+            " docs/superpowers/specs/a.md",
+            "docs/superpowers/specs /a.md",
+            "docs/superpowers/plans /a.md",
+            ".task /card.md",
+        )
+        for path in non_exempt:
+            with self.subTest(path=path):
+                self.assertFalse(agent_exec._host_managed(path))
+                self.assertEqual(agent_exec._non_exempt([path]), {path})
+        for path in (
+            ".task/card.md ",
+            "docs/superpowers/specs/a.md ",
+            "docs/superpowers/plans/ a.md",
+        ):
+            with self.subTest(path=path):
+                self.assertTrue(agent_exec._host_managed(path))
+        self.assertEqual(
+            agent_exec._non_exempt(["src/a.txt ", " src/b.txt"]),
+            {"src/a.txt ", " src/b.txt"},
+        )
+        self.assertEqual(
+            agent_exec._name_status_paths("A\t .task/card.md\nA\tsrc/a.txt \n"),
+            [" .task/card.md", "src/a.txt "],
+        )
+
+    def test_admit_rejects_whitespace_lookalikes_of_exempt_paths(self) -> None:
+        for path in (
+            " .task/card.md",
+            "docs/superpowers/specs /a.md",
+            "docs/superpowers/plans /a.md",
+        ):
+            with self.subTest(path=path):
+                self._reset_branch()
+                self._stage(path, "lookalike\n")
+                before = self._state()
+                proc = self._cli("admit", "--repo", str(self.repo))
+                combined = proc.stdout + proc.stderr
+                self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+                self.assertIn(f"admit: no local integrity record for {path}", proc.stderr)
+                self.assertNotIn("admit: ok", proc.stdout)
+                self.assertEqual(self._state(), before)
+
+    def test_admit_covers_whitespace_path_only_with_exact_transport(self) -> None:
+        exact = "src/marker.txt "
+        _commit, ref = self._transport_commit({exact: "shipped\n"})
+        self._history(parent=self.head, ref=ref, paths=[exact])
+        self._stage(exact, "shipped\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
+        self.assertIn("admit: ok", proc.stdout)
+        self.assertEqual(self._state(), before)
+
+        self._reset_branch()
+        self._history_path().unlink()
+        _commit, trimmed_ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        self._history(parent=self.head, ref=trimmed_ref, paths=["src/marker.txt"])
+        self._stage(exact, "shipped\n")
+        rejected = self._cli("admit", "--repo", str(self.repo))
+        self.assertEqual(rejected.returncode, agent_exec.EXIT_BLOCKED, rejected.stderr)
+        self.assertIn("admit: no local integrity record", rejected.stderr)
+
+    def test_precommit_rejects_unproven_paths_after_branch_checks(self) -> None:
+        self._install_project_hook()
+        self._stage("src/extra.txt", "unproven\n")
+        proc = self._hook()
+        combined = proc.stdout + proc.stderr
+        self.assertNotEqual(proc.returncode, 0, combined)
+        self.assertIn("admit: no local integrity record", combined)
+
+        _git(self.repo, "checkout", "-B", "main", self.head)
+        self._stage("src/extra.txt", "unproven\n")
+        blocked = self._hook()
+        text = blocked.stdout + blocked.stderr
+        self.assertNotEqual(blocked.returncode, 0, text)
+        self.assertIn("集成分支", text)
+        self.assertNotIn("admit:", text)
+
+    def test_precommit_accepts_host_managed_exemptions(self) -> None:
+        self._install_project_hook()
+        self._stage("docs/superpowers/plans/admission.md", "plan\n")
+        proc = self._hook()
+        self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout)
+        self.assertIn("admit: ok", proc.stdout)
+
+    def test_make_agent_exec_apply_requires_receipt(self) -> None:
+        env = os.environ.copy()
+        env.pop("RECEIPT", None)
+        proc = subprocess.run(
+            ["make", "agent-exec-apply"],
+            cwd=str(SOURCE),
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+        combined = proc.stdout + proc.stderr
+        self.assertNotEqual(proc.returncode, 0, combined)
+        self.assertIn("RECEIPT", combined)
+        self.assertNotIn("apply: fast-forwarded", combined)
+
+    def _git_reset_index(self) -> None:
+        _git(self.repo, "reset", "--hard", self.head)
+        _git(self.repo, "clean", "-fd")
+
+    def _install_project_hook(self) -> None:
+        dest = self.repo / "scripts"
+        hooks = dest / "hooks"
+        hooks.mkdir(parents=True)
+        for name in ("agent_exec.py", "agent_exec_config.py", "agent_exec_adapters.py"):
+            shutil.copy(SOURCE / "scripts" / name, dest / name)
+        hook = hooks / "pre-commit"
+        shutil.copy(SOURCE / "scripts" / "hooks" / "pre-commit", hook)
+        hook.chmod(0o755)
+
+    def _hook(self) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            ["bash", str(self.repo / "scripts" / "hooks" / "pre-commit")],
+            cwd=str(self.repo),
+            capture_output=True,
+            text=True,
+        )

tests/test_agent_exec_config.py

@@ -31,10 +31,10 @@ MINIMAL_BLOCK = textwrap.dedent(
       raw_output_bytes: 10485760
 
     roles:
-      implementation: [cursor, claude]
+      implementation: [cursor, antigravity, pi]
-      repair: [cursor]
+      repair: [cursor, antigravity, pi]
-      test: [claude]
+      test: [pi, antigravity, cursor]
-      review: [claude]
+      review: [pi]
 
     profiles:
       cursor:
@@ -43,9 +43,15 @@ MINIMAL_BLOCK = textwrap.dedent(
         model: auto
         enabled: true
 
-      claude:
+      antigravity:
-        adapter: claude
+        adapter: antigravity
-        executable: claude
+        executable: agy
+        model: auto
+        enabled: true
+
+      pi:
+        adapter: pi
+        executable: pi
         model: auto
         enabled: true
     """
@@ -76,7 +82,7 @@ class AgentExecConfigTests(unittest.TestCase):
         self.assertTrue(cfg.execution.fresh_process_per_unit)
         self.assertEqual(cfg.execution.permission, "run-everything")
         self.assertEqual(cfg.execution.fallback_to_host, "all-unavailable")
-        self.assertEqual(cfg.roles["implementation"], ("cursor", "claude"))
+        self.assertEqual(cfg.roles["implementation"], ("cursor", "antigravity", "pi"))
         self.assertEqual(cfg.profiles["cursor"].executable, "cursor-agent")
         self.assertEqual(cfg.custom_adapters, {})
         self.assertEqual(validate_config(self.repo), [])
@@ -112,7 +118,7 @@ class AgentExecConfigTests(unittest.TestCase):
             load_config(self.repo)
 
     def test_role_order_does_not_follow_profile_order(self) -> None:
-        # profiles listed claude before cursor; role order must stay cursor -> claude
+        # profiles listed pi before cursor; role order must stay cursor -> pi
         block = textwrap.dedent(
             """\
             schema: agent-execution/v1
@@ -127,15 +133,20 @@ class AgentExecConfigTests(unittest.TestCase):
               raw_output_bytes: 4096
 
             roles:
-              implementation: [cursor, claude]
+              implementation: [cursor, antigravity, pi]
-              repair: [cursor]
+              repair: [cursor, antigravity, pi]
-              test: [claude]
+              test: [cursor, antigravity, pi]
-              review: [claude]
+              review: [pi]
 
             profiles:
-              claude:
+              pi:
-                adapter: claude
+                adapter: pi
-                executable: claude
+                executable: pi
+                model: auto
+                enabled: true
+              antigravity:
+                adapter: antigravity
+                executable: agy
                 model: auto
                 enabled: true
               cursor:
@@ -147,19 +158,19 @@ class AgentExecConfigTests(unittest.TestCase):
         )
         write_repo(self.repo, block)
         cfg = load_config(self.repo)
-        self.assertEqual(cfg.roles["implementation"], ("cursor", "claude"))
+        self.assertEqual(cfg.roles["implementation"], ("cursor", "antigravity", "pi"))
-        self.assertEqual(list(cfg.profiles.keys())[0], "claude")
+        self.assertEqual(list(cfg.profiles.keys())[0], "pi")
 
     def test_model_auto_is_preserved_as_none(self) -> None:
         write_repo(self.repo, MINIMAL_BLOCK)
         cfg = load_config(self.repo)
         self.assertIsNone(cfg.profiles["cursor"].model)
-        self.assertIsNone(cfg.profiles["claude"].model)
+        self.assertIsNone(cfg.profiles["pi"].model)
 
     def test_rejects_unknown_profile_reference(self) -> None:
         block = MINIMAL_BLOCK.replace(
-            "implementation: [cursor, claude]",
+            "implementation: [cursor, antigravity, pi]",
-            "implementation: [cursor, claude, missing]",
+            "implementation: [cursor, antigravity, pi, missing]",
         )
         write_repo(self.repo, block)
         with self.assertRaises(ConfigError):
@@ -175,15 +186,15 @@ class AgentExecConfigTests(unittest.TestCase):
             load_config(self.repo)
 
         secret_key = MINIMAL_BLOCK.replace(
-            "enabled: true\n\n  claude:",
+            "enabled: true\n\n  pi:",
-            "enabled: true\n    api_token: \"sk-secret-value\"\n\n  claude:",
+            "enabled: true\n    api_token: \"sk-secret-value\"\n\n  pi:",
         )
         write_repo(self.repo, secret_key)
         with self.assertRaises(ConfigError):
             load_config(self.repo)
 
         secret_val = MINIMAL_BLOCK.replace(
-            "executable: claude",
+            "executable: pi",
             "executable: \"Bearer abcdef0123456789abcdef0123456789\"",
         )
         write_repo(self.repo, secret_val)
@@ -191,26 +202,9 @@ class AgentExecConfigTests(unittest.TestCase):
             load_config(self.repo)
 
     def test_custom_placeholders_must_fill_whole_argv_items(self) -> None:
-        good = textwrap.dedent(
+        good = MINIMAL_BLOCK.replace("review: [pi]", "review: [mycustom]") + textwrap.dedent(
             """\
-            schema: agent-execution/v1
 
-            execution:
-              fresh_process_per_unit: true
-              permission: run-everything
-              fallback_to_host: all-unavailable
-              max_parallel: 1
-              timeout_seconds: 60
-              summary_bytes: 1024
-              raw_output_bytes: 4096
-
-            roles:
-              implementation: [mycustom]
-              repair: [mycustom]
-              test: [mycustom]
-              review: [mycustom]
-
-            profiles:
               mycustom:
                 adapter: custom
                 executable: mycli
@@ -320,20 +314,20 @@ class AgentExecConfigTests(unittest.TestCase):
 
     def test_quoted_unicode_string_round_trips(self) -> None:
         block = MINIMAL_BLOCK.replace(
-            "executable: claude",
+            "executable: pi",
             'executable: "模型-β-测试"',
         )
         write_repo(self.repo, block)
         cfg = load_config(self.repo)
-        self.assertEqual(cfg.profiles["claude"].executable, "模型-β-测试")
+        self.assertEqual(cfg.profiles["pi"].executable, "模型-β-测试")
 
         escaped = MINIMAL_BLOCK.replace(
-            "executable: claude",
+            "executable: pi",
             r'executable: "say \"hi\" 模型"',
         )
         write_repo(self.repo, escaped)
         cfg = load_config(self.repo)
-        self.assertEqual(cfg.profiles["claude"].executable, 'say "hi" 模型')
+        self.assertEqual(cfg.profiles["pi"].executable, 'say "hi" 模型')
 
     def test_idle_timeout_defaults_within_total_and_rejects_bounds(self) -> None:
         self.assertEqual(DEFAULT_IDLE_TIMEOUT_SECONDS, 600)
@@ -405,6 +399,553 @@ class AgentExecConfigTests(unittest.TestCase):
         with self.assertRaises(ConfigError):
             load_config(self.repo)
 
+    def test_exposes_max_attempts_per_profile_and_emergency_profile(self) -> None:
+        write_repo(self.repo, _emergency_block())
+        cfg = load_config(self.repo)
+        self.assertEqual(cfg.execution.max_attempts_per_profile, 3)
+        self.assertEqual(cfg.execution.emergency_profile, "codex")
+
+    def test_configured_emergency_profile_must_exist_be_enabled_and_use_adapter_codex(
+        self,
+    ) -> None:
+        write_repo(self.repo, _emergency_block(include_profile=False))
+        with self.assertRaises(ConfigError) as ctx:
+            load_config(self.repo)
+        self.assertIn("does not exist", str(ctx.exception))
+
+        write_repo(self.repo, _emergency_block(enabled="false"))
+        with self.assertRaises(ConfigError) as ctx:
+            load_config(self.repo)
+        self.assertIn("must be enabled", str(ctx.exception))
+
+        write_repo(self.repo, _emergency_block(adapter="pi"))
+        with self.assertRaises(ConfigError) as ctx:
+            load_config(self.repo)
+        self.assertIn("adapter codex", str(ctx.exception))
+
+    def test_emergency_profile_is_excluded_from_normal_roles(self) -> None:
+        for role, replacement in (
+            ("implementation", "[cursor, pi, codex]"),
+            ("repair", "[cursor, codex]"),
+            ("test", "[pi, codex]"),
+        ):
+            write_repo(self.repo, _emergency_block(**{role: replacement}))
+            with self.assertRaises(ConfigError) as ctx:
+                load_config(self.repo)
+            self.assertIn("must not appear", str(ctx.exception))
+            self.assertIn(role, str(ctx.exception))
+
+    def test_review_may_include_emergency_profile(self) -> None:
+        write_repo(self.repo, _emergency_block(review="[pi, codex]"))
+        cfg = load_config(self.repo)
+        self.assertEqual(cfg.execution.emergency_profile, "codex")
+        self.assertTrue(cfg.profiles["codex"].enabled)
+        self.assertEqual(cfg.profiles["codex"].adapter, "codex")
+        self.assertEqual(cfg.roles["review"], ("pi", "codex"))
+        self.assertNotIn("codex", cfg.roles["implementation"])
+        self.assertNotIn("codex", cfg.roles["repair"])
+        self.assertNotIn("codex", cfg.roles["test"])
+
+    def test_accepts_fallback_to_host_never(self) -> None:
+        write_repo(self.repo, _strict_routing_block(fallback="never"))
+        cfg = load_config(self.repo)
+        self.assertEqual(cfg.execution.fallback_to_host, "never")
+        self.assertEqual(cfg.execution.max_attempts_per_profile, 3)
+        self.assertEqual(cfg.execution.emergency_profile, "codex")
+
+    def test_legacy_all_unavailable_remains_parseable(self) -> None:
+        write_repo(self.repo, _strict_routing_block(fallback="all-unavailable"))
+        cfg = load_config(self.repo)
+        self.assertEqual(cfg.execution.fallback_to_host, "all-unavailable")
+
+    def test_emergency_profile_requires_max_attempts_exactly_three(self) -> None:
+        for raw_max in ("1", "2", "4", None):
+            with self.subTest(max_attempts=raw_max):
+                write_repo(
+                    self.repo,
+                    _strict_routing_block(fallback="all-unavailable", max_attempts=raw_max),
+                )
+                with self.assertRaises(ConfigError) as ctx:
+                    load_config(self.repo)
+                self.assertIn("max_attempts_per_profile", str(ctx.exception))
+
+    def test_normal_roles_reject_forbidden_adapters_and_alternate_codex_profile(
+        self,
+    ) -> None:
+        claude = textwrap.dedent(
+            """\
+            claude:
+              adapter: claude
+              executable: claude
+              model: auto
+              enabled: true
+            """
+        )
+        claude = "\n".join(
+            f"  {line}" if line else line for line in claude.strip("\n").splitlines()
+        )
+        write_repo(
+            self.repo,
+            _strict_routing_block(
+                fallback="all-unavailable",
+                implementation="[cursor, claude]",
+                extra_profiles=claude,
+            ),
+        )
+        with self.assertRaises(ConfigError) as ctx:
+            load_config(self.repo)
+        self.assertIn("implementation", str(ctx.exception))
+        self.assertIn("cursor, antigravity, or pi", str(ctx.exception))
+
+        planner = textwrap.dedent(
+            """\
+            planner:
+              adapter: codex
+              executable: planner
+              model: auto
+              enabled: true
+            """
+        )
+        planner = "\n".join(
+            f"  {line}" if line else line for line in planner.strip("\n").splitlines()
+        )
+        write_repo(
+            self.repo,
+            _strict_routing_block(
+                fallback="all-unavailable",
+                implementation="[cursor, planner]",
+                extra_profiles=planner,
+            ),
+        )
+        with self.assertRaises(ConfigError) as ctx:
+            load_config(self.repo)
+        self.assertIn("planner", str(ctx.exception))
+        self.assertIn("implementation", str(ctx.exception))
+        self.assertNotIn("host-fallback", str(ctx.exception).lower())
+
+    def test_strict_emergency_mode_rejects_custom_normal_role_adapters(self) -> None:
+        worker = textwrap.dedent(
+            """\
+            worker:
+              adapter: custom
+              executable: worker
+              model: auto
+              enabled: true
+            """
+        )
+        worker = "\n".join(
+            f"  {line}" if line else line for line in worker.strip("\n").splitlines()
+        )
+        adapters = textwrap.dedent(
+            """\
+            custom_adapters:
+              worker:
+                base_argv: ["${workspace}", "${unit_packet}", "${result_path}"]
+                run_everything_argv: []
+                workspace_argv: []
+                model_argv: []
+                prompt_transport: stdin
+                output_format: text
+            """
+        )
+        for role in ("implementation", "repair", "test"):
+            with self.subTest(role=role):
+                roles = {
+                    "implementation": "[cursor]",
+                    "repair": "[cursor]",
+                    "test": "[pi]",
+                }
+                roles[role] = "[worker]"
+                block = _strict_routing_block(
+                    implementation=roles["implementation"],
+                    repair=roles["repair"],
+                    test=roles["test"],
+                    extra_profiles=worker,
+                )
+                write_repo(self.repo, block.rstrip() + "\n\n" + adapters)
+                with self.assertRaises(ConfigError) as ctx:
+                    load_config(self.repo)
+                message = str(ctx.exception)
+                self.assertIn("worker", message)
+                self.assertIn(role, message)
+                self.assertIn("cursor, antigravity, or pi", message)
+
+    def test_emergency_normal_roles_reject_missing_required_adapter(self) -> None:
+        for role, replacement, missing in (
+            ("implementation", "[cursor, pi]", "antigravity"),
+            ("repair", "[cursor, antigravity]", "pi"),
+            ("test", "[pi, antigravity]", "cursor"),
+        ):
+            with self.subTest(role=role):
+                write_repo(self.repo, _strict_routing_block(**{role: replacement}))
+                with self.assertRaises(ConfigError) as ctx:
+                    load_config(self.repo)
+                message = str(ctx.exception)
+                self.assertIn(role, message)
+                self.assertIn(missing, message)
+                self.assertIn("exactly one enabled", message)
+
+    def test_emergency_normal_roles_reject_disabled_only_adapter(self) -> None:
+        for role in ("implementation", "repair", "test"):
+            with self.subTest(role=role):
+                write_repo(
+                    self.repo,
+                    _strict_routing_block(
+                        **{role: "[cursor, agy_off, pi]"},
+                        extra_profiles=_profile_lines("agy_off", "antigravity", "false"),
+                    ),
+                )
+                with self.assertRaises(ConfigError) as ctx:
+                    load_config(self.repo)
+                message = str(ctx.exception)
+                self.assertIn(role, message)
+                self.assertIn("antigravity", message)
+                self.assertIn("exactly one enabled", message)
+
+    def test_emergency_normal_roles_reject_multiple_enabled_same_adapter(self) -> None:
+        for role in ("implementation", "repair", "test"):
+            with self.subTest(role=role):
+                write_repo(
+                    self.repo,
+                    _strict_routing_block(
+                        **{role: "[cursor, antigravity, pi, cursor2]"},
+                        extra_profiles=_profile_lines("cursor2", "cursor", "true"),
+                    ),
+                )
+                with self.assertRaises(ConfigError) as ctx:
+                    load_config(self.repo)
+                message = str(ctx.exception)
+                self.assertIn(role, message)
+                self.assertIn("cursor2", message)
+                self.assertIn("exactly one enabled", message)
+
+    def test_emergency_normal_roles_allow_extra_disabled_and_keep_order(self) -> None:
+        write_repo(
+            self.repo,
+            _strict_routing_block(
+                implementation="[cursor_off, pi, cursor, antigravity]",
+                repair="[antigravity, pi, cursor_off, cursor]",
+                extra_profiles=_profile_lines("cursor_off", "cursor", "false"),
+            ),
+        )
+        cfg = load_config(self.repo)
+        self.assertEqual(
+            cfg.roles["implementation"], ("cursor_off", "pi", "cursor", "antigravity")
+        )
+        self.assertEqual(cfg.roles["repair"], ("antigravity", "pi", "cursor_off", "cursor"))
+        self.assertEqual(cfg.roles["test"], ("pi", "antigravity", "cursor"))
+        self.assertFalse(cfg.profiles["cursor_off"].enabled)
+
+    def test_emergency_exactness_does_not_apply_to_review(self) -> None:
+        write_repo(self.repo, _strict_routing_block(review="[codex]"))
+        cfg = load_config(self.repo)
+        self.assertEqual(cfg.roles["review"], ("codex",))
+
+    def test_without_emergency_custom_normal_roles_are_rejected(self) -> None:
+        write_repo(self.repo, _legacy_custom_block())
+        with self.assertRaises(ConfigError) as ctx:
+            load_config(self.repo)
+        message = str(ctx.exception)
+        self.assertIn("mycustom", message)
+        self.assertIn("cursor, antigravity, or pi", message)
+
+    def test_without_emergency_review_may_use_custom_adapter(self) -> None:
+        block = _legacy_custom_block().replace(
+            "implementation: [mycustom]\n"
+            "  repair: [mycustom]\n"
+            "  test: [mycustom]\n",
+            "implementation: [cursor, antigravity, pi]\n"
+            "  repair: [cursor, antigravity, pi]\n"
+            "  test: [pi, antigravity, cursor]\n",
+        ).replace(
+            "\ncustom_adapters:",
+            "\n"
+            + _profile_lines("cursor", "cursor", "true")
+            + "\n"
+            + _profile_lines("antigravity", "antigravity", "true")
+            + "\n"
+            + _profile_lines("pi", "pi", "true")
+            + "\n\ncustom_adapters:",
+        )
+        write_repo(self.repo, block)
+        cfg = load_config(self.repo)
+        self.assertIsNone(cfg.execution.emergency_profile)
+        self.assertEqual(cfg.roles["review"], ("mycustom",))
+        self.assertEqual(cfg.profiles["mycustom"].adapter, "custom")
+
+    def test_without_emergency_normal_roles_need_exact_enabled_adapter_set(self) -> None:
+        no_emergency = {"emergency": None, "max_attempts": None, "review": "[pi]"}
+        worker = _profile_lines("worker", "custom", "true")
+        adapters = (
+            "custom_adapters:\n"
+            "  worker:\n"
+            '    base_argv: ["${workspace}", "${unit_packet}", "${result_path}"]\n'
+            "    run_everything_argv: []\n"
+            "    workspace_argv: []\n"
+            "    model_argv: []\n"
+            "    prompt_transport: stdin\n"
+            "    output_format: text\n"
+        )
+        claude = _profile_lines("claude", "claude", "true")
+        for role in ("implementation", "repair", "test"):
+            cases = (
+                ("missing", "[cursor, pi]", "", "exactly one enabled"),
+                (
+                    "disabled-only",
+                    "[cursor, agy_off, pi]",
+                    _profile_lines("agy_off", "antigravity", "false"),
+                    "exactly one enabled",
+                ),
+                (
+                    "duplicate-enabled",
+                    "[cursor, antigravity, pi, cursor2]",
+                    _profile_lines("cursor2", "cursor", "true"),
+                    "exactly one enabled",
+                ),
+                (
+                    "custom-extra",
+                    "[cursor, antigravity, pi, worker]",
+                    worker,
+                    "cursor, antigravity, or pi",
+                ),
+                (
+                    "custom-only",
+                    "[worker]",
+                    worker,
+                    "cursor, antigravity, or pi",
+                ),
+                (
+                    "other-builtin",
+                    "[cursor, antigravity, pi, claude]",
+                    claude,
+                    "cursor, antigravity, or pi",
+                ),
+            )
+            for label, replacement, extra, expected in cases:
+                with self.subTest(role=role, case=label):
+                    block = _strict_routing_block(
+                        **no_emergency, **{role: replacement}, extra_profiles=extra
+                    )
+                    if extra is worker:
+                        block = block.rstrip() + "\n\n" + adapters
+                    write_repo(self.repo, block)
+                    with self.assertRaises(ConfigError) as ctx:
+                        load_config(self.repo)
+                    message = str(ctx.exception)
+                    self.assertIn(role, message)
+                    self.assertIn(expected, message)
+
+    def test_without_emergency_disabled_extras_are_ignored_and_order_kept(self) -> None:
+        write_repo(
+            self.repo,
+            _strict_routing_block(
+                emergency=None,
+                max_attempts=None,
+                review="[pi]",
+                implementation="[pi, cursor_off, antigravity, cursor]",
+                test="[antigravity, cursor, pi, cursor_off]",
+                extra_profiles=_profile_lines("cursor_off", "cursor", "false"),
+            ),
+        )
+        cfg = load_config(self.repo)
+        self.assertIsNone(cfg.execution.emergency_profile)
+        self.assertEqual(
+            cfg.roles["implementation"], ("pi", "cursor_off", "antigravity", "cursor")
+        )
+        self.assertEqual(cfg.roles["repair"], ("cursor", "antigravity", "pi"))
+        self.assertEqual(cfg.roles["test"], ("antigravity", "cursor", "pi", "cursor_off"))
+
+    def test_max_attempts_per_profile_never_exceeds_three(self) -> None:
+        for raw_max in ("1", "2", "3"):
+            with self.subTest(max_attempts=raw_max):
+                write_repo(
+                    self.repo,
+                    _strict_routing_block(
+                        emergency=None, max_attempts=raw_max, review="[pi]"
+                    ),
+                )
+                cfg = load_config(self.repo)
+                self.assertEqual(cfg.execution.max_attempts_per_profile, int(raw_max))
+        for raw_max in ("4", "10"):
+            for emergency in (None, "codex"):
+                with self.subTest(max_attempts=raw_max, emergency=emergency):
+                    write_repo(
+                        self.repo,
+                        _strict_routing_block(emergency=emergency, max_attempts=raw_max),
+                    )
+                    with self.assertRaises(ConfigError) as ctx:
+                        load_config(self.repo)
+                    self.assertIn("max_attempts_per_profile", str(ctx.exception))
+
+    def test_checked_in_project_config_never_authorizes_host_fallback(self) -> None:
+        repo = Path(__file__).resolve().parents[1]
+        cfg = load_config(repo)
+        self.assertEqual(cfg.execution.fallback_to_host, "never")
+        self.assertEqual(cfg.execution.max_attempts_per_profile, 3)
+        self.assertEqual(cfg.execution.emergency_profile, "codex")
+        allowed = {"cursor", "antigravity", "pi"}
+        for role in ("implementation", "repair", "test"):
+            self.assertNotIn("codex", cfg.roles[role])
+            for name in cfg.roles[role]:
+                self.assertIn(cfg.profiles[name].adapter, allowed)
+        self.assertIn("codex", cfg.roles["review"])
+        self.assertTrue(cfg.profiles["codex"].enabled)
+        self.assertEqual(cfg.profiles["codex"].adapter, "codex")
+
+
+def _legacy_custom_block() -> str:
+    """No emergency_profile and adapter custom on every role."""
+    return textwrap.dedent(
+        """\
+        schema: agent-execution/v1
+
+        execution:
+          fresh_process_per_unit: true
+          permission: run-everything
+          fallback_to_host: all-unavailable
+          max_parallel: 1
+          timeout_seconds: 60
+          summary_bytes: 1024
+          raw_output_bytes: 4096
+
+        roles:
+          implementation: [mycustom]
+          repair: [mycustom]
+          test: [mycustom]
+          review: [mycustom]
+
+        profiles:
+          mycustom:
+            adapter: custom
+            executable: mycli
+            model: auto
+            enabled: true
+
+        custom_adapters:
+          mycustom:
+            base_argv: ["run", "${unit_packet}"]
+            run_everything_argv: ["--yes"]
+            workspace_argv: ["--cwd", "${workspace}"]
+            model_argv: ["--model", "${model}"]
+            prompt_transport: stdin
+            output_format: json
+        """
+    )
+
+
+def _strict_routing_block(
+    *,
+    fallback: str = "never",
+    max_attempts: str | None = "3",
+    emergency: str | None = "codex",
+    implementation: str = "[cursor, antigravity, pi]",
+    repair: str = "[cursor, antigravity, pi]",
+    test: str = "[pi, antigravity, cursor]",
+    review: str = "[pi, codex]",
+    extra_profiles: str = "",
+) -> str:
+    lines = [
+        "schema: agent-execution/v1",
+        "",
+        "execution:",
+        "  fresh_process_per_unit: true",
+        "  permission: run-everything",
+        f"  fallback_to_host: {fallback}",
+        "  max_parallel: 1",
+        "  timeout_seconds: 60",
+        "  summary_bytes: 1024",
+        "  raw_output_bytes: 4096",
+    ]
+    if max_attempts is not None:
+        lines.append(f"  max_attempts_per_profile: {max_attempts}")
+    if emergency is not None:
+        lines.append(f"  emergency_profile: {emergency}")
+    lines.extend(
+        [
+            "",
+            "roles:",
+            f"  implementation: {implementation}",
+            f"  repair: {repair}",
+            f"  test: {test}",
+            f"  review: {review}",
+            "",
+            "profiles:",
+            "  cursor:",
+            "    adapter: cursor",
+            "    executable: cursor-agent",
+            "    model: auto",
+            "    enabled: true",
+            "",
+            "  antigravity:",
+            "    adapter: antigravity",
+            "    executable: agy",
+            "    model: auto",
+            "    enabled: true",
+            "",
+            "  pi:",
+            "    adapter: pi",
+            "    executable: pi",
+            "    model: auto",
+            "    enabled: true",
+            "",
+            "  codex:",
+            "    adapter: codex",
+            "    executable: codex",
+            "    model: auto",
+            "    enabled: true",
+        ]
+    )
+    if extra_profiles:
+        lines.append(extra_profiles.rstrip("\n"))
+    return "\n".join(lines) + "\n"
+
+
+def _profile_lines(name: str, adapter: str, enabled: str) -> str:
+    return (
+        f"  {name}:\n"
+        f"    adapter: {adapter}\n"
+        f"    executable: {name}\n"
+        "    model: auto\n"
+        f"    enabled: {enabled}"
+    )
+
+
+def _emergency_block(
+    *,
+    include_profile: bool = True,
+    adapter: str = "codex",
+    enabled: str = "true",
+    implementation: str = "[cursor, antigravity, pi]",
+    repair: str = "[cursor, antigravity, pi]",
+    test: str = "[pi, antigravity, cursor]",
+    review: str = "[pi]",
+) -> str:
+    block = MINIMAL_BLOCK.replace(
+        "  raw_output_bytes: 10485760\n",
+        "  raw_output_bytes: 10485760\n"
+        "  max_attempts_per_profile: 3\n"
+        "  emergency_profile: codex\n",
+    )
+    block = block.replace(
+        "implementation: [cursor, antigravity, pi]", f"implementation: {implementation}"
+    )
+    block = block.replace("repair: [cursor, antigravity, pi]", f"repair: {repair}")
+    block = block.replace("test: [pi, antigravity, cursor]", f"test: {test}")
+    block = block.replace("review: [pi]", f"review: {review}")
+    if not include_profile:
+        return block
+    return block.replace(
+        "    enabled: true\n",
+        "    enabled: true\n"
+        "\n"
+        "  codex:\n"
+        f"    adapter: {adapter}\n"
+        "    executable: codex\n"
+        "    model: auto\n"
+        f"    enabled: {enabled}\n",
+        1,
+    )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_task.py

@@ -728,7 +728,23 @@ class TaskLifecycleTests(unittest.TestCase):
             "| 待定义 | — | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — |",
             agent_receipts,
         )
-        self.assertIn("host-fallback-authorized", agent_receipts)
+        self.assertIn("Cursor、Antigravity（agy）和 PI", agent_receipts)
+        self.assertIn("最多 3 次全新尝试", agent_receipts)
+        self.assertIn("一次全新的 Codex CLI 紧急尝试", agent_receipts)
+        self.assertIn("规划、方案设计、实施计划或独立 Review", agent_receipts)
+        self.assertIn("显式 fail-stop", agent_receipts)
+        self.assertIn("不因推理时长、工具次数、规划方式或首次写文件时间判定失败", agent_receipts)
+        self.assertIn("stdin/stdout/stderr 均无活动的空闲超时", agent_receipts)
+        self.assertIn("自主检查、计划、测试并迭代", agent_receipts)
+        self.assertIn("scripts/agent_exec.py apply", agent_receipts)
+        self.assertIn("pre-commit admit", agent_receipts)
+        self.assertIn(".task/**", agent_receipts)
+        self.assertIn("docs/superpowers/specs/**", agent_receipts)
+        self.assertIn("docs/superpowers/plans/**", agent_receipts)
+        self.assertIn("fallback 列只记录该行是否为 Codex 紧急尝试", agent_receipts)
+        self.assertIn("不授权 host 直接撰写该单元", agent_receipts)
+        self.assertIn("安全耗尽不发出 host-fallback-authorized", agent_receipts)
+        self.assertIn("终态为 unavailable 或 blocked", agent_receipts)
         for field in ("当前阶段", "当前增量", "方案引用", "实施计划引用", "测试计划引用"):
             self.assertIn(field + "：", card)
         self.assertIn("## 交付调度\n", card)