diff --git a/AGENTS.md b/AGENTS.md index 73e0a8f9..00ac747b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,7 +43,10 @@ policy/delivery.md。 - 外部 Agent 在新进程/隔离工作区执行一个有界单元；host 负责计划、拆解、范围审阅、权威验证、规范提交、 PR/CI/合入、证据与任务元数据及安全清理。Worker 不得提交、推送、开或合 PR、部署、改路由或派遣同伴。 - 实施、修复与测试的正常候选只有 Cursor、Antigravity（`agy`）和 PI，顺序以配置为准。每个候选对 - 安全可恢复失败（`unavailable` 或已精确恢复的 `failed-clean`）最多 3 次全新尝试，然后才换下一候选。 + 安全可恢复失败（`unavailable`，或外部无副作用已被机制证明的 `failed-clean`）最多 3 次全新尝试，然后才换下一候选。 + 单元包的 `network=false`/`external_cost=false` 只是声明，不是证明；没有强制隔离或可信对账时， + 已启动的失败尝试即使私有克隆可精确恢复也记 `failed-unknown`。成功还须从该 CLI 实际事件信封解出 + 有效的 `completed` 终态记录（schema 写在 worker 提示词中）；缺失或无效即 fail closed。 三者都安全耗尽后，才允许一次全新的 Codex CLI 紧急尝试。Codex 平时只用于规划、方案设计、实施计划 或独立 Review，不进入这三个正常角色。`fallback_to_host` 为 `never`。安全耗尽不发出 host-fallback-authorized，终态为 unavailable 或 blocked，不授权 host 撰写该单元。旧值 @@ -54,7 +57,9 @@ policy/delivery.md。 - 一次尝试只在显式终态失败、进程退出、总超时或上述空闲超时时结束。CLI 可在目标、范围与安全边界内 自主检查、计划、测试并迭代。 - 实施变更只经成功 transport 由 `scripts/agent_exec.py apply` 集成，或由 pre-commit `admit` 按精确内容 - 证明来源。host 直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**` 与 + 做本地完整性核对。本地 HMAC 历史与该钩子只是本地完整性/篡改证据（SELF），不是 GATE、可信身份， + 也不防同一主机 OS 用户；本地记录缺失或无效仍拒绝，交付信任只来自精确 HEAD 的 CI 或独立 REPLAY。 + host 直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**` 与 `docs/superpowers/plans/**`。host 不直接撰写实施单元。细则见 [policy/delivery.md](policy/delivery.md) 与 [policy/governance.md](policy/governance.md)。 不削弱 goal-first、严格范围、先可用、独立 Review、证据、平台准入与高后果双钥匙规则。 diff --git a/policy/delivery.md b/policy/delivery.md index 925d46f3..e921b171 100644 --- a/policy/delivery.md +++ b/policy/delivery.md @@ -74,8 +74,10 @@ 隔离工作区中尝试外部 Agent。不得跨单元复用同一外部进程，也不得 resume 同一次尝试。 实施、修复与测试的正常候选只有 Cursor、Antigravity（`agy`）和 PI。每个候选对安全可恢复失败 -最多 3 次全新尝试：`unavailable`，或工作区已精确恢复的 `failed-clean`。三次仍属这类失败后才进入 -下一候选。三个正常候选都安全耗尽后，才允许一次全新的 Codex CLI 紧急尝试；这次不按每候选 3 次计算。 +最多 3 次全新尝试：`unavailable`，或外部无副作用已被机制证明的 `failed-clean`。三次仍属这类失败后才进入 +下一候选。单元包的 `network=false` 与 `external_cost=false` 只是声明，不证明外部干净；当前没有强制的 +网络/外部费用隔离或可信对账，所以已启动的失败尝试即使私有克隆可精确恢复，也记为 `failed-unknown` +并保留现场。三个正常候选都安全耗尽后，才允许一次全新的 Codex CLI 紧急尝试；这次不按每候选 3 次计算。 紧急尝试若仍是安全可恢复失败，运行停在 `blocked`。host 不接过该单元继续撰写。 Codex 平时只用于规划、方案设计、实施计划，或独立 Review。执行配置没有单独的 plan 角色； @@ -89,6 +91,12 @@ Codex 平时只用于规划、方案设计、实施计划，或独立 Review。 一次尝试只在显式终态失败、进程退出、总超时或该空闲超时时结束。推理时长、工具次数、规划方式和 首次写文件时间都不是结束条件。CLI 可在该单元的目标、范围与安全边界内自主检查、计划、测试并迭代。 +worker 提示词写明终态记录的 JSON Schema（`terminal_status`、`summary`、`changed_files`、`evidence`、 +`scope_expansion`）。控制器只从该 CLI 实际事件信封解码：Cursor `stream-json` 的 `result` 事件、 +agy `stream-json` 的 `result` 事件或 `json` 输出对象、PI `json` 模式最后一条 assistant `message_end`。 +退出码 0 且有改动时，只有有效的 `completed` 记录才可能成功；记录缺失、无效或仅有散文均记 +`failed-unknown`。`blocked`、`cancelled`、`failed-unknown` 记录直接按该终态停止，不得成功。 + 主模型（host）负责计划、拆解、有界单元包、强制范围、检查与审阅实际 diff、运行权威验证、 规范提交、管理 PR/CI/合入、写入证据与任务管理元数据，以及安全清理。权威验证、集成、合入与 任务卡证据所有权属于 host；worker 的局部自检不能替代任务卡证据、独立 Review、当前 HEAD 的 CI @@ -103,6 +111,9 @@ Codex 平时只用于规划、方案设计、实施计划，或独立 Review。 实施结果进入当前分支，必须来自一次成功 transport，由 host 执行 `scripts/agent_exec.py apply` 在干净分支上快进；或者本次暂存内容通过 pre-commit 调用的 `admit`：非豁免路径与该成功 transport 的路径和精确内容一致，且 transport 的父提交是当前 HEAD。 +apply/admit 依赖的本地 HMAC 执行历史与 checkout 钩子只是本地完整性/篡改证据（SELF）：同一主机 +OS 用户可读取密钥或绕过钩子，因此它不是 GATE、可信身份或防恶意主体的保护。本地记录缺失或无效 +仍拒绝；交付信任只来自精确 HEAD 的必需 CI（GATE）或独立 REPLAY。 host 可以直接编写并提交的豁免只有 `.task/**`、`docs/superpowers/specs/**` 和 `docs/superpowers/plans/**`。除此之外，host 不直接撰写实施单元。 diff --git a/policy/governance.md b/policy/governance.md index 604ee96b..a1f266ec 100644 --- a/policy/governance.md +++ b/policy/governance.md @@ -25,9 +25,12 @@ [迭代交付](delivery.md)。 实施变更进入仓库，只通过成功 transport 的 `scripts/agent_exec.py apply`，或 pre-commit `admit` -对暂存内容的精确来源核对。host 可以直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**` +对暂存内容的本地完整性核对。本地 HMAC 执行历史与该钩子只记作本地完整性/篡改证据（SELF）， +不是 GATE、可信身份，也不防同一主机 OS 用户；本地记录缺失或无效仍拒绝，交付信任来自精确 HEAD 的 +CI 或独立 REPLAY。host 可以直接管理的豁免只有 `.task/**`、`docs/superpowers/specs/**` 与 `docs/superpowers/plans/**`。host 不直接撰写实施单元。`failed-unknown`、`scope-expansion`、 -无效配置、未对账 Git 状态与安全失败都不授权继续路由或代写。 +无效配置、未对账 Git 状态与安全失败都不授权继续路由或代写。`network=false`/`external_cost=false` +声明不证明外部干净，已启动的失败尝试在没有强制隔离或可信对账时记 `failed-unknown`。 ## 启动与恢复 diff --git a/scripts/agent_exec.py b/scripts/agent_exec.py index 2a571b40..80919a38 100644 --- a/scripts/agent_exec.py +++ b/scripts/agent_exec.py @@ -1,4 +1,10 @@ -"""Recoverable agent-exec controller: validate, run, apply transports, admit provenance.""" +"""Recoverable agent-exec controller: validate, run, apply transports, admit staged content. + +Local history MACs and the pre-commit ``admit`` check are local integrity and +tamper evidence only (SELF). They are not GATE, not a trusted identity, and do +not protect against the same OS user. Delivery trust comes from exact-HEAD CI +or an independent REPLAY. +""" from __future__ import annotations import argparse @@ -29,7 +35,11 @@ _REPO_ROOT = Path(__file__).resolve().parents[1] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) -from scripts.agent_exec_adapters import build_invocation, probe_adapter +from scripts.agent_exec_adapters import ( + build_invocation, + decode_terminal_record, + probe_adapter, +) from scripts.agent_exec_config import ( DEFAULT_IDLE_TIMEOUT_SECONDS, SECRET_VALUE_RE, @@ -183,7 +193,10 @@ def main(argv: list[str] | None = None) -> int: p_admit = sub.add_parser( "admit", - help="Admit staged paths backed by a success transport or host exemption", + help=( + "Locally check staged paths against a success transport or host " + "exemption (local integrity evidence, not GATE)" + ), ) p_admit.add_argument("--repo", required=True) @@ -791,7 +804,7 @@ def _run_attempt( pointer_delta = _snapshot_delta(pointer_before, pointer_after) canonical_delta = _snapshot_delta(canonical_before, canonical_after) - worker_result = _load_worker_result(result_path, proc_meta["stdout_tail"]) + worker_result = _load_worker_result(profile.adapter, result_path, proc_meta) summary = _bounded_summary( proc_meta.get("summary_source") or (worker_result or {}).get("summary") or "", cfg.execution.summary_bytes, @@ -922,7 +935,17 @@ def _run_attempt( cleanup_extra={"reason": "worker fail-stop terminal"}, ) - # Never report success from worker prose alone — require exit + admin + in-scope diff + transport. + completed = bool( + worker_result and worker_result.get("terminal_status") == "completed" + ) + if clean_exit and changed and not completed: + return _preserved( + "failed-unknown", + "worker exited 0 with changes but no valid completed terminal record", + cleanup_extra={"reason": "missing or invalid terminal record"}, + ) + + # Success requires exit 0 + completed terminal record + clean admin + in-scope diff + transport. if clean_exit and changed and not out_of_scope: # Recheck path bounds immediately before host-owned transport. try: @@ -975,121 +998,23 @@ def _run_attempt( observed_version=probe.observed_version, ) - if clean_exit and not changed: - # No diff does not prove an authorized external effect was reconciled. - if _external_side_effects_unproven(unit, worker_result): - return _preserved( - "failed-unknown", - ( - "worker exited 0 without in-scope changes; " - "external side effects cannot be reconciled" - ), - cleanup_extra={ - "reason": "network or external_cost authorized without trustworthy receipt", - "network": unit.network, - "external_cost": unit.external_cost, - }, - ) - restored = _restore_worktree(repo, worktree, base["head"]) - if not restored: - return _preserved( - "failed-unknown", - "exact restore failed", - cleanup_extra={"reason": "exact restore failed"}, - ) - return AttemptRecord( - attempt_id=attempt_id, - profile=profile.name, - adapter=profile.adapter, - terminal_status="failed-clean", - reason="worker exited 0 without in-scope changes", - worktree=str(worktree), - pid=proc_meta["pid"], - exit_code=proc_meta["exit_code"], - changed_paths=[], - inventory_paths=[], - raw_log=str(raw_log), - summary=summary, - cleanup={"exact_restore": True}, - observed_version=probe.observed_version, - ) - - # Nonzero / aborted without trustworthy external reconciliation. - if _external_side_effects_unproven(unit, worker_result): - return _preserved( - "failed-unknown", - ( - f"worker failed with exit {proc_meta['exit_code']}; " - "external side effects cannot be reconciled" - ), - cleanup_extra={ - "reason": "network or external_cost authorized without trustworthy receipt", - "network": unit.network, - "external_cost": unit.external_cost, - }, - ) - - # Controlled case: network=false and external_cost=false may become failed-clean - # only after exact worktree restoration proves no unknown authorized external object. - if changed: - restored = _restore_worktree(repo, worktree, base["head"]) - if restored: - return AttemptRecord( - attempt_id=attempt_id, - profile=profile.name, - adapter=profile.adapter, - terminal_status="failed-clean", - reason=f"worker failed with exit {proc_meta['exit_code']}; restored", - worktree=str(worktree), - pid=proc_meta["pid"], - exit_code=proc_meta["exit_code"], - changed_paths=changed, - inventory_paths=list(changed), - raw_log=str(raw_log), - summary=summary, - cleanup={ - "exact_restore": True, - "external_objects_reconciled": True, - "network": False, - "external_cost": False, - }, - observed_version=probe.observed_version, - ) - return _preserved( - "failed-unknown", - "worker failed; exact restore not proven", - ) - - restored = _restore_worktree(repo, worktree, base["head"]) - if not restored: - return _preserved( - "failed-unknown", - "exact restore failed", - cleanup_extra={"reason": "exact restore failed"}, - ) - return AttemptRecord( - attempt_id=attempt_id, - profile=profile.name, - adapter=profile.adapter, - terminal_status="failed-clean", - reason=f"worker failed with exit {proc_meta['exit_code']}", - worktree=str(worktree), - pid=proc_meta["pid"], - exit_code=proc_meta["exit_code"], - changed_paths=[], - inventory_paths=[], - raw_log=str(raw_log), - summary=summary, - cleanup={ - "exact_restore": True, - "external_objects_reconciled": True, - "network": False, - "external_cost": False, + # Launched attempts have no enforced network or external-cost isolation, and + # the unit's declared network/external_cost booleans are not proof. A + # restored private clone therefore cannot make a launched failure clean. + if clean_exit: + reason = "worker exited 0 without in-scope changes" + else: + reason = f"worker failed with exit {proc_meta['exit_code']}" + return _preserved( + "failed-unknown", + f"{reason}; external side effects are not mechanically reconciled", + cleanup_extra={ + "reason": "external side effects not mechanically reconciled", + "declared_network": unit.network, + "declared_external_cost": unit.external_cost, }, - observed_version=probe.observed_version, ) - def _supervise( argv: tuple[str, ...], *, @@ -1113,7 +1038,7 @@ def _supervise( exit_code: int | None = None sig: int | None = None pid: int | None = None - stdout_tail = _ByteTail(OUTPUT_TAIL_BYTES) + stdout_tail = _StdoutTail(OUTPUT_TAIL_BYTES) stderr_tail = _ByteTail(OUTPUT_TAIL_BYTES) # Bounded prefix: raw_cap + exact-secret lookahead. Redact the complete # prefix once, write at most raw_cap, then discard further drained bytes. @@ -1386,6 +1311,7 @@ def _supervise( "signal": sig, "stdout_tail": stdout_text, "stderr_tail": stderr_text, + "terminal_lines": stdout_tail.terminal_lines(), "summary_source": summary_source, } @@ -1408,6 +1334,74 @@ class _ByteTail: return bytes(self._data).decode("utf-8", "replace") +TERMINAL_LINE_MAX_BYTES = 1 << 20 +TERMINAL_LINE_KEEP = 32 +_TERMINAL_ENVELOPE_RE = re.compile( + rb'"(?:type|event)"\s*:\s*"(?:result|message_end)"' +) +_TERMINAL_HINT_RE = re.compile(rb'terminal_status|"is_error"') + + +class _StdoutTail(_ByteTail): + """Byte tail that also keeps recent complete lines that may be terminal envelopes. + + A final envelope can sit behind a later event larger than the byte tail + (for example PI ``agent_end``), so candidate lines are kept separately. + An oversized envelope line is recorded as ``None`` so decoding fails closed. + """ + + def __init__(self, limit: int) -> None: + super().__init__(limit) + self._line = bytearray() + self._overflow_head: bytes | None = None + self._lines: list[bytes | None] = [] + + def extend(self, chunk: bytes) -> None: + super().extend(chunk) + start = 0 + while start <= len(chunk): + nl = chunk.find(b"\n", start) + piece = chunk[start:] if nl == -1 else chunk[start:nl] + self._append_piece(piece) + if nl == -1: + return + self._end_line() + start = nl + 1 + + def _append_piece(self, piece: bytes) -> None: + if self._overflow_head is not None: + return + if len(self._line) + len(piece) > TERMINAL_LINE_MAX_BYTES: + self._overflow_head = bytes(self._line[:256] + piece[:256])[:256] + self._line.clear() + return + self._line.extend(piece) + + def _end_line(self) -> None: + if self._overflow_head is not None: + if _TERMINAL_ENVELOPE_RE.search(self._overflow_head): + self._keep(None) + self._overflow_head = None + return + line = bytes(self._line) + self._line.clear() + if _TERMINAL_ENVELOPE_RE.search(line) or _TERMINAL_HINT_RE.search(line): + self._keep(line) + + def _keep(self, line: bytes | None) -> None: + self._lines.append(line) + if len(self._lines) > TERMINAL_LINE_KEEP: + del self._lines[0] + + def terminal_lines(self) -> list[str | None]: + if self._line or self._overflow_head is not None: + self._end_line() + return [ + None if line is None else line.decode("utf-8", "replace") + for line in self._lines + ] + + def _kill_process_group(pid: int) -> None: try: os.killpg(pid, signal.SIGKILL) @@ -1695,20 +1689,6 @@ def _worktree_pointer_snapshot( return snap -def _external_side_effects_unproven( - unit: UnitPacket, worker_result: dict[str, Any] | None -) -> bool: - """Network or external_cost units cannot be failed-clean without reconciliation. - - A clean exit with no file diff is included. failed-clean still requires exact - restore plus affirmative evidence that no authorized external object remains - unknown. Worker prose is not that evidence; only an explicit trustworthy - reconciliation record would clear the unit, and none is accepted here. - """ - del worker_result # Worker prose alone never proves external reconciliation. - return bool(unit.network or unit.external_cost) - - def _is_within_root(path: Path, root: Path) -> bool: try: path.resolve().relative_to(root.resolve()) @@ -1818,20 +1798,6 @@ def _create_transport_commit( return ref -def _restore_worktree(repo: Path, worktree: Path, base_head: str) -> bool: - """Exact-restore the private clone then remove it after verification.""" - try: - _git(worktree, "reset", "--hard", base_head) - _git(worktree, "clean", "-fdx") - head = _git(worktree, "rev-parse", "HEAD").stdout.strip() - dirty = _git(worktree, "status", "--porcelain").stdout.strip() - ok = head == base_head and not dirty - _safe_remove_worktree(repo, worktree) - return ok - except Exception: - return False - - def _safe_remove_worktree(repo: Path, worktree: Path) -> None: """Remove a private attempt clone directory (not a linked worktree).""" del repo # Private clones are not registered linked worktrees. @@ -1840,29 +1806,21 @@ def _safe_remove_worktree(repo: Path, worktree: Path) -> None: def _load_worker_result( - result_path: Path, stdout_tail: str + adapter: str, result_path: Path, proc_meta: dict[str, Any] ) -> dict[str, Any] | None: + """Return the worker's valid terminal record, or None when absent/invalid.""" + result_text: str | None = None if result_path.is_file(): try: - data = json.loads(result_path.read_text(encoding="utf-8")) - if isinstance(data, dict): - return data - except (OSError, json.JSONDecodeError): - pass - # Try last JSON object in stdout. - text = stdout_tail.strip() - if not text: - return None - for candidate in (text, text[text.rfind("{") :] if "{" in text else ""): - if not candidate.startswith("{"): - continue - try: - data = json.loads(candidate) - if isinstance(data, dict): - return data - except json.JSONDecodeError: - continue - return None + result_text = result_path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + result_text = None + return decode_terminal_record( + adapter, + event_lines=proc_meta.get("terminal_lines") or [], + stdout_tail=proc_meta.get("stdout_tail") or "", + result_text=result_text, + ) def _base_receipt( @@ -1958,7 +1916,7 @@ def _finish_run( ) history_path: Path | None = None try: - history_path, attestation = _append_execution_history( + history_path, integrity_mac = _append_execution_history( repo, receipt, secret_values ) recorded: dict[str, Any] = { @@ -1967,8 +1925,8 @@ def _finish_run( "record_id": receipt.get("run_id"), "run_id": receipt.get("run_id"), } - if attestation: - recorded["attestation"] = attestation + if integrity_mac: + recorded["local_integrity_mac"] = integrity_mac receipt["history"] = recorded except Exception as exc: receipt["history"] = { @@ -2001,12 +1959,12 @@ def _append_execution_history( history_path = history_dir / "execution-history.jsonl" if history_path.is_symlink(): raise RuntimeError("agent-exec history file must not be a symlink") - attestation: str | None = None + integrity_mac: str | None = None if receipt.get("terminal_status") == "success": _bind_success_transport_commit(repo, receipt) - key = _ensure_attestation_key(repo) + key = _ensure_local_integrity_key(repo) record = _execution_history_record(repo, receipt) - line, attestation = _signed_history_line(record, key, secret_values) + line, integrity_mac = _integrity_history_line(record, key, secret_values) else: record = _execution_history_record(repo, receipt) line = _redact_text(_canonical_json(record), secret_values) + "\n" @@ -2027,7 +1985,7 @@ def _append_execution_history( fcntl.flock(fd, fcntl.LOCK_UN) finally: os.close(fd) - return history_path, attestation + return history_path, integrity_mac def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str, Any]: @@ -2335,8 +2293,12 @@ def _valid_transport_ref(ref: object) -> str | None: return ref -_ATTESTATION_KEY_NAME = "attestation.key" -_ATTESTATION_KEY_BYTES = 32 +# Legacy on-disk names keep existing local history readable. The MAC is local +# integrity/tamper evidence only: any process running as the same OS user can +# read the key and forge records, so it is never GATE or a trusted identity. +_LOCAL_INTEGRITY_KEY_NAME = "attestation.key" +_LOCAL_INTEGRITY_FIELD = "attestation" +_LOCAL_INTEGRITY_KEY_BYTES = 32 def _git_common_dir(repo: Path) -> Path: @@ -2363,27 +2325,27 @@ def _canonical_json(data: dict[str, Any]) -> str: return json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":")) -def _history_attestation(key: bytes, record: dict[str, Any]) -> str: - body = {field: value for field, value in record.items() if field != "attestation"} +def _local_integrity_mac(key: bytes, record: dict[str, Any]) -> str: + body = {field: value for field, value in record.items() if field != _LOCAL_INTEGRITY_FIELD} return hmac.new(key, _canonical_json(body).encode("utf-8"), hashlib.sha256).hexdigest() -def _history_attestation_ok(key: bytes, record: dict[str, Any]) -> bool: - presented = record.get("attestation") +def _local_integrity_ok(key: bytes, record: dict[str, Any]) -> bool: + presented = record.get(_LOCAL_INTEGRITY_FIELD) if not isinstance(presented, str) or len(presented) != 64: return False - return hmac.compare_digest(_history_attestation(key, record), presented) + return hmac.compare_digest(_local_integrity_mac(key, record), presented) -def _signed_history_line( +def _integrity_history_line( record: dict[str, Any], key: bytes, secret_values: set[str] ) -> tuple[str, str]: text = _redact_text(_canonical_json(record), secret_values) body = json.loads(text) if not isinstance(body, dict): raise RuntimeError("history record redaction produced non-object JSON") - mac = _history_attestation(key, body) - body["attestation"] = mac + mac = _local_integrity_mac(key, body) + body[_LOCAL_INTEGRITY_FIELD] = mac return _canonical_json(body) + "\n", mac @@ -2456,22 +2418,22 @@ def _locked_agent_exec_dir(repo: Path, *, create: bool) -> Iterator[int | None]: def _read_key_at(dir_fd: int) -> bytes: fd = os.open( - _ATTESTATION_KEY_NAME, + _LOCAL_INTEGRITY_KEY_NAME, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=dir_fd, ) try: current = os.fstat(fd) if not stat.S_ISREG(current.st_mode): - raise RuntimeError("attestation key must be a regular file") + raise RuntimeError("local integrity key must be a regular file") if stat.S_IMODE(current.st_mode) != 0o600: - raise RuntimeError("attestation key mode must be 0600") - if current.st_size != _ATTESTATION_KEY_BYTES: - raise RuntimeError("attestation key must be 32 bytes") + raise RuntimeError("local integrity key mode must be 0600") + if current.st_size != _LOCAL_INTEGRITY_KEY_BYTES: + raise RuntimeError("local integrity key must be 32 bytes") os.lseek(fd, 0, os.SEEK_SET) - key = os.read(fd, _ATTESTATION_KEY_BYTES) - if len(key) != _ATTESTATION_KEY_BYTES: - raise RuntimeError("attestation key must be 32 bytes") + key = os.read(fd, _LOCAL_INTEGRITY_KEY_BYTES) + if len(key) != _LOCAL_INTEGRITY_KEY_BYTES: + raise RuntimeError("local integrity key must be 32 bytes") return key finally: os.close(fd) @@ -2479,63 +2441,63 @@ def _read_key_at(dir_fd: int) -> bytes: def _key_is_symlink(dir_fd: int) -> bool: try: - info = os.stat(_ATTESTATION_KEY_NAME, dir_fd=dir_fd, follow_symlinks=False) + info = os.stat(_LOCAL_INTEGRITY_KEY_NAME, dir_fd=dir_fd, follow_symlinks=False) except FileNotFoundError: return False return stat.S_ISLNK(info.st_mode) -def _read_attestation_key(repo: Path) -> bytes | None: - """Read the local attestation key. Never creates one and never follows links.""" +def _read_local_integrity_key(repo: Path) -> bytes | None: + """Read the local integrity key. Never creates one and never follows links.""" with _locked_agent_exec_dir(repo, create=False) as dir_fd: if dir_fd is None: return None if _key_is_symlink(dir_fd): - raise RuntimeError("attestation key must not be a symlink") + raise RuntimeError("local integrity key must not be a symlink") try: return _read_key_at(dir_fd) except FileNotFoundError: return None except OSError as exc: - raise _symlink_oserror(exc, "attestation key must not be a symlink") from exc + raise _symlink_oserror(exc, "local integrity key must not be a symlink") from exc -def _ensure_attestation_key(repo: Path) -> bytes: +def _ensure_local_integrity_key(repo: Path) -> bytes: """Create or read the 32-byte key under the Git common dir. - This binds controller history to later apply/admit checks. It does not - protect against the same OS user deliberately reading the key file. + The MAC lets later apply/admit checks detect accidental or casual edits of + local history. It is not an identity: the same OS user can read the key. """ with _locked_agent_exec_dir(repo, create=True) as dir_fd: if dir_fd is None: raise RuntimeError("agent-exec history directory is unavailable") if _key_is_symlink(dir_fd): - raise RuntimeError("attestation key must not be a symlink") + raise RuntimeError("local integrity key must not be a symlink") flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC try: - fd = os.open(_ATTESTATION_KEY_NAME, flags, 0o600, dir_fd=dir_fd) + fd = os.open(_LOCAL_INTEGRITY_KEY_NAME, flags, 0o600, dir_fd=dir_fd) except OSError as exc: - raise _symlink_oserror(exc, "attestation key must not be a symlink") from exc + raise _symlink_oserror(exc, "local integrity key must not be a symlink") from exc try: os.fchmod(fd, 0o600) current = os.fstat(fd) if not stat.S_ISREG(current.st_mode): - raise RuntimeError("attestation key must be a regular file") + raise RuntimeError("local integrity key must be a regular file") if current.st_size == 0: - key = os.urandom(_ATTESTATION_KEY_BYTES) + key = os.urandom(_LOCAL_INTEGRITY_KEY_BYTES) written = os.write(fd, key) - if written != _ATTESTATION_KEY_BYTES: - raise RuntimeError("attestation key write was short") + if written != _LOCAL_INTEGRITY_KEY_BYTES: + raise RuntimeError("local integrity key write was short") os.fsync(fd) return key - if current.st_size != _ATTESTATION_KEY_BYTES: - raise RuntimeError("attestation key must be 32 bytes") + if current.st_size != _LOCAL_INTEGRITY_KEY_BYTES: + raise RuntimeError("local integrity key must be 32 bytes") os.lseek(fd, 0, os.SEEK_SET) - key = os.read(fd, _ATTESTATION_KEY_BYTES) - if len(key) != _ATTESTATION_KEY_BYTES: - raise RuntimeError("attestation key must be 32 bytes") + key = os.read(fd, _LOCAL_INTEGRITY_KEY_BYTES) + if len(key) != _LOCAL_INTEGRITY_KEY_BYTES: + raise RuntimeError("local integrity key must be 32 bytes") if stat.S_IMODE(os.fstat(fd).st_mode) != 0o600: - raise RuntimeError("attestation key mode must be 0600") + raise RuntimeError("local integrity key mode must be 0600") return key finally: os.close(fd) @@ -2574,7 +2536,7 @@ def _transport_covers( required: set[str], key: bytes | None, ) -> bool: - if key is None or not _history_attestation_ok(key, record): + if key is None or not _local_integrity_ok(key, record): return False if record.get("schema") != HISTORY_SCHEMA: return False @@ -2587,32 +2549,32 @@ def _transport_covers( return False ref = _valid_transport_ref(transport.get("ref")) parent = transport.get("parent") - signed_commit = transport.get("commit") - if ref is None or not isinstance(parent, str) or not isinstance(signed_commit, str): + recorded_commit = transport.get("commit") + if ref is None or not isinstance(parent, str) or not isinstance(recorded_commit, str): return False - if _resolve_commit(repo, ref) != signed_commit: + if _resolve_commit(repo, ref) != recorded_commit: return False if _resolve_commit(repo, parent) != head: return False - if _commit_parents(repo, signed_commit) != [head]: + if _commit_parents(repo, recorded_commit) != [head]: return False changed = record.get("changed_paths") if not isinstance(changed, list) or any(not isinstance(path, str) for path in changed): return False - diff_paths = _diff_paths(repo, head, signed_commit) + diff_paths = _diff_paths(repo, head, recorded_commit) if _non_exempt(changed) != required or _non_exempt(diff_paths) != required: return False - return _index_matches_commit(repo, signed_commit, required) + return _index_matches_commit(repo, recorded_commit, required) -def _apply_signed_history_ok( +def _apply_local_integrity_ok( repo: Path, receipt: dict[str, Any], *, commit: str, base_head: str, ) -> bool: - key = _read_attestation_key(repo) + key = _read_local_integrity_key(repo) if key is None: return False run_id = receipt.get("run_id") @@ -2648,7 +2610,7 @@ def _apply_signed_history_ok( continue if rec_transport.get("commit") != commit: continue - if _history_attestation_ok(key, record): + if _local_integrity_ok(key, record): return True return False @@ -2717,13 +2679,13 @@ def cmd_apply(repo: Path, receipt_path: Path) -> int: return _reject("apply: mismatched transport", secrets) try: - authorized = _apply_signed_history_ok( + authorized = _apply_local_integrity_ok( repo, receipt, commit=commit, base_head=resolved_base ) except Exception as exc: - return _reject(f"apply: no signed history: {exc}", secrets) + return _reject(f"apply: no local integrity record: {exc}", secrets) if not authorized: - return _reject("apply: no signed history", secrets) + return _reject("apply: no local integrity record", secrets) merge = _git(repo, "merge", "--ff-only", "--quiet", commit, check=False) if merge.returncode != 0: @@ -2736,7 +2698,11 @@ def cmd_apply(repo: Path, receipt_path: Path) -> int: def cmd_admit(repo: Path) -> int: - """Accept staged host-managed paths, or paths exactly covered by one transport.""" + """Accept staged host-managed paths, or paths exactly covered by one transport. + + A pass is local integrity evidence (SELF), not GATE; invalid or missing + local records still reject. + """ repo = repo.resolve() secrets = _secret_set() try: @@ -2747,13 +2713,13 @@ def cmd_admit(repo: Path) -> int: return EXIT_SUCCESS head = _resolve_commit(repo, "HEAD") if head is None: - return _reject("admit: no provenance", secrets) + return _reject("admit: no local integrity record", secrets) history_text = "" history_path = _history_file(repo) if history_path is not None and history_path.is_file(): history_text = history_path.read_text(encoding="utf-8") secrets = _secret_set(history_text) - key = _read_attestation_key(repo) + key = _read_local_integrity_key(repo) for record in _load_history_records(repo): if _transport_covers( repo, record, head=head, required=required, key=key @@ -2761,9 +2727,9 @@ def cmd_admit(repo: Path) -> int: print("admit: ok") return EXIT_SUCCESS except Exception as exc: - return _reject(f"admit: no provenance: {exc}", secrets) + return _reject(f"admit: no local integrity record: {exc}", secrets) listed = ", ".join(sorted(required)) - return _reject(f"admit: no provenance for {listed}", secrets) + return _reject(f"admit: no local integrity record for {listed}", secrets) if __name__ == "__main__": diff --git a/scripts/agent_exec_adapters.py b/scripts/agent_exec_adapters.py index 12fdbda0..dee9f352 100644 --- a/scripts/agent_exec_adapters.py +++ b/scripts/agent_exec_adapters.py @@ -5,11 +5,14 @@ does not contact providers or launch real agent runs. """ from __future__ import annotations +import json import re import shutil import subprocess +from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path +from typing import Any from scripts.agent_exec_config import ( ALLOWED_PLACEHOLDERS, @@ -23,6 +26,34 @@ _PROBE_TIMEOUT_SECONDS = 5 MAX_UNIT_PACKET_BYTES = 262_144 UNIT_PACKET_DISPLAY_MARKER = "" +TERMINAL_STATUSES = ( + "completed", + "failed", + "blocked", + "cancelled", + "failed-unknown", + "scope-expansion", +) +TERMINAL_RECORD_SCHEMA: dict[str, Any] = { + "type": "object", + "required": [ + "terminal_status", + "summary", + "changed_files", + "evidence", + "scope_expansion", + ], + "properties": { + "terminal_status": {"enum": list(TERMINAL_STATUSES)}, + "summary": {"type": "string"}, + "changed_files": {"type": "array", "items": {"type": "string"}}, + "evidence": {"type": "array", "items": {"type": "string"}}, + "scope_expansion": {"type": ["string", "null"]}, + }, +} +# Bounds the backwards brace scan over one final message. +_MAX_TERMINAL_TEXT_CHARS = 262_144 + _PLACEHOLDER_MAP = { "${workspace}": "workspace", "${unit_packet}": "unit_packet", @@ -378,17 +409,177 @@ def _read_unit_packet_text(path: Path) -> str: def _build_bounded_unit_prompt(packet_text: str) -> str: + schema = json.dumps(TERMINAL_RECORD_SCHEMA, separators=(",", ":")) return ( "Implement exactly the supplied bounded implementation unit. " "Work only inside the current workspace. " "Leave all changes uncommitted. " "Do not expand scope beyond the unit packet. " "Return a structured final result matching the unit's required schema.\n\n" + "--- TERMINAL RECORD ---\n\n" + "End your final message with exactly one terminal record: a single-line " + "JSON object that validates against this JSON Schema:\n" + f"{schema}\n" + "Use terminal_status completed only when the unit is fully implemented. " + "Use blocked, cancelled, failed-unknown, failed, or scope-expansion " + "otherwise. A missing or invalid terminal record fails the attempt.\n\n" "--- BOUNDED IMPLEMENTATION UNIT PACKET ---\n\n" f"{packet_text}" ) +def extract_terminal_record(text: str | None) -> dict[str, Any] | None: + """Return the last valid terminal record embedded in one final message.""" + if not isinstance(text, str) or not text: + return None + text = text[-_MAX_TERMINAL_TEXT_CHARS:] + decoder = json.JSONDecoder() + pos = text.rfind("{") + while pos != -1: + try: + candidate, _ = decoder.raw_decode(text, pos) + except json.JSONDecodeError: + candidate = None + record = _valid_terminal_record(candidate) + if record is not None: + return record + pos = text.rfind("{", 0, pos) + return None + + +def _valid_terminal_record(obj: object) -> dict[str, Any] | None: + if not isinstance(obj, dict): + return None + status = obj.get("terminal_status") + if not isinstance(status, str): + return None + status = status.strip().replace("_", "-") + if status not in TERMINAL_STATUSES: + return None + if not isinstance(obj.get("summary"), str): + return None + for key in ("changed_files", "evidence"): + items = obj.get(key) + if not isinstance(items, list) or any(not isinstance(i, str) for i in items): + return None + if "scope_expansion" not in obj or not ( + obj["scope_expansion"] is None or isinstance(obj["scope_expansion"], str) + ): + return None + record = dict(obj) + record["terminal_status"] = status + return record + + +def decode_terminal_record( + adapter: str, + *, + event_lines: Sequence[str | None], + stdout_tail: str, + result_text: str | None, +) -> dict[str, Any] | None: + """Decode the worker terminal record from the adapter's real output. + + ``event_lines`` are recent complete stdout lines that may carry a terminal + envelope, oldest first; ``None`` marks an envelope line too large to keep. + Event-stream adapters are read only from their final assistant/result + envelope, so prompt echoes and tool output never count as a record. + """ + if adapter in ("cursor", "claude"): + return extract_terminal_record(_cursor_final_text(event_lines)) + if adapter == "antigravity": + return extract_terminal_record(_agy_final_text(event_lines, stdout_tail)) + if adapter == "pi": + return extract_terminal_record(_pi_final_text(event_lines)) + record = extract_terminal_record(result_text) + if record is None: + record = extract_terminal_record(stdout_tail) + return record + + +def _event_objects(event_lines: Sequence[str | None]) -> list[dict[str, Any] | None]: + """Newest-first parsed events; ``None`` keeps the oversized-envelope marker.""" + out: list[dict[str, Any] | None] = [] + for line in reversed(event_lines): + if line is None: + out.append(None) + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + out.append(obj) + return out + + +def _cursor_final_text(event_lines: Sequence[str | None]) -> str | None: + for obj in _event_objects(event_lines): + if obj is None: + return None + if obj.get("type") != "result": + continue + if obj.get("is_error") is not False or obj.get("subtype") != "success": + return None + text = obj.get("result") + return text if isinstance(text, str) else None + return None + + +def _agy_final_text( + event_lines: Sequence[str | None], stdout_tail: str +) -> str | None: + candidates = _event_objects(event_lines) + try: + whole = json.loads(stdout_tail.strip()) + except json.JSONDecodeError: + whole = None + if isinstance(whole, dict): + candidates.append(whole) + for obj in candidates: + if obj is None: + return None + if obj.get("event") == "result": + payload = obj.get("result") + elif "event" not in obj and "response" in obj: + payload = obj + else: + continue + if not isinstance(payload, dict): + return None + if payload.get("is_error") or payload.get("error"): + return None + text = payload.get("response") + return text if isinstance(text, str) else None + return None + + +def _pi_final_text(event_lines: Sequence[str | None]) -> str | None: + for obj in _event_objects(event_lines): + if obj is None: + return None + if obj.get("type") != "message_end": + continue + message = obj.get("message") + if not isinstance(message, dict) or message.get("role") != "assistant": + continue + if message.get("stopReason") in ("error", "aborted"): + return None + content = message.get("content") + if isinstance(content, str): + return content + if not isinstance(content, list): + return None + return "".join( + block["text"] + for block in content + if isinstance(block, dict) + and block.get("type") == "text" + and isinstance(block.get("text"), str) + ) + return None + + def _expand_argv( items: tuple[str, ...], values: dict[str, str], @@ -480,11 +671,15 @@ def _sanitize_display( __all__ = [ "MAX_UNIT_PACKET_BYTES", + "TERMINAL_RECORD_SCHEMA", + "TERMINAL_STATUSES", "UNIT_PACKET_DISPLAY_MARKER", "AdapterProbe", "AdapterSpec", "BUILTIN_SPECS", "Invocation", "build_invocation", + "decode_terminal_record", + "extract_terminal_record", "probe_adapter", ] diff --git a/scripts/hooks/pre-commit b/scripts/hooks/pre-commit index 08778e53..c9a0871f 100755 --- a/scripts/hooks/pre-commit +++ b/scripts/hooks/pre-commit @@ -1,5 +1,5 @@ #!/usr/bin/env bash -# 本地硬闸（pre-commit）：集成分支直提拒绝 + 分支命名。权威准入在 PR（baseline.md:26）。 +# 本地检查（pre-commit）：集成分支直提拒绝 + 分支命名 + 本地完整性核对。权威准入在 PR（baseline.md:26）。 branch=$(git rev-parse --abbrev-ref HEAD) case "$branch" in main|master) @@ -11,4 +11,6 @@ echo "$branch" | grep -qE '^(feature|fix|hotfix|release)/[A-Za-z0-9._-]+$' || { exit 1 } root=$(git rev-parse --show-toplevel) +# admit 只是本地完整性/篡改证据（SELF）：同一 OS 用户可读取本地 HMAC 密钥并绕过本钩子， +# 它不是 GATE，也不是可信身份。缺失或无效的本地记录仍拒绝；交付信任来自精确 HEAD 的 CI 或独立 REPLAY。 PYTHONDONTWRITEBYTECODE=1 python3 "$root/scripts/agent_exec.py" admit --repo "$root" diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py index 009fced1..029375d9 100644 --- a/tests/test_agent_exec.py +++ b/tests/test_agent_exec.py @@ -212,6 +212,84 @@ def _antigravity_help() -> str: return "\n".join(_REQUIRED_HELP_MARKERS["antigravity"]) + "\n" +def _terminal_json(status: str = "completed") -> str: + return json.dumps( + { + "terminal_status": status, + "summary": "done", + "changed_files": ["src/marker.txt"], + "evidence": ["wrote marker"], + "scope_expansion": None, + }, + separators=(",", ":"), + ) + + +def _agy_stream_events(final_text: str, *, is_error: bool = False) -> list[str]: + """Representative Antigravity ``--output-format stream-json`` lines.""" + conv = "b97cb31f-a6bd-40bf-aca8-aaeaa2e1999a" + result: dict[str, object] = { + "conversation_id": conv, + "response": final_text, + "num_turns": 2, + "duration_ms": 10, + } + if is_error: + result["is_error"] = True + result["error"] = "model error" + events: list[dict[str, object]] = [ + {"event": "init", "conversation_id": conv, "init": {"cwd": "/w", "tools": ["run_command"]}}, + { + "event": "step_update", + "step_update": {"conversation_id": conv, "step_index": 0, "state": "DONE", "step_type": "user_input"}, + }, + { + "event": "step_update", + "step_update": {"conversation_id": conv, "step_index": 1, "state": "DONE", "step_type": "agent_response"}, + }, + {"event": "result", "conversation_id": conv, "result": result}, + ] + return [json.dumps(event, separators=(",", ":")) for event in events] + + +def _pi_json_events( + final_text: str, + *, + prompt_text: str = "unit", + stop_reason: str = "stop", + agent_end_padding: int = 0, +) -> list[str]: + """Representative PI ``--mode json`` lines (see pi docs/json.md).""" + user = {"role": "user", "content": prompt_text, "timestamp": 1} + assistant = { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "plan"}, + {"type": "text", "text": final_text}, + ], + "stopReason": stop_reason, + } + padding = {"role": "toolResult", "content": [{"type": "text", "text": "x" * agent_end_padding}]} + events: list[dict[str, object]] = [ + {"type": "session", "version": 3, "id": "s", "timestamp": "2026-10-15T00:00:00.000Z", "cwd": "/w"}, + {"type": "agent_start"}, + {"type": "turn_start"}, + {"type": "message_start", "message": user}, + {"type": "message_end", "message": user}, + {"type": "message_start", "message": {"role": "assistant", "content": [], "stopReason": "pending"}}, + { + "type": "message_update", + "usage": {"input": 1, "output": 1}, + "assistantMessageEvent": {"type": "text_delta", "contentIndex": 1, "delta": final_text}, + }, + {"type": "message_end", "message": assistant}, + {"type": "turn_end", "message": assistant, "toolResults": []}, + {"type": "agent_end", "messages": [user, padding, assistant], "willRetry": False}, + {"type": "agent_settled"}, + ] + return [json.dumps(event, separators=(",", ":")) for event in events] + + class AgentExecControllerTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-ctrl-") @@ -308,19 +386,6 @@ class AgentExecControllerTests(unittest.TestCase): def test_run_uses_fresh_process_and_worktree_per_attempt(self) -> None: meta = self.root / "meta" meta.mkdir() - _write_fake_worker( - self.bindir, - "worker-a", - body=textwrap.dedent( - f"""\ - workspace="$1" - echo "$$" > "{meta}/a.pid" - echo "$workspace" > "{meta}/a.wt" - # fail clean after no writes - exit 7 - """ - ), - ) _write_fake_worker( self.bindir, "worker-b", @@ -328,8 +393,8 @@ class AgentExecControllerTests(unittest.TestCase): f"""\ workspace="$1" result="$3" - echo "$$" > "{meta}/b.pid" - echo "$workspace" > "{meta}/b.wt" + echo "$$" >> "{meta}/pids" + echo "$workspace" >> "{meta}/wts" mkdir -p "$workspace/src" printf 'ok\\n' > "$workspace/src/marker.txt" printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":["wrote marker"],"scope_expansion":null}}' > "$result" @@ -338,38 +403,36 @@ class AgentExecControllerTests(unittest.TestCase): ), ) self._configure( - profiles={ - "a": {"adapter": "custom", "executable": "worker-a"}, - "b": {"adapter": "custom", "executable": "worker-b"}, - }, - role_order=["a", "b"], + profiles={"b": {"adapter": "custom", "executable": "worker-b"}}, + role_order=["b"], ) - self.base_snap = _canonical_snapshot(self.repo) unit = _write_unit(self.root / "unit.json") - receipt = self.receipts / "fresh.json" - proc = self._run_cli( - "run", - "--repo", - str(self.repo), - "--role", - "implementation", - "--unit", - str(unit), - "--receipt", - str(receipt), - ) - self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) - data = json.loads(receipt.read_text(encoding="utf-8")) - self.assertEqual(data["schema"], "agent-exec-receipt/v1") - self.assertEqual(data["terminal_status"], "success") - a_pid = (meta / "a.pid").read_text(encoding="utf-8").strip() - b_pid = (meta / "b.pid").read_text(encoding="utf-8").strip() - a_wt = (meta / "a.wt").read_text(encoding="utf-8").strip() - b_wt = (meta / "b.wt").read_text(encoding="utf-8").strip() - self.assertNotEqual(a_pid, b_pid) - self.assertNotEqual(a_wt, b_wt) - self.assertTrue(a_wt.startswith(f"/tmp/agent-exec-{os.getuid()}/")) - self.assertTrue(b_wt.startswith(f"/tmp/agent-exec-{os.getuid()}/")) + for label in ("first", "second"): + receipt = self.receipts / f"fresh-{label}.json" + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + self.assertEqual( + proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout + ) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["schema"], "agent-exec-receipt/v1") + self.assertEqual(data["terminal_status"], "success") + pids = (meta / "pids").read_text(encoding="utf-8").split() + wts = (meta / "wts").read_text(encoding="utf-8").split() + self.assertEqual(len(pids), 2) + self.assertNotEqual(pids[0], pids[1]) + self.assertNotEqual(wts[0], wts[1]) + for wt in wts: + self.assertTrue(wt.startswith(f"/tmp/agent-exec-{os.getuid()}/")) self._assert_canonical_untouched() def test_missing_profiles_fall_through_in_declared_order(self) -> None: @@ -499,7 +562,10 @@ class AgentExecControllerTests(unittest.TestCase): self.assertFalse((self.repo / "src" / "marker.txt").exists()) self._assert_canonical_untouched() - def test_failed_clean_can_continue_after_exact_restore(self) -> None: + def test_declared_offline_launched_failure_is_failed_unknown_and_stops(self) -> None: + """network=false/external_cost=false are declarations, not isolation proof.""" + launched = self.root / "declared-offline-next.log" + launched.write_text("", encoding="utf-8") _write_fake_worker( self.bindir, "dirty", @@ -516,12 +582,8 @@ class AgentExecControllerTests(unittest.TestCase): self.bindir, "ok", body=textwrap.dedent( - """\ - workspace="$1" - result="$3" - mkdir -p "$workspace/src" - printf 'final\\n' > "$workspace/src/marker.txt" - printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result" + f"""\ + echo next >> "{launched}" exit 0 """ ), @@ -534,7 +596,10 @@ class AgentExecControllerTests(unittest.TestCase): role_order=["dirty", "ok"], ) unit = _write_unit(self.root / "unit.json") - receipt = self.receipts / "clean.json" + packet = json.loads(unit.read_text(encoding="utf-8")) + self.assertFalse(packet["network"]) + self.assertFalse(packet["external_cost"]) + receipt = self.receipts / "declared-offline.json" proc = self._run_cli( "run", "--repo", @@ -546,29 +611,26 @@ class AgentExecControllerTests(unittest.TestCase): "--receipt", str(receipt), ) - self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) + self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout) data = json.loads(receipt.read_text(encoding="utf-8")) - self.assertEqual(data["attempts"][0]["terminal_status"], "failed-clean") - self.assertTrue(data["attempts"][0]["cleanup"]["exact_restore"]) - self.assertEqual(data["terminal_status"], "success") + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(len(data["attempts"]), 1) + attempt = data["attempts"][0] + self.assertEqual(attempt["terminal_status"], "failed-unknown") + self.assertFalse(attempt["cleanup"]["exact_restore"]) + self.assertTrue(attempt["workspace_preserved"]) + self.assertFalse(attempt["cleanup"]["declared_network"]) + self.assertFalse(attempt["cleanup"]["declared_external_cost"]) + self.assertTrue( + (Path(attempt["worktree"]) / "src" / "marker.txt").is_file() + ) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self.assertEqual(data.get("host_fallback_authorizations", 0), 0) self._assert_canonical_untouched() - def test_failed_clean_retries_profile_before_next_success(self) -> None: + def test_safe_unavailable_retries_profile_before_next_success(self) -> None: launches = self.root / "profile-launches.log" launches.write_text("", encoding="utf-8") - _write_fake_worker( - self.bindir, - "retryme", - body=textwrap.dedent( - f"""\ - printf 'retryme\\n' >> "{launches}" - workspace="$1" - mkdir -p "$workspace/src" - printf 'tmp\\n' > "$workspace/src/marker.txt" - exit 9 - """ - ), - ) _write_fake_worker( self.bindir, "thenok", @@ -586,7 +648,7 @@ class AgentExecControllerTests(unittest.TestCase): ) self._configure( profiles={ - "retryme": {"adapter": "custom", "executable": "retryme"}, + "retryme": {"adapter": "custom", "executable": "retryme-missing"}, "thenok": {"adapter": "custom", "executable": "thenok"}, }, role_order=["retryme", "thenok"], @@ -611,7 +673,7 @@ class AgentExecControllerTests(unittest.TestCase): observed = [ line for line in launches.read_text(encoding="utf-8").splitlines() if line ] - self.assertEqual(observed, ["retryme", "retryme", "retryme", "thenok"]) + self.assertEqual(observed, ["thenok"]) data = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual( [ @@ -623,9 +685,9 @@ class AgentExecControllerTests(unittest.TestCase): for attempt in data["attempts"] ], [ - ("retryme", 1, "failed-clean"), - ("retryme", 2, "failed-clean"), - ("retryme", 3, "failed-clean"), + ("retryme", 1, "unavailable"), + ("retryme", 2, "unavailable"), + ("retryme", 3, "unavailable"), ("thenok", 1, "success"), ], ) @@ -634,13 +696,15 @@ class AgentExecControllerTests(unittest.TestCase): ) self._assert_canonical_untouched() - def test_emergency_profile_runs_once_after_regular_failed_clean(self) -> None: + def test_emergency_profile_runs_once_after_regular_safe_exhaustion(self) -> None: launches = self.root / "emergency-launches.log" launches.write_text("", encoding="utf-8") + # Installed but missing required capability flags: each attempt is a + # safe pre-launch `unavailable`, which is what may route onward. normal = ( - ("cursor", "cursor-regular", _cursor_help()), - ("antigravity", "agy-regular", _antigravity_help()), - ("pi", "pi-regular", _pi_help()), + ("cursor", "cursor-regular", "incompatible\n"), + ("antigravity", "agy-regular", "incompatible\n"), + ("pi", "pi-regular", "incompatible\n"), ) for adapter, executable, help_text in normal: _write_fake_worker( @@ -650,9 +714,6 @@ class AgentExecControllerTests(unittest.TestCase): body=textwrap.dedent( f"""\ printf '{adapter}\\n' >> "{launches}" - mkdir -p src - printf 'tmp\\n' > src/marker.txt - printf '%s\\n' '{{"terminal_status":"failed","summary":"{adapter} failed clean","changed_files":["src/marker.txt"]}}' exit 9 """ ), @@ -736,7 +797,7 @@ class AgentExecControllerTests(unittest.TestCase): self.assertEqual( recorded, [ - (adapter, ordinal, "failed-clean", False) + (adapter, ordinal, "unavailable", False) for adapter in order for ordinal in (1, 2, 3) ] @@ -748,10 +809,7 @@ class AgentExecControllerTests(unittest.TestCase): observed = [ line for line in launches.read_text(encoding="utf-8").splitlines() if line ] - self.assertEqual( - observed, - ["cursor"] * 3 + ["antigravity"] * 3 + ["pi"] * 3 + ["codex"], - ) + self.assertEqual(observed, ["codex"]) self.assertEqual(data["attempts"][-1]["summary"], "done") self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) self._assert_canonical_untouched() @@ -2068,8 +2126,8 @@ class AgentExecControllerTests(unittest.TestCase): self.assertEqual(data.get("host_fallback_authorizations", 0), 0) self._assert_canonical_untouched() - def test_unproven_restore_is_failed_unknown_and_stops(self) -> None: - """failed-clean is allowed only after _restore_worktree succeeds.""" + def test_launched_failure_without_diff_is_failed_unknown_and_stops(self) -> None: + """No diff and declared-offline packets still do not prove external cleanliness.""" cases = { "clean-exit": "exit 0\n", "nonzero-exit": "exit 7\n", @@ -2104,15 +2162,12 @@ class AgentExecControllerTests(unittest.TestCase): ) unit = _write_unit(self.root / f"unit-restore-{label}.json") receipt = self.receipts / f"restore-{label}.json" - with mock.patch.object( - agent_exec, "_restore_worktree", return_value=False - ): - code = agent_exec.cmd_run( - self.repo, - role="implementation", - unit_path=unit, - receipt_path=receipt, - ) + code = agent_exec.cmd_run( + self.repo, + role="implementation", + unit_path=unit, + receipt_path=receipt, + ) self.assertEqual(code, agent_exec.EXIT_BLOCKED) data = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual(data["terminal_status"], "failed-unknown") @@ -2251,6 +2306,218 @@ class AgentExecControllerTests(unittest.TestCase): self.assertEqual(data.get("host_fallback_authorizations", 0), 0) self._assert_canonical_untouched() + def test_builtin_event_envelopes_gate_success(self) -> None: + """Real Cursor/agy/PI output envelopes decide the terminal; prose fails closed.""" + prose = "All done, the marker is written." + cases = ( + ("cursor", _cursor_help(), _cursor_stream_events("ok\n" + _terminal_json()), "success"), + ("cursor", _cursor_help(), _cursor_stream_events(prose), "failed-unknown"), + ("cursor", _cursor_help(), _cursor_stream_events(_terminal_json("blocked")), "blocked"), + ( + "cursor", + _cursor_help(), + _cursor_stream_events(_terminal_json(), is_error=True), + "failed-unknown", + ), + ("antigravity", _antigravity_help(), _agy_stream_events("ok " + _terminal_json()), "success"), + ("antigravity", _antigravity_help(), _agy_stream_events(prose), "failed-unknown"), + ( + "pi", + _pi_help(), + _pi_json_events( + "ok\n```json\n" + _terminal_json() + "\n```", + agent_end_padding=2 * agent_exec.TERMINAL_LINE_MAX_BYTES, + ), + "success", + ), + ("pi", _pi_help(), _pi_json_events(prose, prompt_text=_terminal_json()), "failed-unknown"), + ("pi", _pi_help(), [], "failed-unknown"), + ) + expected_exit = { + "success": agent_exec.EXIT_SUCCESS, + "failed-unknown": agent_exec.EXIT_BLOCKED, + "blocked": agent_exec.EXIT_BLOCKED, + } + for index, (adapter, help_text, events, expected) in enumerate(cases): + label = f"{adapter}-{index}-{expected}" + with self.subTest(case=label): + events_path = self.root / f"events-{label}.jsonl" + events_path.write_text( + "".join(line + "\n" for line in events), encoding="utf-8" + ) + executable = f"envelope-{label}" + _write_fake_worker( + self.bindir, + executable, + help_text=help_text, + body=textwrap.dedent( + f"""\ + mkdir -p src + printf 'ok\\n' > src/marker.txt + cat "{events_path}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + adapter: {"adapter": adapter, "executable": executable} + }, + role_order=[adapter], + ) + unit = _write_unit(self.root / f"unit-{label}.json") + receipt = self.receipts / f"envelope-{label}.json" + code = agent_exec.cmd_run( + self.repo, + role="implementation", + unit_path=unit, + receipt_path=receipt, + ) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], expected, data.get("reason")) + self.assertEqual(code, expected_exit[expected]) + self.assertEqual(len(data["attempts"]), 1) + attempt = data["attempts"][0] + self.assertEqual(attempt["exit_code"], 0) + self.assertEqual(attempt["changed_paths"], ["src/marker.txt"]) + if expected == "success": + self.assertTrue(attempt["transport_ref"]) + else: + self.assertIsNone(attempt.get("transport_ref")) + self.assertTrue(attempt["workspace_preserved"]) + if expected == "failed-unknown": + self.assertIn("no valid completed terminal record", attempt["reason"]) + self._assert_canonical_untouched() + + +class TerminalRecordDecodingTests(unittest.TestCase): + """Adapter-specific terminal decoding over representative real envelopes.""" + + def _decode( + self, + adapter: str, + lines: list[str | None], + *, + stdout_tail: str | None = None, + result_text: str | None = None, + ) -> dict[str, object] | None: + from scripts.agent_exec_adapters import decode_terminal_record + + tail = stdout_tail + if tail is None: + tail = "".join(line + "\n" for line in lines if line is not None) + return decode_terminal_record( + adapter, event_lines=lines, stdout_tail=tail, result_text=result_text + ) + + def test_cursor_stream_json_result_is_decoded(self) -> None: + lines = _cursor_stream_events("Done.\n" + _terminal_json() + "\nThanks.") + record = self._decode("cursor", lines) + assert record is not None + self.assertEqual(record["terminal_status"], "completed") + self.assertEqual(record["changed_files"], ["src/marker.txt"]) + + def test_cursor_user_echo_and_error_result_are_not_records(self) -> None: + lines = _cursor_stream_events("prose only") + echo = json.loads(lines[1]) + echo["message"]["content"][0]["text"] = _terminal_json() + lines[1] = json.dumps(echo) + self.assertIsNone(self._decode("cursor", lines)) + self.assertIsNone( + self._decode("cursor", _cursor_stream_events(_terminal_json(), is_error=True)) + ) + self.assertIsNone(self._decode("cursor", _cursor_stream_events(_terminal_json())[:-1])) + + def test_cursor_blocked_record_is_decoded_as_fail_stop(self) -> None: + record = self._decode("cursor", _cursor_stream_events(_terminal_json("blocked"))) + assert record is not None + self.assertEqual(record["terminal_status"], "blocked") + + def test_agy_stream_json_and_json_results_are_decoded(self) -> None: + record = self._decode("antigravity", _agy_stream_events("ok " + _terminal_json())) + assert record is not None + self.assertEqual(record["terminal_status"], "completed") + single = json.dumps( + {"conversation_id": "c", "response": "ok\n" + _terminal_json(), "num_turns": 1}, + indent=2, + ) + record = self._decode("antigravity", [], stdout_tail=single) + assert record is not None + self.assertEqual(record["terminal_status"], "completed") + self.assertIsNone( + self._decode("antigravity", _agy_stream_events(_terminal_json(), is_error=True)) + ) + self.assertIsNone(self._decode("antigravity", _agy_stream_events(_terminal_json())[:-1])) + + def test_pi_json_final_assistant_message_is_decoded(self) -> None: + lines = _pi_json_events("ok\n" + _terminal_json()) + record = self._decode("pi", lines) + assert record is not None + self.assertEqual(record["terminal_status"], "completed") + self.assertIsNone( + self._decode("pi", _pi_json_events("prose", prompt_text=_terminal_json())) + ) + self.assertIsNone( + self._decode("pi", _pi_json_events(_terminal_json(), stop_reason="error")) + ) + + def test_prose_missing_or_invalid_terminal_fails_closed(self) -> None: + for adapter, lines in ( + ("cursor", _cursor_stream_events("finished, no record")), + ("antigravity", _agy_stream_events("finished, no record")), + ("pi", _pi_json_events("finished, no record")), + ("cursor", []), + ("antigravity", []), + ("pi", []), + ): + with self.subTest(adapter=adapter, lines=len(lines)): + self.assertIsNone(self._decode(adapter, lines)) + invalid = json.dumps({"terminal_status": "done", "summary": "x"}) + self.assertIsNone(self._decode("cursor", _cursor_stream_events(invalid))) + missing_keys = json.dumps({"terminal_status": "completed", "summary": "x"}) + self.assertIsNone(self._decode("cursor", _cursor_stream_events(missing_keys))) + + def test_raw_stdout_record_is_not_an_envelope_for_builtins(self) -> None: + bare = [_terminal_json()] + for adapter in ("cursor", "antigravity", "pi"): + with self.subTest(adapter=adapter): + self.assertIsNone(self._decode(adapter, bare)) + record = self._decode("custom", [], stdout_tail="", result_text=_terminal_json()) + assert record is not None + self.assertEqual(record["terminal_status"], "completed") + + def test_prompt_states_schema_without_being_a_record(self) -> None: + from scripts import agent_exec_adapters as adapters + + prompt = adapters._build_bounded_unit_prompt('{"unit_id":"u"}') + self.assertIn( + json.dumps(adapters.TERMINAL_RECORD_SCHEMA, separators=(",", ":")), prompt + ) + for status in adapters.TERMINAL_STATUSES: + self.assertIn(status, prompt) + self.assertIsNone(adapters.extract_terminal_record(prompt)) + + def test_stdout_tail_keeps_final_envelope_behind_oversized_event(self) -> None: + big = 2 * agent_exec.TERMINAL_LINE_MAX_BYTES + stream = "".join( + line + "\n" + for line in _pi_json_events("ok " + _terminal_json(), agent_end_padding=big) + ).encode("utf-8") + tail = agent_exec._StdoutTail(agent_exec.OUTPUT_TAIL_BYTES) + for start in range(0, len(stream), 4093): + tail.extend(stream[start : start + 4093]) + self.assertNotIn(b"message_end", tail.text().encode("utf-8")) + record = self._decode("pi", tail.terminal_lines(), stdout_tail=tail.text()) + assert record is not None + self.assertEqual(record["terminal_status"], "completed") + + oversized = _cursor_stream_events("x" * big + _terminal_json()) + tail = agent_exec._StdoutTail(agent_exec.OUTPUT_TAIL_BYTES) + tail.extend("".join(line + "\n" for line in oversized).encode("utf-8")) + lines = tail.terminal_lines() + self.assertIsNone(lines[-1]) + self.assertIsNone(self._decode("cursor", lines, stdout_tail=tail.text())) + class AgentExecHistoryTests(unittest.TestCase): def setUp(self) -> None: @@ -2493,7 +2760,8 @@ class AgentExecHistoryTests(unittest.TestCase): self.assertEqual(identity["status"], "recorded") self.assertEqual(identity["run_id"], receipt["run_id"]) self.assertEqual(identity["record_id"], record["record_id"]) - self.assertEqual(identity["attestation"], record["attestation"]) + self.assertEqual(identity["local_integrity_mac"], record["attestation"]) + self.assertNotIn("attestation", identity) self.assertNotIn(key.hex(), json.dumps(identity)) def test_attestation_key_symlink_is_not_followed(self) -> None: @@ -2517,7 +2785,7 @@ class AgentExecHistoryTests(unittest.TestCase): def load() -> None: try: barrier.wait(timeout=5) - found.append(agent_exec._ensure_attestation_key(self.repo)) + found.append(agent_exec._ensure_local_integrity_key(self.repo)) except BaseException as exc: # noqa: BLE001 - race failures are the assertion errors.append(exc) @@ -2563,8 +2831,9 @@ _PROBE_NORMAL_HOLDER = "piholder" U5_BRANCH = "feature/u5-real-probe" U5_MARKER_REL = "src/u5-marker.txt" U5_MARKER_BODY = "U5-REAL-PROBE-MARKER\n" +# Envelope adapters JSON-escape the final text, so quotes may be backslashed. _MODEL_RECEIPT_RE = re.compile( - r'"model_receipt_status"\s*:\s*"(verified|not-observable|mismatch|unavailable)"' + r'\\?"model_receipt_status\\?"\s*:\s*\\?"(verified|not-observable|mismatch|unavailable)\\?"' ) @@ -2671,18 +2940,46 @@ def _marker_run_body(*, body: str, model_receipt: str | None) -> str: } if model_receipt is not None: payload["model_receipt_status"] = model_receipt - encoded = json.dumps(payload, separators=(",", ":")) + final_text = "Wrote the marker.\n" + json.dumps(payload, separators=(",", ":")) + events = _cursor_stream_events(final_text) writer = ( "from pathlib import Path; " f"p=Path({json.dumps(U5_MARKER_REL)}); " "p.parent.mkdir(parents=True); " f"p.write_text({json.dumps(body)}, encoding='utf-8')" ) - return ( - f"python3 -c {shlex.quote(writer)}\n" - f"printf '%s\\n' {shlex.quote(encoded)}\n" - "exit 0\n" - ) + printed = "".join(f"printf '%s\\n' {shlex.quote(line)}\n" for line in events) + return f"python3 -c {shlex.quote(writer)}\n" + printed + "exit 0\n" + + +def _cursor_stream_events(final_text: str, *, is_error: bool = False) -> list[str]: + """Representative Cursor ``--output-format stream-json`` lines.""" + session = "00000000-0000-4000-8000-000000000000" + events: list[dict[str, object]] = [ + {"type": "system", "subtype": "init", "session_id": session}, + { + "type": "user", + "message": {"role": "user", "content": [{"type": "text", "text": "unit"}]}, + "session_id": session, + }, + { + "type": "assistant", + "message": { + "role": "assistant", + "content": [{"type": "text", "text": final_text}], + }, + "session_id": session, + }, + { + "type": "result", + "subtype": "error" if is_error else "success", + "duration_ms": 10, + "is_error": is_error, + "result": final_text, + "session_id": session, + }, + ] + return [json.dumps(event, separators=(",", ":")) for event in events] def _write_probe_unit(path: Path, timeout_seconds: int) -> None: diff --git a/tests/test_agent_exec_admission.py b/tests/test_agent_exec_admission.py index c97bb854..1d581aa6 100644 --- a/tests/test_agent_exec_admission.py +++ b/tests/test_agent_exec_admission.py @@ -409,7 +409,7 @@ class AgentExecAdmissionTests(unittest.TestCase): ) combined = proc.stdout + proc.stderr self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined) - self.assertIn("apply: no signed history", proc.stderr) + self.assertIn("apply: no local integrity record", proc.stderr) self.assertNotIn("apply: fast-forwarded", proc.stdout) self.assertEqual(self._state(), before, label) @@ -471,7 +471,7 @@ class AgentExecAdmissionTests(unittest.TestCase): 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 provenance", proc.stderr) + self.assertIn("admit: no local integrity record", proc.stderr) self.assertEqual(self._state(), before) def test_admit_accepts_exact_transport_paths(self) -> None: @@ -496,7 +496,7 @@ class AgentExecAdmissionTests(unittest.TestCase): 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 provenance", 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"}) @@ -515,7 +515,7 @@ class AgentExecAdmissionTests(unittest.TestCase): 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 provenance", proc.stderr) + self.assertIn("admit: no local integrity record", proc.stderr) self.assertNotIn("admit: ok", proc.stdout) self.assertEqual(self._state(), before) @@ -556,7 +556,7 @@ class AgentExecAdmissionTests(unittest.TestCase): 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 provenance", moved_proc.stderr) + 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) @@ -568,7 +568,7 @@ class AgentExecAdmissionTests(unittest.TestCase): 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 provenance", proc.stderr) + 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: @@ -610,7 +610,7 @@ class AgentExecAdmissionTests(unittest.TestCase): 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 provenance", rejected.stderr) + self.assertIn("admit: no local integrity record", rejected.stderr) def test_precommit_rejects_unproven_paths_after_branch_checks(self) -> None: self._install_project_hook() @@ -618,7 +618,7 @@ class AgentExecAdmissionTests(unittest.TestCase): proc = self._hook() combined = proc.stdout + proc.stderr self.assertNotEqual(proc.returncode, 0, combined) - self.assertIn("admit: no provenance", combined) + self.assertIn("admit: no local integrity record", combined) _git(self.repo, "checkout", "-B", "main", self.head) self._stage("src/extra.txt", "unproven\n")