agent-exec transport 8c36954f5b4fe6129e117666f9114c7b/7cc66ad08d16
This commit is contained in:
agent-exec-transport
@@ -56,7 +56,7 @@ policy/delivery.md。
|
||||
都没有活动的空闲超时，归入 `failed-unknown`。
|
||||
- 一次尝试只在显式终态失败、进程退出、总超时或上述空闲超时时结束。CLI 可在目标、范围与安全边界内
|
||||
自主检查、计划、测试并迭代。
|
||||
- 实施变更只经成功 transport 由 `scripts/agent_exec.py apply` 集成，或由 pre-commit `admit` 按精确内容
|
||||
- 实施变更只经成功 transport 由 `scripts/agent-exec apply` 集成，或由 pre-commit `admit` 按精确内容
|
||||
做本地完整性核对。本地 HMAC 历史与该钩子只是本地完整性/篡改证据（SELF），不是 GATE、可信身份，
|
||||
也不防同一主机 OS 用户；本地记录缺失或无效仍拒绝，交付信任只来自精确 HEAD 的 CI 或独立 REPLAY。
|
||||
host 直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**` 与
|
||||
|
||||
+1
-1
@@ -20,7 +20,7 @@
|
||||
| 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；实施/修复/测试正常候选仅为 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 |
|
||||
| AGENT-EXECUTION.md + agent-exec | 项目执行路由 | 本地 SELF：parser/adapter/controller 假 CLI 契约由共享执行器自身测试强制（Development-Manager internal/agentexec，经 deps/development-manager 钉住），并由 CI 中的 `scripts/agent-exec validate --repo .` 覆盖；本地薄入口拒绝路径由 tests/test_agent_exec_entry.py 强制；已安装 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 |
|
||||
| core/baseline.md「规范层级」+ documentation.md「记忆分层/机制化」（draft.9） | 层级效力、记忆权威、经验机制化 | manual：本仓 policy/ 仅收紧不放宽；记忆四层按条文执行（任务卡=任务记忆，documents/records=项目记忆，Dev 仓 AGENTS.md 指针已就位） |
|
||||
|
||||
+1
-1
@@ -109,7 +109,7 @@ agy `stream-json` 的 `result` 事件或 `json` 输出对象、PI `json` 模式
|
||||
规范 Git、推送、PR、合入、部署或清理权威。
|
||||
|
||||
实施结果进入当前分支，必须来自一次成功 transport，由 host 执行
|
||||
`scripts/agent_exec.py apply` 在干净分支上快进；或者本次暂存内容通过 pre-commit 调用的
|
||||
`scripts/agent-exec apply` 在干净分支上快进；或者本次暂存内容通过 pre-commit 调用的
|
||||
`admit`：非豁免路径与该成功 transport 的路径和精确内容一致，且 transport 的父提交是当前 HEAD。
|
||||
apply/admit 依赖的本地 HMAC 执行历史与 checkout 钩子只是本地完整性/篡改证据（SELF）：同一主机
|
||||
OS 用户可读取密钥或绕过钩子，因此它不是 GATE、可信身份或防恶意主体的保护。本地记录缺失或无效
|
||||
|
||||
@@ -24,7 +24,7 @@
|
||||
都不是结束条件。CLI 可在目标、范围与安全边界内自主检查、计划、测试并迭代。路由细则见
|
||||
[迭代交付](delivery.md)。
|
||||
|
||||
实施变更进入仓库，只通过成功 transport 的 `scripts/agent_exec.py apply`，或 pre-commit `admit`
|
||||
实施变更进入仓库，只通过成功 transport 的 `scripts/agent-exec apply`，或 pre-commit `admit`
|
||||
对暂存内容的本地完整性核对。本地 HMAC 执行历史与该钩子只记作本地完整性/篡改证据（SELF），
|
||||
不是 GATE、可信身份，也不防同一主机 OS 用户；本地记录缺失或无效仍拒绝，交付信任来自精确 HEAD 的
|
||||
CI 或独立 REPLAY。host 可以直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**`
|
||||
|
||||
@@ -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）；模板默认连续执行，仅真实决策点需人工，没有全流程授权强制；实施/修复/测试仅 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/平台准入 |
|
||||
| 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 来源表述，及钩子不自动写入；`scripts/agent-exec validate`（CI）与 parser·adapter·controller 假 CLI 契约（Development-Manager internal/agentexec 测试，经 deps/development-manager 钉住）为本地 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` 固定对象、权限、覆盖拒绝、原样导入与幂等 |
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,702 +0,0 @@
|
||||
"""Built-in and controlled-custom CLI adapter registry for agent-exec.
|
||||
|
||||
Builds direct argv only (never shell). Probes local executable/version/help;
|
||||
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,
|
||||
SECRET_VALUE_RE,
|
||||
ExecutionConfig,
|
||||
Profile,
|
||||
)
|
||||
|
||||
_PROBE_TIMEOUT_SECONDS = 5
|
||||
# Conservative bound for embedding unit packets into prompts (policy-compatible).
|
||||
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",
|
||||
"${result_path}": "result_path",
|
||||
"${model}": "model",
|
||||
}
|
||||
|
||||
_CUSTOM_OUTPUT_ARGV: dict[str, tuple[str, ...]] = {
|
||||
"json": ("--output-format", "json"),
|
||||
"stream-json": ("--output-format", "stream-json"),
|
||||
"jsonl": ("--output-format", "jsonl"),
|
||||
"text": (),
|
||||
}
|
||||
|
||||
# Substrings that must appear in --help before a built-in is available.
|
||||
_REQUIRED_HELP_MARKERS: dict[str, tuple[str, ...]] = {
|
||||
"cursor": (
|
||||
"--print",
|
||||
"--force",
|
||||
"--sandbox",
|
||||
"--trust",
|
||||
"--workspace",
|
||||
"--model",
|
||||
"--output-format",
|
||||
),
|
||||
"antigravity": (
|
||||
"--print",
|
||||
"--dangerously-skip-permissions",
|
||||
"--mode",
|
||||
"--model",
|
||||
"--output-format",
|
||||
),
|
||||
"claude": (
|
||||
"--print",
|
||||
"--permission-mode",
|
||||
"--model",
|
||||
"--output-format",
|
||||
),
|
||||
"codex": (
|
||||
"exec",
|
||||
"--sandbox",
|
||||
"workspace-write",
|
||||
"--config",
|
||||
"--model",
|
||||
"--cd",
|
||||
"--json",
|
||||
"--output-last-message",
|
||||
),
|
||||
"pi": ("--print", "--approve", "--model", "--mode"),
|
||||
"kimi": ("--prompt", "--auto", "--model", "--output-format"),
|
||||
"opencode": (
|
||||
"--print",
|
||||
"--yes",
|
||||
"--workspace",
|
||||
"--output-format",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AdapterSpec:
|
||||
name: str
|
||||
noninteractive_argv: tuple[str, ...]
|
||||
run_everything_argv: tuple[str, ...]
|
||||
workspace_flag: str | None
|
||||
model_flag: str | None
|
||||
output_argv: tuple[str, ...]
|
||||
prompt_transport: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AdapterProbe:
|
||||
available: bool
|
||||
reason: str | None
|
||||
executable: str
|
||||
observed_version: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Invocation:
|
||||
argv: tuple[str, ...]
|
||||
cwd: Path
|
||||
stdin: bytes | None
|
||||
sanitized_display: tuple[str, ...]
|
||||
|
||||
|
||||
BUILTIN_SPECS: dict[str, AdapterSpec] = {
|
||||
"cursor": AdapterSpec(
|
||||
name="cursor",
|
||||
noninteractive_argv=("-p",),
|
||||
run_everything_argv=("--force", "--sandbox", "enabled", "--trust"),
|
||||
workspace_flag="--workspace",
|
||||
model_flag="--model",
|
||||
output_argv=("--output-format", "stream-json"),
|
||||
prompt_transport="positional",
|
||||
),
|
||||
"antigravity": AdapterSpec(
|
||||
name="antigravity",
|
||||
noninteractive_argv=(),
|
||||
run_everything_argv=(
|
||||
"--dangerously-skip-permissions",
|
||||
"--mode",
|
||||
"auto",
|
||||
),
|
||||
workspace_flag=None,
|
||||
model_flag="--model",
|
||||
output_argv=("--output-format", "stream-json"),
|
||||
prompt_transport="print",
|
||||
),
|
||||
"claude": AdapterSpec(
|
||||
name="claude",
|
||||
noninteractive_argv=("-p",),
|
||||
run_everything_argv=("--permission-mode", "bypassPermissions"),
|
||||
workspace_flag=None,
|
||||
model_flag="--model",
|
||||
output_argv=("--output-format", "stream-json"),
|
||||
prompt_transport="positional",
|
||||
),
|
||||
"codex": AdapterSpec(
|
||||
name="codex",
|
||||
noninteractive_argv=("exec",),
|
||||
run_everything_argv=(
|
||||
"--sandbox",
|
||||
"workspace-write",
|
||||
"-c",
|
||||
"approval_policy=never",
|
||||
),
|
||||
workspace_flag="--cd",
|
||||
model_flag="--model",
|
||||
output_argv=("--json", "--output-last-message"),
|
||||
prompt_transport="positional",
|
||||
),
|
||||
"pi": AdapterSpec(
|
||||
name="pi",
|
||||
noninteractive_argv=("-p",),
|
||||
run_everything_argv=("--approve",),
|
||||
workspace_flag=None,
|
||||
model_flag="--model",
|
||||
output_argv=("--mode", "json"),
|
||||
prompt_transport="positional",
|
||||
),
|
||||
"kimi": AdapterSpec(
|
||||
name="kimi",
|
||||
noninteractive_argv=(),
|
||||
run_everything_argv=("--auto",),
|
||||
workspace_flag=None,
|
||||
model_flag="--model",
|
||||
output_argv=("--output-format", "stream-json"),
|
||||
prompt_transport="flag",
|
||||
),
|
||||
"opencode": AdapterSpec(
|
||||
name="opencode",
|
||||
noninteractive_argv=("--print",),
|
||||
run_everything_argv=("--yes",),
|
||||
workspace_flag="--workspace",
|
||||
model_flag="--model",
|
||||
output_argv=("--output-format", "json"),
|
||||
prompt_transport="positional",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def probe_adapter(profile: Profile, config: ExecutionConfig) -> AdapterProbe:
|
||||
"""Local executable/version/help probe; never contacts a provider."""
|
||||
del config # reserved for custom-adapter-aware probes in later units
|
||||
executable = profile.executable
|
||||
resolved = shutil.which(executable)
|
||||
if resolved is None:
|
||||
return AdapterProbe(
|
||||
available=False,
|
||||
reason="executable not found on PATH",
|
||||
executable=executable,
|
||||
observed_version=None,
|
||||
)
|
||||
|
||||
version = _bounded_probe_output(resolved, ["--version"])
|
||||
if version is None:
|
||||
version = _bounded_probe_output(resolved, ["-v"])
|
||||
if version is None:
|
||||
version = _bounded_probe_output(resolved, ["-V"])
|
||||
|
||||
help_text = _bounded_probe_output(resolved, ["--help"]) or ""
|
||||
help_text += "\n" + (_bounded_probe_output(resolved, ["-h"]) or "")
|
||||
# Codex documents exec options under the subcommand.
|
||||
if profile.adapter == "codex":
|
||||
help_text += "\n" + (
|
||||
_bounded_probe_output(resolved, ["exec", "--help"]) or ""
|
||||
)
|
||||
|
||||
markers = _REQUIRED_HELP_MARKERS.get(profile.adapter)
|
||||
if markers is not None:
|
||||
missing = [m for m in markers if not _help_has_marker(help_text, m)]
|
||||
if missing:
|
||||
return AdapterProbe(
|
||||
available=False,
|
||||
reason=(
|
||||
"unavailable: missing required capability flags in help: "
|
||||
+ ", ".join(missing)
|
||||
),
|
||||
executable=resolved,
|
||||
observed_version=version,
|
||||
)
|
||||
|
||||
if profile.adapter == "custom":
|
||||
# Custom adapters are validated at config load; help is advisory only.
|
||||
pass
|
||||
|
||||
return AdapterProbe(
|
||||
available=True,
|
||||
reason=None,
|
||||
executable=resolved,
|
||||
observed_version=version,
|
||||
)
|
||||
|
||||
|
||||
def build_invocation(
|
||||
profile: Profile,
|
||||
config: ExecutionConfig,
|
||||
workspace: Path,
|
||||
unit_packet: Path,
|
||||
result_path: Path,
|
||||
) -> Invocation:
|
||||
"""Compose a direct argv invocation; never uses a shell."""
|
||||
workspace = Path(workspace)
|
||||
unit_packet = Path(unit_packet)
|
||||
result_path = Path(result_path)
|
||||
packet_text = _read_unit_packet_text(unit_packet)
|
||||
prompt = _build_bounded_unit_prompt(packet_text)
|
||||
|
||||
if profile.adapter == "custom":
|
||||
return _build_custom_invocation(
|
||||
profile,
|
||||
config,
|
||||
workspace,
|
||||
unit_packet,
|
||||
result_path,
|
||||
packet_text=packet_text,
|
||||
prompt=prompt,
|
||||
)
|
||||
|
||||
spec = BUILTIN_SPECS.get(profile.adapter)
|
||||
if spec is None:
|
||||
raise ValueError(f"unknown adapter: {profile.adapter!r}")
|
||||
|
||||
argv: list[str] = [profile.executable]
|
||||
argv.extend(spec.noninteractive_argv)
|
||||
argv.extend(spec.run_everything_argv)
|
||||
if spec.workspace_flag is not None:
|
||||
argv.extend([spec.workspace_flag, str(workspace)])
|
||||
if profile.model is not None and spec.model_flag is not None:
|
||||
argv.extend([spec.model_flag, profile.model])
|
||||
argv.extend(spec.output_argv)
|
||||
if spec.name == "codex":
|
||||
# --output-last-message requires the result path as the next argv item.
|
||||
argv.append(str(result_path))
|
||||
|
||||
stdin: bytes | None = None
|
||||
prompt_values: set[str] = set()
|
||||
if spec.prompt_transport == "positional":
|
||||
argv.append(prompt)
|
||||
prompt_values.add(prompt)
|
||||
elif spec.prompt_transport == "stdin":
|
||||
stdin = packet_text.encode("utf-8")
|
||||
elif spec.prompt_transport == "flag":
|
||||
argv.extend(["--prompt", prompt])
|
||||
prompt_values.add(prompt)
|
||||
elif spec.prompt_transport == "print":
|
||||
# agy 1.3.5 consumes the argv item immediately after --print.
|
||||
argv.extend(["--print", prompt])
|
||||
prompt_values.add(prompt)
|
||||
else:
|
||||
raise ValueError(f"unsupported prompt_transport: {spec.prompt_transport!r}")
|
||||
|
||||
argv_t = tuple(argv)
|
||||
return Invocation(
|
||||
argv=argv_t,
|
||||
cwd=workspace,
|
||||
stdin=stdin,
|
||||
sanitized_display=_sanitize_display(argv_t, prompt_values=prompt_values),
|
||||
)
|
||||
|
||||
|
||||
def _build_custom_invocation(
|
||||
profile: Profile,
|
||||
config: ExecutionConfig,
|
||||
workspace: Path,
|
||||
unit_packet: Path,
|
||||
result_path: Path,
|
||||
*,
|
||||
packet_text: str,
|
||||
prompt: str,
|
||||
) -> Invocation:
|
||||
custom = config.custom_adapters.get(profile.name)
|
||||
if custom is None:
|
||||
raise ValueError(
|
||||
f"profile {profile.name!r} uses custom adapter without definition"
|
||||
)
|
||||
|
||||
values = {
|
||||
"workspace": str(workspace),
|
||||
# Placeholder expands to the packet path, not the body.
|
||||
"unit_packet": str(unit_packet),
|
||||
"result_path": str(result_path),
|
||||
"model": profile.model or "",
|
||||
}
|
||||
|
||||
argv: list[str] = [profile.executable]
|
||||
argv.extend(_expand_argv(custom.base_argv, values, omit_model=False))
|
||||
argv.extend(_expand_argv(custom.run_everything_argv, values, omit_model=False))
|
||||
argv.extend(_expand_argv(custom.workspace_argv, values, omit_model=False))
|
||||
if profile.model is not None:
|
||||
argv.extend(_expand_argv(custom.model_argv, values, omit_model=False))
|
||||
# model is None: omit the complete model flag/value pair (entire model_argv).
|
||||
argv.extend(_CUSTOM_OUTPUT_ARGV.get(custom.output_format, ()))
|
||||
|
||||
stdin: bytes | None = None
|
||||
prompt_values: set[str] = set()
|
||||
if custom.prompt_transport == "stdin":
|
||||
stdin = packet_text.encode("utf-8")
|
||||
elif custom.prompt_transport == "positional":
|
||||
argv.append(prompt)
|
||||
prompt_values.add(prompt)
|
||||
elif custom.prompt_transport == "flag":
|
||||
argv.extend(["--prompt", prompt])
|
||||
prompt_values.add(prompt)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"unsupported prompt_transport: {custom.prompt_transport!r}"
|
||||
)
|
||||
|
||||
argv_t = tuple(argv)
|
||||
return Invocation(
|
||||
argv=argv_t,
|
||||
cwd=workspace,
|
||||
stdin=stdin,
|
||||
sanitized_display=_sanitize_display(argv_t, prompt_values=prompt_values),
|
||||
)
|
||||
|
||||
|
||||
def _read_unit_packet_text(path: Path) -> str:
|
||||
try:
|
||||
raw = path.read_bytes()
|
||||
except OSError as exc:
|
||||
raise ValueError(f"unable to read unit packet: {path}") from exc
|
||||
if len(raw) > MAX_UNIT_PACKET_BYTES:
|
||||
raise ValueError(
|
||||
f"unit packet exceeds size limit of {MAX_UNIT_PACKET_BYTES} bytes"
|
||||
)
|
||||
try:
|
||||
return raw.decode("utf-8")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise ValueError("unit packet is not valid UTF-8") from exc
|
||||
|
||||
|
||||
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],
|
||||
*,
|
||||
omit_model: bool,
|
||||
) -> list[str]:
|
||||
out: list[str] = []
|
||||
for item in items:
|
||||
if item in ALLOWED_PLACEHOLDERS:
|
||||
key = _PLACEHOLDER_MAP[item]
|
||||
if key == "model" and (omit_model or not values.get("model")):
|
||||
continue
|
||||
out.append(values[key])
|
||||
continue
|
||||
if "${" in item:
|
||||
# Config validation already rejects non-whole-item placeholders.
|
||||
raise ValueError(f"placeholder must fill whole argv item: {item!r}")
|
||||
out.append(item)
|
||||
return out
|
||||
|
||||
|
||||
def _help_has_marker(help_text: str, marker: str) -> bool:
|
||||
if marker in help_text:
|
||||
return True
|
||||
# Accept short forms when long forms are required markers.
|
||||
short_aliases = {
|
||||
"--print": ("-p",),
|
||||
"--force": ("-f",),
|
||||
"--model": ("-m",),
|
||||
"--cd": ("-C",),
|
||||
"--config": ("-c",),
|
||||
"--approve": ("-a",),
|
||||
"--prompt": ("-p",),
|
||||
}
|
||||
for alias in short_aliases.get(marker, ()):
|
||||
if re.search(rf"(^|\s){re.escape(alias)}(\s|,|$)", help_text):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _bounded_probe_output(executable: str, args: list[str]) -> str | None:
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
[executable, *args],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=_PROBE_TIMEOUT_SECONDS,
|
||||
shell=False,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return None
|
||||
text = (completed.stdout or "") + (completed.stderr or "")
|
||||
text = text.strip()
|
||||
return text or None
|
||||
|
||||
|
||||
_SENSITIVE_FLAG_RE = re.compile(
|
||||
r"^(?:--api-key|--token|--password|--secret|-H|--header|--authorization)$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _sanitize_display(
|
||||
argv: tuple[str, ...],
|
||||
*,
|
||||
prompt_values: set[str],
|
||||
) -> tuple[str, ...]:
|
||||
out: list[str] = []
|
||||
redact_next = False
|
||||
for item in argv:
|
||||
if item in prompt_values:
|
||||
out.append(UNIT_PACKET_DISPLAY_MARKER)
|
||||
continue
|
||||
if redact_next:
|
||||
out.append("***")
|
||||
redact_next = False
|
||||
continue
|
||||
if _SENSITIVE_FLAG_RE.match(item):
|
||||
out.append(item)
|
||||
redact_next = True
|
||||
continue
|
||||
if SECRET_VALUE_RE.match(item.strip()):
|
||||
out.append("***")
|
||||
continue
|
||||
out.append(item)
|
||||
return tuple(out)
|
||||
|
||||
|
||||
__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",
|
||||
]
|
||||
@@ -1,735 +0,0 @@
|
||||
"""Markdown extraction and constrained YAML parsing for AGENT-EXECUTION.md."""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any
|
||||
|
||||
SCHEMA_V1 = "agent-execution/v1"
|
||||
CONFIG_NAME = "AGENT-EXECUTION.md"
|
||||
BLOCK_OPEN = "```agent-execution"
|
||||
BLOCK_CLOSE = "```"
|
||||
|
||||
BUILTIN_ADAPTERS = frozenset(
|
||||
{"cursor", "antigravity", "claude", "codex", "pi", "kimi", "opencode", "custom"}
|
||||
)
|
||||
REQUIRED_ROLES = ("implementation", "repair", "test", "review")
|
||||
ALLOWED_PLACEHOLDERS = frozenset(
|
||||
{"${workspace}", "${unit_packet}", "${result_path}", "${model}"}
|
||||
)
|
||||
PLACEHOLDER_RE = re.compile(r"\$\{[A-Za-z0-9_]+\}")
|
||||
|
||||
PERMISSION_VALUES = frozenset({"run-everything"})
|
||||
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"})
|
||||
|
||||
MAX_PARALLEL = 32
|
||||
MAX_TIMEOUT_SECONDS = 86400
|
||||
MAX_SUMMARY_BYTES = 1_048_576
|
||||
MAX_RAW_OUTPUT_BYTES = 104_857_600
|
||||
# Omitted idle_timeout_seconds uses this cap, then clamps to timeout_seconds.
|
||||
DEFAULT_IDLE_TIMEOUT_SECONDS = 600
|
||||
|
||||
SHELL_META_RE = re.compile(r"[;&|`$<>(){}]|&&|\|\||\$\(")
|
||||
SECRET_KEY_RE = re.compile(
|
||||
r"(password|secret|token|api[_-]?key|authorization|credential|passwd)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
SECRET_VALUE_RE = re.compile(
|
||||
r"(?i)^(bearer\s+\S+|sk-[A-Za-z0-9_\-]{8,}|[A-Fa-f0-9]{32,})$"
|
||||
)
|
||||
|
||||
ROOT_KEYS = frozenset(
|
||||
{"schema", "execution", "roles", "profiles", "custom_adapters", "integration"}
|
||||
)
|
||||
INTEGRATION_KEYS = frozenset({"exempt_paths", "transport", "admission", "cleanup"})
|
||||
ENTRYPOINT_KEYS = frozenset({"command"})
|
||||
# Mirrors the host-managed prefixes enforced by scripts/agent_exec.py
|
||||
# (_HOST_MANAGED_PREFIXES, the enforcement source). A declared
|
||||
# integration.exempt_paths must equal this set so it cannot widen host authority.
|
||||
HOST_EXEMPT_PATHS = frozenset(
|
||||
{".task/**", "docs/superpowers/specs/**", "docs/superpowers/plans/**"}
|
||||
)
|
||||
EXECUTION_KEYS = frozenset(
|
||||
{
|
||||
"fresh_process_per_unit",
|
||||
"permission",
|
||||
"fallback_to_host",
|
||||
"max_parallel",
|
||||
"timeout_seconds",
|
||||
"idle_timeout_seconds",
|
||||
"summary_bytes",
|
||||
"raw_output_bytes",
|
||||
"max_attempts_per_profile",
|
||||
"emergency_profile",
|
||||
}
|
||||
)
|
||||
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(
|
||||
{
|
||||
"base_argv",
|
||||
"run_everything_argv",
|
||||
"workspace_argv",
|
||||
"model_argv",
|
||||
"prompt_transport",
|
||||
"output_format",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ConfigError(ValueError):
|
||||
"""Raised when AGENT-EXECUTION.md cannot be parsed or fails schema checks."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExecutionPolicy:
|
||||
fresh_process_per_unit: bool
|
||||
permission: str
|
||||
fallback_to_host: str
|
||||
max_parallel: int
|
||||
timeout_seconds: int
|
||||
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)
|
||||
class Profile:
|
||||
name: str
|
||||
adapter: str
|
||||
executable: str
|
||||
model: str | None
|
||||
enabled: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CustomAdapter:
|
||||
name: str
|
||||
base_argv: tuple[str, ...]
|
||||
run_everything_argv: tuple[str, ...]
|
||||
workspace_argv: tuple[str, ...]
|
||||
model_argv: tuple[str, ...]
|
||||
prompt_transport: str
|
||||
output_format: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IntegrationConfig:
|
||||
exempt_paths: tuple[str, ...]
|
||||
transport_command: str
|
||||
admission_command: str
|
||||
cleanup_command: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExecutionConfig:
|
||||
schema: str
|
||||
execution: ExecutionPolicy
|
||||
roles: Mapping[str, tuple[str, ...]]
|
||||
profiles: Mapping[str, Profile]
|
||||
custom_adapters: Mapping[str, CustomAdapter]
|
||||
integration: IntegrationConfig | None = None
|
||||
|
||||
|
||||
def load_config(repo: Path) -> ExecutionConfig:
|
||||
"""Load and validate the single agent-execution block from repo/AGENT-EXECUTION.md."""
|
||||
path = Path(repo) / CONFIG_NAME
|
||||
if not path.is_file():
|
||||
raise ConfigError(f"missing {CONFIG_NAME}")
|
||||
text = path.read_text(encoding="utf-8")
|
||||
raw = _parse_subset_yaml(_extract_block(text))
|
||||
return _build_config(raw)
|
||||
|
||||
|
||||
def validate_config(repo: Path) -> list[str]:
|
||||
"""Return an empty list when configuration is valid; raise ConfigError otherwise."""
|
||||
load_config(repo)
|
||||
return []
|
||||
|
||||
|
||||
def _extract_block(text: str) -> str:
|
||||
lines = text.splitlines()
|
||||
starts: list[int] = []
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip() == BLOCK_OPEN:
|
||||
starts.append(i)
|
||||
if not starts:
|
||||
raise ConfigError("missing agent-execution fenced block")
|
||||
if len(starts) > 1:
|
||||
raise ConfigError("duplicate agent-execution fenced blocks")
|
||||
start = starts[0]
|
||||
end = None
|
||||
for j in range(start + 1, len(lines)):
|
||||
if lines[j].strip() == BLOCK_CLOSE:
|
||||
end = j
|
||||
break
|
||||
if end is None:
|
||||
raise ConfigError("unclosed agent-execution fenced block")
|
||||
return "\n".join(lines[start + 1 : end])
|
||||
|
||||
|
||||
def _parse_subset_yaml(text: str) -> dict[str, Any]:
|
||||
if "\t" in text:
|
||||
raise ConfigError("tabs are not allowed; use two-space indentation")
|
||||
lines = text.splitlines()
|
||||
root: dict[str, Any] = {}
|
||||
stack: list[tuple[int, dict[str, Any]]] = [(-2, root)]
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
raw = lines[i]
|
||||
if not raw.strip() or raw.lstrip().startswith("#"):
|
||||
i += 1
|
||||
continue
|
||||
indent = len(raw) - len(raw.lstrip(" "))
|
||||
stripped = raw.lstrip(" ")
|
||||
if stripped.startswith(("- ", "?", "|", ">", "&", "*", "!")):
|
||||
raise ConfigError(f"unsupported YAML construct: {raw.strip()!r}")
|
||||
if re.search(r"(^|\s)<<", stripped):
|
||||
raise ConfigError(f"unsupported YAML merge/alias: {raw.strip()!r}")
|
||||
content = raw[indent:]
|
||||
if ":" not in content:
|
||||
raise ConfigError(f"expected key: value, got {raw.strip()!r}")
|
||||
key, _, rest = content.partition(":")
|
||||
key = key.strip()
|
||||
rest = rest.strip()
|
||||
if not key or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key):
|
||||
raise ConfigError(f"invalid mapping key: {key!r}")
|
||||
_reject_secret_key(key)
|
||||
|
||||
# Pop to parent whose indent is strictly less than current.
|
||||
while len(stack) > 1 and stack[-1][0] >= indent:
|
||||
stack.pop()
|
||||
parent_indent, parent = stack[-1]
|
||||
expected = parent_indent + 2
|
||||
if indent != expected:
|
||||
raise ConfigError(
|
||||
f"indentation error: expected {expected} spaces, got {indent}"
|
||||
)
|
||||
if key in parent:
|
||||
raise ConfigError(f"duplicate mapping key: {key!r}")
|
||||
|
||||
if rest == "":
|
||||
child: dict[str, Any] = {}
|
||||
parent[key] = child
|
||||
stack.append((indent, child))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
value = _parse_scalar(rest, key=key)
|
||||
parent[key] = value
|
||||
i += 1
|
||||
return root
|
||||
|
||||
|
||||
def _parse_scalar(text: str, *, key: str) -> Any:
|
||||
if text.startswith(("|", ">")):
|
||||
raise ConfigError("multiline scalars are not supported")
|
||||
if text.startswith("["):
|
||||
return _parse_flow_sequence(text)
|
||||
if text in ("true", "false"):
|
||||
return text == "true"
|
||||
if text in ("null", "~"):
|
||||
return None
|
||||
if re.fullmatch(r"-?\d+", text):
|
||||
return int(text)
|
||||
if re.fullmatch(r"\d{4}-\d{2}-\d{2}.*", text):
|
||||
raise ConfigError(f"implicit dates are not supported: {text!r}")
|
||||
if text[:1] in ("'", '"'):
|
||||
value = _parse_quoted_string(text)
|
||||
else:
|
||||
if any(ch in text for ch in ":#[]{},"):
|
||||
raise ConfigError(f"unsafe unquoted scalar: {text!r}")
|
||||
value = text
|
||||
_reject_shell_or_secret_value(key, value)
|
||||
return value
|
||||
|
||||
|
||||
def _parse_quoted_string(text: str) -> str:
|
||||
quote = text[0]
|
||||
if quote not in ("'", '"'):
|
||||
raise ConfigError(f"invalid quoted string: {text!r}")
|
||||
if quote == "'":
|
||||
# Single-quoted YAML-ish: only '' escape; keep non-ASCII intact.
|
||||
if not text.endswith("'") or len(text) < 2:
|
||||
raise ConfigError(f"unclosed single-quoted string: {text!r}")
|
||||
inner = text[1:-1]
|
||||
return inner.replace("''", "'")
|
||||
# Double-quoted: exact Python/JSON-compatible escapes via ast.literal_eval.
|
||||
try:
|
||||
value = ast.literal_eval(text)
|
||||
except (SyntaxError, ValueError) as exc:
|
||||
raise ConfigError(f"invalid double-quoted string: {text!r}") from exc
|
||||
if not isinstance(value, str):
|
||||
raise ConfigError(f"quoted scalar must be a string: {text!r}")
|
||||
return value
|
||||
|
||||
|
||||
def _parse_flow_sequence(text: str) -> list[Any]:
|
||||
if not text.startswith("[") or not text.endswith("]"):
|
||||
raise ConfigError(f"invalid flow sequence: {text!r}")
|
||||
inner = text[1:-1].strip()
|
||||
if not inner:
|
||||
return []
|
||||
items: list[Any] = []
|
||||
buf: list[str] = []
|
||||
in_quote: str | None = None
|
||||
escape = False
|
||||
i = 0
|
||||
while i < len(inner):
|
||||
ch = inner[i]
|
||||
if in_quote:
|
||||
buf.append(ch)
|
||||
if escape:
|
||||
escape = False
|
||||
elif ch == "\\" and in_quote == '"':
|
||||
escape = True
|
||||
elif ch == in_quote:
|
||||
in_quote = None
|
||||
i += 1
|
||||
continue
|
||||
if ch in ("'", '"'):
|
||||
in_quote = ch
|
||||
buf.append(ch)
|
||||
i += 1
|
||||
continue
|
||||
if ch == ",":
|
||||
items.append(_parse_scalar("".join(buf).strip(), key="list_item"))
|
||||
buf = []
|
||||
i += 1
|
||||
continue
|
||||
buf.append(ch)
|
||||
i += 1
|
||||
if in_quote:
|
||||
raise ConfigError(f"unclosed quote in flow sequence: {text!r}")
|
||||
if buf and "".join(buf).strip():
|
||||
items.append(_parse_scalar("".join(buf).strip(), key="list_item"))
|
||||
return items
|
||||
|
||||
|
||||
def _reject_secret_key(key: str) -> None:
|
||||
if SECRET_KEY_RE.search(key):
|
||||
raise ConfigError(f"credential-looking key rejected: {key!r}")
|
||||
|
||||
|
||||
def _reject_shell_or_secret_value(key: str, value: Any) -> None:
|
||||
if not isinstance(value, str):
|
||||
return
|
||||
# Whole-item placeholders are validated later for argv contexts.
|
||||
if value in ALLOWED_PLACEHOLDERS:
|
||||
return
|
||||
if SHELL_META_RE.search(value):
|
||||
raise ConfigError(f"shell metacharacter template rejected for {key!r}")
|
||||
if SECRET_VALUE_RE.match(value.strip()):
|
||||
raise ConfigError(f"secret-shaped value rejected for {key!r}")
|
||||
|
||||
|
||||
def _build_config(raw: dict[str, Any]) -> ExecutionConfig:
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError("root must be a mapping")
|
||||
unknown = set(raw) - ROOT_KEYS
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown top-level keys: {sorted(unknown)}")
|
||||
for required in ("schema", "execution", "roles", "profiles"):
|
||||
if required not in raw:
|
||||
raise ConfigError(f"missing required key: {required}")
|
||||
|
||||
schema = raw["schema"]
|
||||
if schema != SCHEMA_V1:
|
||||
raise ConfigError(f"unknown schema: {schema!r}")
|
||||
|
||||
execution = _build_execution(raw["execution"])
|
||||
profiles = _build_profiles(raw["profiles"])
|
||||
custom_adapters = _build_custom_adapters(raw.get("custom_adapters", {}))
|
||||
roles = _build_roles(raw["roles"], profiles)
|
||||
|
||||
for profile in profiles.values():
|
||||
if profile.adapter == "custom":
|
||||
if profile.name not in custom_adapters:
|
||||
raise ConfigError(
|
||||
f"profile {profile.name!r} uses adapter custom without "
|
||||
f"custom_adapters.{profile.name}"
|
||||
)
|
||||
elif profile.adapter not in BUILTIN_ADAPTERS - {"custom"}:
|
||||
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)
|
||||
|
||||
integration = None
|
||||
if "integration" in raw:
|
||||
integration = _build_integration(raw["integration"])
|
||||
|
||||
return ExecutionConfig(
|
||||
schema=schema,
|
||||
execution=execution,
|
||||
roles=roles,
|
||||
profiles=profiles,
|
||||
custom_adapters=custom_adapters,
|
||||
integration=integration,
|
||||
)
|
||||
|
||||
|
||||
def _build_execution(raw: Any) -> ExecutionPolicy:
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError("execution must be a mapping")
|
||||
unknown = set(raw) - EXECUTION_KEYS
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown execution keys: {sorted(unknown)}")
|
||||
missing = EXECUTION_REQUIRED - set(raw)
|
||||
if missing:
|
||||
raise ConfigError(f"missing execution keys: {sorted(missing)}")
|
||||
|
||||
fresh = raw["fresh_process_per_unit"]
|
||||
if fresh is not True:
|
||||
raise ConfigError("fresh_process_per_unit must be true")
|
||||
|
||||
permission = raw["permission"]
|
||||
if permission not in PERMISSION_VALUES:
|
||||
raise ConfigError(f"unsupported permission: {permission!r}")
|
||||
fallback = raw["fallback_to_host"]
|
||||
if fallback not in FALLBACK_VALUES:
|
||||
raise ConfigError(f"unsupported fallback_to_host: {fallback!r}")
|
||||
|
||||
max_parallel = _positive_bounded(
|
||||
raw["max_parallel"], "max_parallel", 1, MAX_PARALLEL
|
||||
)
|
||||
timeout_seconds = _positive_bounded(
|
||||
raw["timeout_seconds"], "timeout_seconds", 1, MAX_TIMEOUT_SECONDS
|
||||
)
|
||||
if "idle_timeout_seconds" in raw:
|
||||
idle_timeout_seconds = _positive_bounded(
|
||||
raw["idle_timeout_seconds"],
|
||||
"idle_timeout_seconds",
|
||||
1,
|
||||
timeout_seconds,
|
||||
)
|
||||
else:
|
||||
idle_timeout_seconds = min(DEFAULT_IDLE_TIMEOUT_SECONDS, timeout_seconds)
|
||||
summary_bytes = _positive_bounded(
|
||||
raw["summary_bytes"], "summary_bytes", 1, MAX_SUMMARY_BYTES
|
||||
)
|
||||
raw_output_bytes = _positive_bounded(
|
||||
raw["raw_output_bytes"], "raw_output_bytes", 1, MAX_RAW_OUTPUT_BYTES
|
||||
)
|
||||
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,
|
||||
fallback_to_host=fallback,
|
||||
max_parallel=max_parallel,
|
||||
timeout_seconds=timeout_seconds,
|
||||
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")
|
||||
if value < lo or value > hi:
|
||||
raise ConfigError(f"{name} out of bounds: {value}")
|
||||
return value
|
||||
|
||||
|
||||
def _build_profiles(raw: Any) -> dict[str, Profile]:
|
||||
if not isinstance(raw, dict) or not raw:
|
||||
raise ConfigError("profiles must be a non-empty mapping")
|
||||
out: dict[str, Profile] = {}
|
||||
for name, body in raw.items():
|
||||
if not isinstance(body, dict):
|
||||
raise ConfigError(f"profile {name!r} must be a mapping")
|
||||
unknown = set(body) - PROFILE_KEYS
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown keys in profile {name!r}: {sorted(unknown)}")
|
||||
missing = PROFILE_KEYS - set(body)
|
||||
if missing:
|
||||
raise ConfigError(f"missing keys in profile {name!r}: {sorted(missing)}")
|
||||
adapter = body["adapter"]
|
||||
executable = body["executable"]
|
||||
enabled = body["enabled"]
|
||||
model_raw = body["model"]
|
||||
if not isinstance(adapter, str) or not adapter:
|
||||
raise ConfigError(f"profile {name!r} adapter must be a string")
|
||||
if not isinstance(executable, str) or not executable:
|
||||
raise ConfigError(f"profile {name!r} executable must be a string")
|
||||
if type(enabled) is not bool:
|
||||
raise ConfigError(f"profile {name!r} enabled must be a boolean")
|
||||
if model_raw is None or model_raw == "auto":
|
||||
model: str | None = None
|
||||
elif isinstance(model_raw, str) and model_raw:
|
||||
model = model_raw
|
||||
else:
|
||||
raise ConfigError(f"profile {name!r} model must be a string or auto")
|
||||
out[name] = Profile(
|
||||
name=name,
|
||||
adapter=adapter,
|
||||
executable=executable,
|
||||
model=model,
|
||||
enabled=enabled,
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _build_roles(
|
||||
raw: Any, profiles: Mapping[str, Profile]
|
||||
) -> dict[str, tuple[str, ...]]:
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError("roles must be a mapping")
|
||||
unknown = set(raw) - set(REQUIRED_ROLES)
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown roles: {sorted(unknown)}")
|
||||
missing = set(REQUIRED_ROLES) - set(raw)
|
||||
if missing:
|
||||
raise ConfigError(f"missing roles: {sorted(missing)}")
|
||||
out: dict[str, tuple[str, ...]] = {}
|
||||
for role in REQUIRED_ROLES:
|
||||
value = raw[role]
|
||||
if not isinstance(value, list) or not value:
|
||||
raise ConfigError(f"role {role!r} must be a non-empty list")
|
||||
names: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for item in value:
|
||||
if not isinstance(item, str) or not item:
|
||||
raise ConfigError(f"role {role!r} entries must be strings")
|
||||
if item in seen:
|
||||
raise ConfigError(f"duplicate profile reference in role {role!r}: {item}")
|
||||
if item not in profiles:
|
||||
raise ConfigError(f"unknown profile reference: {item!r}")
|
||||
seen.add(item)
|
||||
names.append(item)
|
||||
out[role] = tuple(names)
|
||||
return out
|
||||
|
||||
|
||||
def _build_custom_adapters(raw: Any) -> dict[str, CustomAdapter]:
|
||||
if raw is None:
|
||||
return {}
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError("custom_adapters must be a mapping")
|
||||
out: dict[str, CustomAdapter] = {}
|
||||
for name, body in raw.items():
|
||||
if not isinstance(body, dict):
|
||||
raise ConfigError(f"custom adapter {name!r} must be a mapping")
|
||||
unknown = set(body) - CUSTOM_ADAPTER_KEYS
|
||||
if unknown:
|
||||
raise ConfigError(
|
||||
f"unknown keys in custom adapter {name!r}: {sorted(unknown)}"
|
||||
)
|
||||
missing = CUSTOM_ADAPTER_KEYS - set(body)
|
||||
if missing:
|
||||
raise ConfigError(
|
||||
f"missing keys in custom adapter {name!r}: {sorted(missing)}"
|
||||
)
|
||||
base_argv = _argv_tuple(body["base_argv"], f"custom_adapters.{name}.base_argv")
|
||||
run_everything_argv = _argv_tuple(
|
||||
body["run_everything_argv"],
|
||||
f"custom_adapters.{name}.run_everything_argv",
|
||||
)
|
||||
workspace_argv = _argv_tuple(
|
||||
body["workspace_argv"], f"custom_adapters.{name}.workspace_argv"
|
||||
)
|
||||
model_argv = _argv_tuple(
|
||||
body["model_argv"], f"custom_adapters.{name}.model_argv"
|
||||
)
|
||||
prompt_transport = body["prompt_transport"]
|
||||
output_format = body["output_format"]
|
||||
if prompt_transport not in PROMPT_TRANSPORTS:
|
||||
raise ConfigError(
|
||||
f"unsupported prompt_transport for {name!r}: {prompt_transport!r}"
|
||||
)
|
||||
if output_format not in OUTPUT_FORMATS:
|
||||
raise ConfigError(
|
||||
f"unsupported output_format for {name!r}: {output_format!r}"
|
||||
)
|
||||
out[name] = CustomAdapter(
|
||||
name=name,
|
||||
base_argv=base_argv,
|
||||
run_everything_argv=run_everything_argv,
|
||||
workspace_argv=workspace_argv,
|
||||
model_argv=model_argv,
|
||||
prompt_transport=prompt_transport,
|
||||
output_format=output_format,
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _build_integration(raw: Any) -> IntegrationConfig:
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError("integration must be a mapping")
|
||||
unknown = set(raw) - INTEGRATION_KEYS
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown keys in integration: {sorted(unknown)}")
|
||||
missing = INTEGRATION_KEYS - set(raw)
|
||||
if missing:
|
||||
raise ConfigError(f"missing keys in integration: {sorted(missing)}")
|
||||
exempt_paths = _exempt_paths_tuple(raw["exempt_paths"])
|
||||
if set(exempt_paths) != HOST_EXEMPT_PATHS:
|
||||
raise ConfigError(
|
||||
"integration.exempt_paths must exactly match the host-managed set "
|
||||
f"{sorted(HOST_EXEMPT_PATHS)}; got {list(exempt_paths)}"
|
||||
)
|
||||
return IntegrationConfig(
|
||||
exempt_paths=exempt_paths,
|
||||
transport_command=_entrypoint_command(raw["transport"], "integration.transport"),
|
||||
admission_command=_entrypoint_command(raw["admission"], "integration.admission"),
|
||||
cleanup_command=_entrypoint_command(raw["cleanup"], "integration.cleanup"),
|
||||
)
|
||||
|
||||
|
||||
def _exempt_paths_tuple(value: Any) -> tuple[str, ...]:
|
||||
if not isinstance(value, list) or not value:
|
||||
raise ConfigError("integration.exempt_paths must be a non-empty list")
|
||||
items: list[str] = []
|
||||
for item in value:
|
||||
if not isinstance(item, str) or not item:
|
||||
raise ConfigError("integration.exempt_paths entries must be non-empty strings")
|
||||
if item.startswith("/") or ".." in PurePosixPath(item).parts:
|
||||
raise ConfigError(
|
||||
f"integration.exempt_paths must be relative without '..': {item!r}"
|
||||
)
|
||||
if item in items:
|
||||
raise ConfigError(f"duplicate integration.exempt_paths entry: {item!r}")
|
||||
items.append(item)
|
||||
return tuple(items)
|
||||
|
||||
|
||||
def _entrypoint_command(raw: Any, label: str) -> str:
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError(f"{label} must be a mapping")
|
||||
unknown = set(raw) - ENTRYPOINT_KEYS
|
||||
if unknown:
|
||||
raise ConfigError(f"unknown keys in {label}: {sorted(unknown)}")
|
||||
missing = ENTRYPOINT_KEYS - set(raw)
|
||||
if missing:
|
||||
raise ConfigError(f"missing keys in {label}: {sorted(missing)}")
|
||||
command = raw["command"]
|
||||
if not isinstance(command, str) or not command.strip():
|
||||
raise ConfigError(f"{label}.command must be a non-empty string")
|
||||
_reject_shell_or_secret_value(f"{label}.command", command)
|
||||
return command
|
||||
|
||||
|
||||
def _argv_tuple(value: Any, label: str) -> tuple[str, ...]:
|
||||
if not isinstance(value, list):
|
||||
raise ConfigError(f"{label} must be a list")
|
||||
items: list[str] = []
|
||||
for item in value:
|
||||
if not isinstance(item, str):
|
||||
raise ConfigError(f"{label} entries must be strings")
|
||||
matches = PLACEHOLDER_RE.findall(item)
|
||||
if matches:
|
||||
if len(matches) != 1 or item != matches[0]:
|
||||
raise ConfigError(
|
||||
f"placeholder must fill whole argv item in {label}: {item!r}"
|
||||
)
|
||||
if matches[0] not in ALLOWED_PLACEHOLDERS:
|
||||
raise ConfigError(f"unknown placeholder {matches[0]!r} in {label}")
|
||||
else:
|
||||
_reject_shell_or_secret_value(label, item)
|
||||
items.append(item)
|
||||
return tuple(items)
|
||||
+1
-1
@@ -264,7 +264,7 @@ ${parent_note}
|
||||
|
||||
## Agent 实施收据
|
||||
|
||||
按根目录 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 直接撰写该单元。原始日志保留在源码外收据路径。
|
||||
按根目录 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 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 | 源码外收据路径 |
|
||||
|---|---|---|---|---|---|---|---|---|---|---|
|
||||
|
||||
Vendored
-68
@@ -1,68 +0,0 @@
|
||||
# U5 真实 Agent CLI 探针
|
||||
|
||||
本夹具描述 host 稍后执行的 opt-in 探针。七个 profile 的真实结果都还是 `NOT_RUN`。这里不记录探针 `PASS`。
|
||||
|
||||
## 门禁
|
||||
|
||||
默认测试不设置下面两个变量，因此不会启动真实 CLI，也不会联系 provider。
|
||||
|
||||
- `RUN_REAL_AGENT_EXEC=1`：显式打开真实探针。其他值（包括 `true`）保持关闭。
|
||||
- `AGENT_EXEC_REAL_PROFILE`：每次恰好一个 profile。允许的名字是 `cursor`、`antigravity`、`claude`、`codex`、`pi`、`kimi`、`opencode`。空值、空白、逗号或空格列表都会拒绝，并且不会启动 CLI。
|
||||
|
||||
一次只跑一个 profile，不要对同一个目标并行开多个探针。
|
||||
|
||||
```bash
|
||||
RUN_REAL_AGENT_EXEC=1 AGENT_EXEC_REAL_PROFILE=cursor \
|
||||
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest \
|
||||
tests.test_agent_exec.RealAgentCliProbeTests.test_opt_in_real_profile_probe -v
|
||||
```
|
||||
|
||||
把 `cursor` 换成要测的那一个 profile。建议顺序：`cursor`，然后 `antigravity`（可执行文件 `agy`），`claude`，`pi`，`codex`，`kimi`。`opencode` 单独跑。
|
||||
|
||||
## 夹具
|
||||
|
||||
探针在临时 Git 仓库里进行，不使用本仓的规范工作区。
|
||||
|
||||
- 分支是非默认分支 `feature/u5-real-probe`。
|
||||
- 配置里只启用被选中的那一个 profile，`model` 为 `auto`（省略模型参数，不把字符串 `auto` 传给 CLI）。
|
||||
- 单元只允许一个文件：`src/u5-marker.txt`。
|
||||
- 文件字节必须是 ASCII `U5-REAL-PROBE-MARKER` 再加一个换行 `0x0A`。
|
||||
|
||||
## 核对
|
||||
|
||||
Host 只在机械检查全部成立时把能力结论记为 `PASS`：
|
||||
|
||||
- transport 树和工作区里的标记内容与上面的字节完全一致。
|
||||
- transport 提交只有一个父提交，且该父提交就是夹具基线 HEAD。
|
||||
- `changed_paths` 只有 `src/u5-marker.txt`。
|
||||
- 成功时保留隔离工作区，`transport_created` 为真，`exact_restore` 为假。
|
||||
- 模型收据只记录实际看得到的状态。原始输出里没有 `model_receipt_status` 时记 `not-observable`，不得编造成 `verified`。观察到 `mismatch` 时结论不是 `PASS`。
|
||||
- 规范检出的 HEAD、分支、状态和索引保持不变，标记文件不得出现在规范检出里。
|
||||
|
||||
## 不可用与回退
|
||||
|
||||
缺失或不兼容的 OpenCode 必须记 `unavailable`，不得记 `PASS`。不兼容指本地 help 缺少 adapter 要求的能力标志；这种情况下不得把 CLI 当成已经跑通。
|
||||
|
||||
唯一候选耗尽时，控制器的运行级终态可以是 `host-fallback-authorized`。那是回退授权，不是探针通过。夹具不会代替 CLI 去写标记文件。
|
||||
|
||||
Provider 配额、认证失败，或进程启动后没有留下标记且规范检出仍然完好，同样记 `unavailable`，不得记 `PASS`。范围外改动、规范检出被改写、标记内容不符，记 `failed`。
|
||||
|
||||
## 源码外收据
|
||||
|
||||
真实 opt-in 运行会保留临时仓库，并把收据复制到 `/tmp/agent-exec-<uid>/u5-probes/`。原始 CLI 日志留在控制器的私有 run 目录。这些文件不进入 Git。
|
||||
|
||||
默认单元测试使用隔离 PATH 上的假 CLI，只验证上述核对逻辑。
|
||||
|
||||
## 仍为 NOT_RUN
|
||||
|
||||
| profile | 探针 | 模型收据 | transport | 变更路径 | 清理/保留 | 规范检出 | 能力结论 |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| cursor | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
| antigravity | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
| claude | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
| codex | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
| pi | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
| kimi | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
| opencode | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN |
|
||||
|
||||
Host 回退演练、完整治理回归、独立 Review、精确 HEAD CI 与合入也不在本夹具的已执行范围内。
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,515 +0,0 @@
|
||||
"""Adapter argv, probe, and custom-expansion contract for agent-exec."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import stat
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from scripts.agent_exec_adapters import (
|
||||
MAX_UNIT_PACKET_BYTES,
|
||||
AdapterProbe,
|
||||
Invocation,
|
||||
build_invocation,
|
||||
probe_adapter,
|
||||
)
|
||||
from scripts.agent_exec_config import (
|
||||
CustomAdapter,
|
||||
ExecutionConfig,
|
||||
ExecutionPolicy,
|
||||
Profile,
|
||||
)
|
||||
|
||||
|
||||
UNIT_PACKET_MARKER = "<unit-packet>"
|
||||
|
||||
|
||||
def _policy() -> ExecutionPolicy:
|
||||
return ExecutionPolicy(
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def _config(
|
||||
profiles: dict[str, Profile],
|
||||
custom_adapters: dict[str, CustomAdapter] | None = None,
|
||||
) -> ExecutionConfig:
|
||||
names = tuple(profiles)
|
||||
return ExecutionConfig(
|
||||
schema="agent-execution/v1",
|
||||
execution=_policy(),
|
||||
roles={
|
||||
"implementation": names,
|
||||
"repair": names[:1],
|
||||
"test": names[:1],
|
||||
"review": names[:1],
|
||||
},
|
||||
profiles=profiles,
|
||||
custom_adapters=custom_adapters or {},
|
||||
)
|
||||
|
||||
|
||||
def _write_fake_cli(
|
||||
directory: Path,
|
||||
name: str,
|
||||
*,
|
||||
help_text: str,
|
||||
version: str = "0.0.1-test",
|
||||
) -> Path:
|
||||
path = directory / name
|
||||
# Shebang must start at column 0; keep help body out of textwrap.dedent.
|
||||
path.write_text(
|
||||
"#!/bin/sh\n"
|
||||
'if [ "$1" = "--version" ] || [ "$1" = "-v" ] || [ "$1" = "-V" ]; then\n'
|
||||
f' echo "{version}"\n'
|
||||
" exit 0\n"
|
||||
"fi\n"
|
||||
"cat <<'EOF'\n"
|
||||
f"{help_text.rstrip()}\n"
|
||||
"EOF\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
path.chmod(path.stat().st_mode | stat.S_IEXEC)
|
||||
return path
|
||||
|
||||
|
||||
def _assert_bounded_prompt(test: unittest.TestCase, prompt: str, packet_body: str) -> None:
|
||||
test.assertIn(packet_body, prompt)
|
||||
test.assertNotEqual(prompt, packet_body)
|
||||
lowered = prompt.lower()
|
||||
test.assertRegex(lowered, r"implement exactly|bounded")
|
||||
test.assertRegex(lowered, r"workspace")
|
||||
test.assertRegex(lowered, r"uncommitted")
|
||||
test.assertRegex(lowered, r"scope")
|
||||
test.assertRegex(lowered, r"structured|final result")
|
||||
|
||||
|
||||
CURSOR_HELP = """
|
||||
Usage: agent [options] [prompt...]
|
||||
-p, --print
|
||||
--model <model>
|
||||
-f, --force
|
||||
--yolo
|
||||
--sandbox <mode> choices: enabled, disabled
|
||||
--trust
|
||||
--workspace <path>
|
||||
--output-format <format> text | json | stream-json
|
||||
"""
|
||||
|
||||
AGY_HELP = """
|
||||
Usage of agy:
|
||||
--print
|
||||
--dangerously-skip-permissions
|
||||
--mode accept-edits, auto, plan
|
||||
--model
|
||||
--output-format text, json, stream-json
|
||||
"""
|
||||
|
||||
CLAUDE_HELP = """
|
||||
Usage: claude [options] [prompt]
|
||||
-p, --print
|
||||
--permission-mode <mode> bypassPermissions, manual, dontAsk
|
||||
--model <model>
|
||||
--output-format <format> text, json, stream-json
|
||||
"""
|
||||
|
||||
CODEX_HELP = """
|
||||
Usage: codex exec [OPTIONS] [PROMPT]
|
||||
exec
|
||||
-s, --sandbox <SANDBOX_MODE> read-only, workspace-write, danger-full-access
|
||||
-c, --config <key=value>
|
||||
-m, --model <MODEL>
|
||||
-C, --cd <DIR>
|
||||
--json
|
||||
-o, --output-last-message <FILE>
|
||||
"""
|
||||
|
||||
PI_HELP = """
|
||||
Usage: pi [options]
|
||||
--print, -p
|
||||
--approve, -a
|
||||
--model <pattern>
|
||||
--mode <mode> text, json, or rpc
|
||||
"""
|
||||
|
||||
KIMI_HELP = """
|
||||
Usage: kimi [options]
|
||||
-p, --prompt <prompt>
|
||||
--auto
|
||||
-m, --model <model>
|
||||
--output-format <format> text, stream-json
|
||||
"""
|
||||
|
||||
|
||||
class AgentExecAdapterTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-adapters-")
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.bindir = self.root / "bin"
|
||||
self.bindir.mkdir()
|
||||
self.workspace = self.root / "workspace"
|
||||
self.workspace.mkdir()
|
||||
self.unit_packet = self.root / "unit.md"
|
||||
self.packet_body = "# unit\nsecret-token-should-not-leak-in-display\n"
|
||||
self.unit_packet.write_text(self.packet_body, encoding="utf-8")
|
||||
self.result_path = self.root / "result.json"
|
||||
self._old_path = os.environ.get("PATH", "")
|
||||
os.environ["PATH"] = f"{self.bindir}{os.pathsep}{self._old_path}"
|
||||
self.addCleanup(self._restore_path)
|
||||
|
||||
def _restore_path(self) -> None:
|
||||
os.environ["PATH"] = self._old_path
|
||||
|
||||
def test_cursor_run_everything_and_auto_model_argv(self) -> None:
|
||||
_write_fake_cli(self.bindir, "cursor-agent", help_text=CURSOR_HELP)
|
||||
profile = Profile(
|
||||
name="cursor",
|
||||
adapter="cursor",
|
||||
executable="cursor-agent",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"cursor": profile})
|
||||
probe = probe_adapter(profile, cfg)
|
||||
self.assertTrue(probe.available, probe.reason)
|
||||
inv = build_invocation(
|
||||
profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertIsInstance(inv, Invocation)
|
||||
self.assertEqual(inv.cwd, self.workspace)
|
||||
self.assertEqual(
|
||||
inv.argv[:-1],
|
||||
(
|
||||
"cursor-agent",
|
||||
"-p",
|
||||
"--force",
|
||||
"--sandbox",
|
||||
"enabled",
|
||||
"--trust",
|
||||
"--workspace",
|
||||
str(self.workspace),
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
),
|
||||
)
|
||||
_assert_bounded_prompt(self, inv.argv[-1], self.packet_body)
|
||||
self.assertNotIn(str(self.unit_packet), inv.argv)
|
||||
self.assertNotIn("--model", inv.argv)
|
||||
self.assertIsNone(inv.stdin)
|
||||
self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER)
|
||||
self.assertNotIn(self.packet_body, " ".join(inv.sanitized_display))
|
||||
self.assertNotIn(inv.argv[-1], inv.sanitized_display)
|
||||
|
||||
def test_antigravity_run_everything_argv(self) -> None:
|
||||
_write_fake_cli(self.bindir, "agy", help_text=AGY_HELP)
|
||||
profile = Profile(
|
||||
name="antigravity",
|
||||
adapter="antigravity",
|
||||
executable="agy",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"antigravity": profile})
|
||||
self.assertTrue(probe_adapter(profile, cfg).available)
|
||||
inv = build_invocation(
|
||||
profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertEqual(
|
||||
inv.argv[:-2],
|
||||
(
|
||||
"agy",
|
||||
"--dangerously-skip-permissions",
|
||||
"--mode",
|
||||
"auto",
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
),
|
||||
)
|
||||
self.assertEqual(inv.argv[-2], "--print")
|
||||
_assert_bounded_prompt(self, inv.argv[-1], self.packet_body)
|
||||
self.assertNotIn("--model", inv.argv)
|
||||
self.assertIsNone(inv.stdin)
|
||||
self.assertEqual(inv.cwd, self.workspace)
|
||||
self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER)
|
||||
self.assertEqual(inv.sanitized_display[-2], "--print")
|
||||
self.assertNotIn(self.packet_body, " ".join(inv.sanitized_display))
|
||||
|
||||
def test_claude_run_everything_argv(self) -> None:
|
||||
_write_fake_cli(self.bindir, "claude", help_text=CLAUDE_HELP)
|
||||
profile = Profile(
|
||||
name="claude",
|
||||
adapter="claude",
|
||||
executable="claude",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"claude": profile})
|
||||
self.assertTrue(probe_adapter(profile, cfg).available)
|
||||
inv = build_invocation(
|
||||
profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertEqual(
|
||||
inv.argv[:-1],
|
||||
(
|
||||
"claude",
|
||||
"-p",
|
||||
"--permission-mode",
|
||||
"bypassPermissions",
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
),
|
||||
)
|
||||
_assert_bounded_prompt(self, inv.argv[-1], self.packet_body)
|
||||
self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER)
|
||||
|
||||
def test_codex_uses_noninteractive_workspace_sandbox(self) -> None:
|
||||
_write_fake_cli(self.bindir, "codex", help_text=CODEX_HELP)
|
||||
profile = Profile(
|
||||
name="codex",
|
||||
adapter="codex",
|
||||
executable="codex",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"codex": profile})
|
||||
self.assertTrue(probe_adapter(profile, cfg).available)
|
||||
inv = build_invocation(
|
||||
profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertEqual(
|
||||
inv.argv[:-1],
|
||||
(
|
||||
"codex",
|
||||
"exec",
|
||||
"--sandbox",
|
||||
"workspace-write",
|
||||
"-c",
|
||||
"approval_policy=never",
|
||||
"--cd",
|
||||
str(self.workspace),
|
||||
"--json",
|
||||
"--output-last-message",
|
||||
str(self.result_path),
|
||||
),
|
||||
)
|
||||
_assert_bounded_prompt(self, inv.argv[-1], self.packet_body)
|
||||
self.assertNotIn("--dangerously-bypass-approvals-and-sandbox", inv.argv)
|
||||
self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER)
|
||||
|
||||
def test_pi_and_kimi_auto_model_omit_model_flag(self) -> None:
|
||||
_write_fake_cli(self.bindir, "pi", help_text=PI_HELP)
|
||||
_write_fake_cli(self.bindir, "kimi", help_text=KIMI_HELP)
|
||||
pi = Profile(
|
||||
name="pi", adapter="pi", executable="pi", model=None, enabled=True
|
||||
)
|
||||
kimi = Profile(
|
||||
name="kimi", adapter="kimi", executable="kimi", model=None, enabled=True
|
||||
)
|
||||
cfg = _config({"pi": pi, "kimi": kimi})
|
||||
pi_inv = build_invocation(
|
||||
pi, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
kimi_inv = build_invocation(
|
||||
kimi, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertEqual(
|
||||
pi_inv.argv[:-1],
|
||||
(
|
||||
"pi",
|
||||
"-p",
|
||||
"--approve",
|
||||
"--mode",
|
||||
"json",
|
||||
),
|
||||
)
|
||||
_assert_bounded_prompt(self, pi_inv.argv[-1], self.packet_body)
|
||||
self.assertNotIn("--model", pi_inv.argv)
|
||||
self.assertEqual(
|
||||
kimi_inv.argv[:-1],
|
||||
(
|
||||
"kimi",
|
||||
"--auto",
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
"--prompt",
|
||||
),
|
||||
)
|
||||
_assert_bounded_prompt(self, kimi_inv.argv[-1], self.packet_body)
|
||||
self.assertNotIn("--model", kimi_inv.argv)
|
||||
self.assertNotIn("-m", kimi_inv.argv)
|
||||
self.assertEqual(kimi_inv.sanitized_display[-1], UNIT_PACKET_MARKER)
|
||||
self.assertNotIn(self.packet_body, " ".join(kimi_inv.sanitized_display))
|
||||
self.assertNotIn(kimi_inv.argv[-1], kimi_inv.sanitized_display)
|
||||
|
||||
def test_missing_opencode_is_unavailable(self) -> None:
|
||||
profile = Profile(
|
||||
name="opencode",
|
||||
adapter="opencode",
|
||||
executable="opencode",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"opencode": profile})
|
||||
# Ensure no opencode on PATH from bindir; real PATH may also lack it.
|
||||
probe = probe_adapter(profile, cfg)
|
||||
self.assertIsInstance(probe, AdapterProbe)
|
||||
self.assertFalse(probe.available)
|
||||
self.assertIsNotNone(probe.reason)
|
||||
|
||||
def test_fixed_model_uses_only_adapter_model_flag(self) -> None:
|
||||
_write_fake_cli(self.bindir, "cursor-agent", help_text=CURSOR_HELP)
|
||||
profile = Profile(
|
||||
name="cursor",
|
||||
adapter="cursor",
|
||||
executable="cursor-agent",
|
||||
model="gpt-5",
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"cursor": profile})
|
||||
inv = build_invocation(
|
||||
profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertIn("--model", inv.argv)
|
||||
model_idx = inv.argv.index("--model")
|
||||
self.assertEqual(inv.argv[model_idx + 1], "gpt-5")
|
||||
self.assertEqual(inv.argv.count("--model"), 1)
|
||||
self.assertNotIn("auto", inv.argv)
|
||||
_assert_bounded_prompt(self, inv.argv[-1], self.packet_body)
|
||||
|
||||
def test_custom_adapter_builds_direct_argv_without_shell(self) -> None:
|
||||
_write_fake_cli(
|
||||
self.bindir,
|
||||
"mycli",
|
||||
help_text="Usage: mycli --yes --cwd --model --output-format",
|
||||
)
|
||||
profile = Profile(
|
||||
name="mycustom",
|
||||
adapter="custom",
|
||||
executable="mycli",
|
||||
model="fixed-model",
|
||||
enabled=True,
|
||||
)
|
||||
custom = CustomAdapter(
|
||||
name="mycustom",
|
||||
base_argv=("run", "${unit_packet}"),
|
||||
run_everything_argv=("--yes",),
|
||||
workspace_argv=("--cwd", "${workspace}"),
|
||||
model_argv=("--model", "${model}"),
|
||||
prompt_transport="stdin",
|
||||
output_format="json",
|
||||
)
|
||||
cfg = _config({"mycustom": profile}, {"mycustom": custom})
|
||||
inv = build_invocation(
|
||||
profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertEqual(
|
||||
inv.argv,
|
||||
(
|
||||
"mycli",
|
||||
"run",
|
||||
str(self.unit_packet),
|
||||
"--yes",
|
||||
"--cwd",
|
||||
str(self.workspace),
|
||||
"--model",
|
||||
"fixed-model",
|
||||
"--output-format",
|
||||
"json",
|
||||
),
|
||||
)
|
||||
self.assertEqual(inv.stdin, self.unit_packet.read_bytes())
|
||||
self.assertNotIn("&&", " ".join(inv.argv))
|
||||
self.assertNotIn("|", " ".join(inv.argv))
|
||||
# Path placeholder is fine; body must not appear in display.
|
||||
self.assertNotIn(self.packet_body.strip(), " ".join(inv.sanitized_display))
|
||||
# auto model must omit the whole model_argv pair
|
||||
auto_profile = Profile(
|
||||
name="mycustom",
|
||||
adapter="custom",
|
||||
executable="mycli",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
auto_inv = build_invocation(
|
||||
auto_profile, cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
self.assertNotIn("--model", auto_inv.argv)
|
||||
self.assertNotIn("auto", auto_inv.argv)
|
||||
|
||||
# Positional custom transport may embed the bounded prompt, but display
|
||||
# must use the stable marker.
|
||||
positional = CustomAdapter(
|
||||
name="mycustom",
|
||||
base_argv=("run",),
|
||||
run_everything_argv=("--yes",),
|
||||
workspace_argv=("--cwd", "${workspace}"),
|
||||
model_argv=("--model", "${model}"),
|
||||
prompt_transport="positional",
|
||||
output_format="json",
|
||||
)
|
||||
pos_cfg = _config({"mycustom": profile}, {"mycustom": positional})
|
||||
pos_inv = build_invocation(
|
||||
profile, pos_cfg, self.workspace, self.unit_packet, self.result_path
|
||||
)
|
||||
_assert_bounded_prompt(self, pos_inv.argv[-1], self.packet_body)
|
||||
self.assertEqual(pos_inv.sanitized_display[-1], UNIT_PACKET_MARKER)
|
||||
self.assertNotIn(pos_inv.argv[-1], pos_inv.sanitized_display)
|
||||
|
||||
def test_unverified_run_everything_capability_is_unavailable(self) -> None:
|
||||
_write_fake_cli(
|
||||
self.bindir,
|
||||
"cursor-agent",
|
||||
help_text="Usage: agent\n -p, --print\n --model <model>\n",
|
||||
)
|
||||
profile = Profile(
|
||||
name="cursor",
|
||||
adapter="cursor",
|
||||
executable="cursor-agent",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"cursor": profile})
|
||||
probe = probe_adapter(profile, cfg)
|
||||
self.assertFalse(probe.available)
|
||||
self.assertIsNotNone(probe.reason)
|
||||
self.assertRegex(probe.reason or "", r"unavailable|missing|capability|flag", re.I)
|
||||
|
||||
def test_rejects_oversized_and_invalid_utf8_unit_packet(self) -> None:
|
||||
_write_fake_cli(self.bindir, "cursor-agent", help_text=CURSOR_HELP)
|
||||
profile = Profile(
|
||||
name="cursor",
|
||||
adapter="cursor",
|
||||
executable="cursor-agent",
|
||||
model=None,
|
||||
enabled=True,
|
||||
)
|
||||
cfg = _config({"cursor": profile})
|
||||
|
||||
oversized = self.root / "oversized.md"
|
||||
oversized.write_bytes(b"x" * (MAX_UNIT_PACKET_BYTES + 1))
|
||||
with self.assertRaises(ValueError) as oversized_ctx:
|
||||
build_invocation(
|
||||
profile, cfg, self.workspace, oversized, self.result_path
|
||||
)
|
||||
self.assertRegex(str(oversized_ctx.exception).lower(), r"size|large|bound|limit")
|
||||
|
||||
invalid = self.root / "invalid.md"
|
||||
invalid.write_bytes(b"\xff\xfe not utf-8")
|
||||
with self.assertRaises(ValueError) as utf8_ctx:
|
||||
build_invocation(
|
||||
profile, cfg, self.workspace, invalid, self.result_path
|
||||
)
|
||||
self.assertRegex(str(utf8_ctx.exception).lower(), r"utf-8|utf8|encoding|decode")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,697 +0,0 @@
|
||||
"""Provenance admission: apply a success transport, or admit exact staged paths."""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
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_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")
|
||||
File diff suppressed because it is too large
Load Diff
+1
-1
@@ -841,7 +841,7 @@ class TaskLifecycleTests(unittest.TestCase):
|
||||
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("scripts/agent-exec apply", agent_receipts)
|
||||
self.assertIn("pre-commit admit", agent_receipts)
|
||||
self.assertIn(".task/**", agent_receipts)
|
||||
self.assertIn("docs/superpowers/specs/**", agent_receipts)
|
||||
|
||||
Reference in New Issue
Block a user