From 8f11d002aec74ed21d3dae8ccbb760963b99e93a Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 15 Oct 2026 04:46:25 +0000 Subject: [PATCH] Add recoverable execution agent controller Why: provide isolated, recoverable per-unit CLI execution and safe host fallback Evidence: 43 focused tests, configuration validation, governance regression, and diff check Impact: adds agent-exec validate/run controller, receipts, transports, and Make entrypoint Refs: U3; docs/superpowers/plans/2026-10-15-project-agent-execution.md --- Makefile | 6 +- scripts/agent_exec.py | 1751 ++++++++++++++++++++++++++++++++++++++ tests/test_agent_exec.py | 1372 +++++++++++++++++++++++++++++ 3 files changed, 3125 insertions(+), 4 deletions(-) create mode 100644 scripts/agent_exec.py create mode 100644 tests/test_agent_exec.py diff --git a/Makefile b/Makefile index d3f13a07..f1246f76 100644 --- a/Makefile +++ b/Makefile @@ -77,12 +77,10 @@ help: @echo "make dev 开发模式说明（前后端分离热更新）" @echo "make pin 推进 submodule 指针到各仓远端 main（提交前须 make test 通过）" @echo "make status submodule 状态" - @echo "make agent-exec-validate 校验 AGENT-EXECUTION.md 配置合同（U1：配置解析单测）" + @echo "make agent-exec-validate 校验 AGENT-EXECUTION.md（agent_exec validate）" -# U1 temporary target: configuration contract tests. Task 3 replaces the recipe -# with `python3 scripts/agent_exec.py validate --repo .`. agent-exec-validate: - PYTHONDONTWRITEBYTECODE=1 python3 -m unittest tests.test_agent_exec_config -v + PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo . sync: git submodule update --init --recursive diff --git a/scripts/agent_exec.py b/scripts/agent_exec.py new file mode 100644 index 00000000..83b2d24a --- /dev/null +++ b/scripts/agent_exec.py @@ -0,0 +1,1751 @@ +"""Recoverable agent-exec controller: validate, isolate, supervise, receipt, transport.""" +from __future__ import annotations + +import argparse +import fcntl +import hashlib +import json +import os +import re +import select +import shutil +import signal +import subprocess +import sys +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +OUTPUT_TAIL_BYTES = 65536 + +_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_config import ( + SECRET_VALUE_RE, + ConfigError, + ExecutionConfig, + Profile, + load_config, + validate_config, +) + +RECEIPT_SCHEMA = "agent-exec-receipt/v1" + +# Distinct stable exit codes. +EXIT_SUCCESS = 0 +EXIT_INVALID_CONFIG = 2 +EXIT_UNAVAILABLE_EXHAUSTION = 3 +EXIT_BLOCKED = 4 +EXIT_SCOPE_EXPANSION = 5 +EXIT_CANCELLED = 6 +EXIT_HOST_FALLBACK = 10 + +DEFAULT_BRANCHES = frozenset({"main", "master"}) +SECRET_PATH_MARKERS = ( + ".secrets", + "credentials", + "id_rsa", + "id_ed25519", + ".env", + "ACCESS_TOKEN", +) +ENV_ALLOW_PREFIXES = ("LC_",) +ENV_ALLOW_EXACT = frozenset( + { + "PATH", + "HOME", + "USER", + "LOGNAME", + "LANG", + "LANGUAGE", + "TMPDIR", + "TMP", + "TEMP", + "TERM", + "TZ", + "XDG_RUNTIME_DIR", + "XDG_CACHE_HOME", + "XDG_CONFIG_HOME", + "XDG_DATA_HOME", + "SSL_CERT_FILE", + "SSL_CERT_DIR", + "REQUESTS_CA_BUNDLE", + "CURL_CA_BUNDLE", + "HTTP_PROXY", + "HTTPS_PROXY", + "NO_PROXY", + "http_proxy", + "https_proxy", + "no_proxy", + "ALL_PROXY", + "all_proxy", + # Provider discovery names only — values never enter receipts. + "OPENAI_API_KEY", + "ANTHROPIC_API_KEY", + "CURSOR_API_KEY", + "AGENT_EXEC_TEST_SECRET", + } +) + +UNIT_REQUIRED = ( + "unit_id", + "goal", + "allowed_paths", + "verification", + "constraints", + "network", + "external_cost", + "timeout_seconds", +) + + +class UnitPacketError(ValueError): + """Raised when a unit packet fails contract checks.""" + + +@dataclass(frozen=True) +class UnitPacket: + unit_id: str + goal: str + allowed_paths: tuple[str, ...] + verification: tuple[Any, ...] + constraints: tuple[Any, ...] + network: bool + external_cost: bool + timeout_seconds: int + raw: dict[str, Any] + digest: str + + +@dataclass +class AttemptRecord: + attempt_id: str + profile: str + adapter: str + terminal_status: str + reason: str | None = None + worktree: str | None = None + pid: int | None = None + exit_code: int | None = None + timed_out: bool = False + signal: int | None = None + changed_paths: list[str] = field(default_factory=list) + inventory_paths: list[str] = field(default_factory=list) + raw_log: str | None = None + summary: str = "" + cleanup: dict[str, Any] = field(default_factory=dict) + workspace_preserved: bool = False + transport_ref: str | None = None + observed_version: str | None = None + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="agent_exec") + sub = parser.add_subparsers(dest="command", required=True) + + p_val = sub.add_parser("validate", help="Validate AGENT-EXECUTION.md (no provider calls)") + p_val.add_argument("--repo", required=True) + + p_run = sub.add_parser("run", help="Run one bounded unit through configured profiles") + p_run.add_argument("--repo", required=True) + p_run.add_argument("--role", required=True) + p_run.add_argument("--unit", required=True) + p_run.add_argument("--receipt", required=True) + + args = parser.parse_args(argv) + if args.command == "validate": + return cmd_validate(Path(args.repo)) + if args.command == "run": + return cmd_run( + Path(args.repo), + role=args.role, + unit_path=Path(args.unit), + receipt_path=Path(args.receipt), + ) + return EXIT_INVALID_CONFIG + + +def cmd_validate(repo: Path) -> int: + try: + cfg = load_config(repo) + validate_config(repo) + except ConfigError as exc: + print(f"invalid-config: {exc}", file=sys.stderr) + return EXIT_INVALID_CONFIG + # Local capability probes only — never contact a provider. + for name, profile in cfg.profiles.items(): + if not profile.enabled: + continue + probe_adapter(profile, cfg) + print("validate: ok") + return EXIT_SUCCESS + + +def cmd_run( + repo: Path, + *, + role: str, + unit_path: Path, + receipt_path: Path, +) -> int: + run_id = uuid.uuid4().hex + attempts: list[AttemptRecord] = [] + secret_values: set[str] = set() + _collect_secret_values(secret_values) + + try: + cfg = load_config(repo) + except ConfigError as exc: + _write_receipt( + receipt_path, + _base_receipt( + run_id=run_id, + repo=repo, + role=role, + unit_id=None, + config=None, + terminal_status="invalid-config", + reason=str(exc), + attempts=[], + secret_values=secret_values, + ), + secret_values=secret_values, + summary_bytes=256, + ) + return EXIT_INVALID_CONFIG + + try: + base = _require_canonical_ready(repo) + unit = load_unit_packet(unit_path, cfg) + profiles = _select_profiles(cfg, role) + except (UnitPacketError, ConfigError, RuntimeError, ValueError) as exc: + _write_receipt( + receipt_path, + _base_receipt( + run_id=run_id, + repo=repo, + role=role, + unit_id=None, + config=cfg, + terminal_status="invalid-config", + reason=str(exc), + attempts=[], + secret_values=secret_values, + ), + secret_values=secret_values, + summary_bytes=cfg.execution.summary_bytes, + ) + return EXIT_INVALID_CONFIG + + try: + _validate_allowed_path_symlinks(repo, unit.allowed_paths) + except UnitPacketError as exc: + # Symlink escape is a safety failure: stop routing, never authorize fallback. + _write_receipt( + receipt_path, + _base_receipt( + run_id=run_id, + repo=repo, + role=role, + unit_id=unit.unit_id, + config=cfg, + terminal_status="failed-unknown", + reason=str(exc), + attempts=[], + secret_values=secret_values, + ), + secret_values=secret_values, + summary_bytes=cfg.execution.summary_bytes, + ) + return EXIT_BLOCKED + + run_dir = _create_private_run_dir(run_id) + config_digest = _file_digest(repo / "AGENT-EXECUTION.md") + + terminal = "unavailable" + reason: str | None = None + transport: dict[str, Any] | None = None + changed_paths: list[str] = [] + host_scope: dict[str, Any] | None = None + host_fallback_authorizations = 0 + stop = False + + for profile_name in profiles: + if stop: + break + profile = cfg.profiles[profile_name] + attempt_id = uuid.uuid4().hex[:12] + attempt = _run_attempt( + repo=repo, + cfg=cfg, + profile=profile, + unit=unit, + unit_path=unit_path, + base=base, + run_id=run_id, + attempt_id=attempt_id, + run_dir=run_dir, + secret_values=secret_values, + ) + attempts.append(attempt) + + if attempt.terminal_status == "success": + terminal = "success" + reason = None + changed_paths = list(attempt.changed_paths) + transport = { + "ref": attempt.transport_ref, + "parent": base["head"], + "attempt_id": attempt_id, + } + stop = True + break + if attempt.terminal_status == "unavailable": + terminal = "unavailable" + reason = attempt.reason + continue + if attempt.terminal_status == "failed-clean": + terminal = "unavailable" + reason = attempt.reason + continue + if attempt.terminal_status == "scope-expansion": + terminal = "scope-expansion" + reason = attempt.reason + changed_paths = list(attempt.changed_paths) + stop = True + break + if attempt.terminal_status in {"failed-unknown", "blocked"}: + terminal = attempt.terminal_status + reason = attempt.reason + stop = True + break + if attempt.terminal_status == "cancelled": + terminal = "cancelled" + reason = attempt.reason + stop = True + break + + exhausted_ok = ( + not stop + and attempts + and all(a.terminal_status in {"unavailable", "failed-clean"} for a in attempts) + and len(attempts) == len(profiles) + ) + if exhausted_ok and cfg.execution.fallback_to_host == "all-unavailable": + terminal = "host-fallback-authorized" + host_fallback_authorizations = 1 + host_scope = { + "allowed_paths": list(unit.allowed_paths), + "unit_id": unit.unit_id, + "goal": unit.goal, + "verification": list(unit.verification), + "constraints": list(unit.constraints), + "network": unit.network, + "external_cost": unit.external_cost, + } + reason = "all configured profiles exhausted with unavailable or failed-clean" + + summary = _bounded_summary( + _build_run_summary(terminal, attempts, reason), + cfg.execution.summary_bytes, + secret_values, + ) + receipt = _base_receipt( + run_id=run_id, + repo=repo, + role=role, + unit_id=unit.unit_id, + config=cfg, + terminal_status=terminal, + reason=reason, + attempts=attempts, + secret_values=secret_values, + ) + receipt.update( + { + "unit_digest": unit.digest, + "config_digest": config_digest, + "base": {"head": base["head"], "tree": base["tree"], "branch": base["branch"]}, + "summary": summary, + "changed_paths": changed_paths, + "transport": transport, + "host_scope": host_scope, + "host_fallback_authorizations": host_fallback_authorizations, + "run_dir": str(run_dir), + "private_root_mode": oct(run_dir.parent.stat().st_mode & 0o777), + } + ) + _write_receipt( + receipt_path, + receipt, + secret_values=secret_values, + summary_bytes=cfg.execution.summary_bytes, + ) + return _exit_for_terminal(terminal) + + +def load_unit_packet(path: Path, cfg: ExecutionConfig) -> UnitPacket: + try: + raw_text = path.read_text(encoding="utf-8") + except OSError as exc: + raise UnitPacketError(f"unable to read unit packet: {exc}") from exc + try: + raw = json.loads(raw_text) + except json.JSONDecodeError as exc: + raise UnitPacketError(f"unit packet is not JSON: {exc}") from exc + if not isinstance(raw, dict): + raise UnitPacketError("unit packet must be an object") + missing = [k for k in UNIT_REQUIRED if k not in raw] + if missing: + raise UnitPacketError(f"missing unit packet fields: {missing}") + + unit_id = raw["unit_id"] + goal = raw["goal"] + if not isinstance(unit_id, str) or not unit_id.strip(): + raise UnitPacketError("unit_id must be a non-empty string") + if not isinstance(goal, str) or not goal.strip(): + raise UnitPacketError("goal must be a non-empty string") + + allowed = raw["allowed_paths"] + if not isinstance(allowed, list) or not allowed: + raise UnitPacketError("allowed_paths must be a non-empty list") + paths: list[str] = [] + for item in allowed: + if not isinstance(item, str) or not item.strip(): + raise UnitPacketError("allowed_paths entries must be non-empty strings") + _validate_allowed_path(item) + paths.append(item) + + for key in ("verification", "constraints"): + if not isinstance(raw[key], list): + raise UnitPacketError(f"{key} must be a list") + for key in ("network", "external_cost"): + if type(raw[key]) is not bool: + raise UnitPacketError(f"{key} must be a boolean") + + timeout = raw["timeout_seconds"] + if type(timeout) is not int or isinstance(timeout, bool) or timeout < 1: + raise UnitPacketError("timeout_seconds must be a positive integer") + if timeout > cfg.execution.timeout_seconds: + raise UnitPacketError( + f"timeout_seconds {timeout} exceeds project policy " + f"{cfg.execution.timeout_seconds}" + ) + + digest = hashlib.sha256(raw_text.encode("utf-8")).hexdigest() + return UnitPacket( + unit_id=unit_id.strip(), + goal=goal.strip(), + allowed_paths=tuple(paths), + verification=tuple(raw["verification"]), + constraints=tuple(raw["constraints"]), + network=bool(raw["network"]), + external_cost=bool(raw["external_cost"]), + timeout_seconds=timeout, + raw=raw, + digest=digest, + ) + + +def _validate_allowed_path(path: str) -> None: + if path.startswith("/") or path.startswith("~"): + raise UnitPacketError(f"absolute allowed path rejected: {path!r}") + parts = Path(path).parts + if ".." in parts: + raise UnitPacketError(f"parent traversal rejected: {path!r}") + if ".git" in parts or path == ".git" or path.startswith(".git/"): + raise UnitPacketError(f".git path rejected: {path!r}") + lowered = path.lower() + for marker in SECRET_PATH_MARKERS: + if marker.lower() in lowered: + raise UnitPacketError(f"secret-looking path rejected: {path!r}") + + +def _select_profiles(cfg: ExecutionConfig, role: str) -> list[str]: + if role not in cfg.roles: + raise ConfigError(f"unknown role: {role!r}") + names: list[str] = [] + for name in cfg.roles[role]: + profile = cfg.profiles[name] + if not profile.enabled: + continue + names.append(name) + if not names: + raise ConfigError(f"role {role!r} has no enabled profiles") + return names + + +def _require_canonical_ready(repo: Path) -> dict[str, str]: + repo = repo.resolve() + if not (repo / ".git").exists(): + raise RuntimeError(f"not a git repository: {repo}") + status = _git(repo, "status", "--porcelain") + if status.stdout.strip(): + raise RuntimeError("canonical checkout is dirty") + branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip() + if branch in DEFAULT_BRANCHES: + raise RuntimeError(f"canonical checkout must not be on default branch {branch}") + if branch == "HEAD": + # Detached HEAD is acceptable only as a fixed base; still ok for tests that + # check out a branch. Prefer named non-default branches. + pass + head = _git(repo, "rev-parse", "HEAD").stdout.strip() + tree = _git(repo, "rev-parse", "HEAD^{tree}").stdout.strip() + return {"head": head, "tree": tree, "branch": branch} + + +def _create_private_run_dir(run_id: str) -> Path: + root = Path(f"/tmp/agent-exec-{os.getuid()}") + if root.exists(): + if root.is_symlink(): + raise RuntimeError(f"private root must not be a symlink: {root}") + if not root.is_dir(): + raise RuntimeError(f"private root is not a directory: {root}") + st = root.stat() + if st.st_uid != os.getuid(): + raise RuntimeError(f"private root owned by another user: {root}") + os.chmod(root, 0o700) + else: + root.mkdir(mode=0o700) + os.chmod(root, 0o700) + run_dir = root / run_id + run_dir.mkdir(mode=0o700) + os.chmod(run_dir, 0o700) + if run_dir.is_symlink(): + raise RuntimeError(f"run dir must not be a symlink: {run_dir}") + return run_dir + + +def _run_attempt( + *, + repo: Path, + cfg: ExecutionConfig, + profile: Profile, + unit: UnitPacket, + unit_path: Path, + base: dict[str, str], + run_id: str, + attempt_id: str, + run_dir: Path, + secret_values: set[str], +) -> AttemptRecord: + probe = probe_adapter(profile, cfg) + if not probe.available: + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status="unavailable", + reason=probe.reason or "adapter unavailable", + observed_version=probe.observed_version, + cleanup={"exact_restore": True, "skipped": True}, + ) + + worktree = run_dir / f"wt-{attempt_id}" + try: + _git( + repo, + "worktree", + "add", + "--detach", + str(worktree), + base["head"], + ) + except subprocess.CalledProcessError as exc: + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status="unavailable", + reason=f"worktree create failed: {exc.stderr}", + cleanup={"exact_restore": True, "skipped": True}, + observed_version=probe.observed_version, + ) + + result_path = run_dir / f"result-{attempt_id}.json" + packet_copy = run_dir / f"unit-{attempt_id}.json" + shutil.copy2(unit_path, packet_copy) + raw_log = run_dir / f"raw-{attempt_id}.log" + + try: + invocation = build_invocation( + profile, cfg, worktree, packet_copy, result_path + ) + except ValueError as exc: + _safe_remove_worktree(repo, worktree) + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status="unavailable", + reason=str(exc), + worktree=str(worktree), + cleanup={"exact_restore": True}, + observed_version=probe.observed_version, + ) + + try: + _validate_allowed_path_symlinks(worktree, unit.allowed_paths) + except UnitPacketError as exc: + _safe_remove_worktree(repo, worktree) + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status="failed-unknown", + reason=str(exc), + worktree=str(worktree), + cleanup={"exact_restore": True, "reason": "symlink escape before launch"}, + observed_version=probe.observed_version, + ) + + admin_before = _git_admin_snapshot(worktree, repo) + pointer_before = _worktree_pointer_snapshot(worktree, require_git=True) + canonical_before = _canonical_checkout_snapshot(repo) + timeout = min(unit.timeout_seconds, cfg.execution.timeout_seconds) + proc_meta = _supervise( + invocation.argv, + cwd=invocation.cwd, + stdin=invocation.stdin, + timeout=timeout, + raw_log=raw_log, + raw_cap=cfg.execution.raw_output_bytes, + secret_values=secret_values, + ) + + changed: list[str] = [] + inventory: dict[str, Any] = {"paths": [], "details": []} + try: + pointer_after = _worktree_pointer_snapshot(worktree, require_git=True) + admin_after = _git_admin_snapshot(worktree, repo) + canonical_after = _canonical_checkout_snapshot(repo) + inventory = _inventory_changes(worktree) + changed = list(inventory["paths"]) + except Exception as exc: + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status="failed-unknown", + reason=f"post-launch git inspection failed: {exc}", + worktree=str(worktree), + pid=proc_meta["pid"], + exit_code=proc_meta["exit_code"], + timed_out=proc_meta["timed_out"], + signal=proc_meta["signal"], + changed_paths=[], + inventory_paths=[], + raw_log=str(raw_log), + summary=_bounded_summary( + proc_meta.get("summary_source") or str(exc), + cfg.execution.summary_bytes, + secret_values, + ), + workspace_preserved=True, + cleanup={ + "exact_restore": False, + "reason": "post-launch snapshot or inventory failure", + "error": str(exc), + "stdout_tail_bytes": len(proc_meta.get("stdout_tail") or ""), + "stderr_tail_bytes": len(proc_meta.get("stderr_tail") or ""), + }, + observed_version=probe.observed_version, + ) + + admin_delta = _git_admin_delta(admin_before, admin_after) + 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"]) + summary = _bounded_summary( + proc_meta.get("summary_source") or (worker_result or {}).get("summary") or "", + cfg.execution.summary_bytes, + secret_values, + ) + + def _preserved( + status: str, + reason: str, + *, + exact_restore: bool = False, + cleanup_extra: dict[str, Any] | None = None, + ) -> AttemptRecord: + cleanup: dict[str, Any] = {"exact_restore": exact_restore} + if cleanup_extra: + cleanup.update(cleanup_extra) + cleanup.setdefault( + "stdout_tail_bytes", len(proc_meta.get("stdout_tail") or "") + ) + cleanup.setdefault( + "stderr_tail_bytes", len(proc_meta.get("stderr_tail") or "") + ) + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status=status, + reason=reason, + worktree=str(worktree), + pid=proc_meta["pid"], + exit_code=proc_meta["exit_code"], + timed_out=proc_meta["timed_out"], + signal=proc_meta["signal"], + changed_paths=list(changed), + inventory_paths=list(changed), + raw_log=str(raw_log), + summary=summary, + workspace_preserved=not exact_restore, + cleanup=cleanup, + observed_version=probe.observed_version, + ) + + if pointer_delta: + return _preserved( + "failed-unknown", + f"worktree .git pointer/admin identity changed: {', '.join(pointer_delta)}", + cleanup_extra={ + "reason": "worktree git pointer mismatch", + "pointer_delta": pointer_delta, + }, + ) + + # Git administration snapshot dominates worker prose. + if admin_delta: + return _preserved( + "failed-unknown", + f"unexplained git administration change: {', '.join(admin_delta)}", + cleanup_extra={"reason": "git admin snapshot mismatch", "admin_delta": admin_delta}, + ) + + if canonical_delta: + return _preserved( + "failed-unknown", + f"canonical checkout mutated: {', '.join(canonical_delta)}", + cleanup_extra={ + "reason": "canonical checkout snapshot mismatch", + "canonical_delta": canonical_delta, + }, + ) + + escaped = [p for p in changed if _path_symlink_escapes(worktree, p)] + if escaped: + return _preserved( + "failed-unknown", + f"changed path escapes worktree via symlink: {escaped}", + cleanup_extra={"reason": "symlink escape on changed paths"}, + ) + + out_of_scope = [p for p in changed if not _path_allowed(p, unit.allowed_paths)] + worker_requests_expand = bool( + worker_result + and worker_result.get("terminal_status") in {"scope_expansion", "scope-expansion"} + ) + if out_of_scope or worker_requests_expand: + return _preserved( + "scope-expansion", + ( + f"paths outside allowed_paths: {out_of_scope}" + if out_of_scope + else "worker requested scope expansion" + ), + cleanup_extra={"preserved": True}, + ) + + if proc_meta["timed_out"] or proc_meta["signal"] is not None: + # Timeout/disconnect cannot prove absence of external side effects. + return _preserved( + "failed-unknown", + ( + "timed out; process group terminated" + if proc_meta["timed_out"] + else f"abnormal termination signal={proc_meta['signal']}" + ), + cleanup_extra={"reason": "unproven external side effects"}, + ) + + clean_exit = proc_meta["exit_code"] == 0 + # Never report success from worker prose alone — require exit + admin + in-scope diff + transport. + if clean_exit and changed and not out_of_scope: + # Recheck path bounds immediately before host-owned transport. + try: + _validate_allowed_path_symlinks(worktree, tuple(changed)) + except UnitPacketError as exc: + return _preserved("failed-unknown", str(exc)) + try: + transport_ref = _create_transport_commit( + repo=repo, + worktree=worktree, + base_head=base["head"], + run_id=run_id, + attempt_id=attempt_id, + inventory=inventory, + ) + except Exception as exc: + # Staging / commit-tree / update-ref failure after a clean worker exit + # must stay recoverable: retained failed-unknown, no routing, no fallback. + return _preserved( + "failed-unknown", + f"transport construction failed: {exc}", + cleanup_extra={ + "exact_restore": False, + "transport_created": False, + "reason": "transport construction failure", + "error": str(exc), + }, + ) + return AttemptRecord( + attempt_id=attempt_id, + profile=profile.name, + adapter=profile.adapter, + terminal_status="success", + reason=None, + 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, + transport_ref=transport_ref, + workspace_preserved=True, + cleanup={ + "exact_restore": False, + "transport_created": True, + "stdout_tail_bytes": len(proc_meta.get("stdout_tail") or ""), + "stderr_tail_bytes": len(proc_meta.get("stderr_tail") or ""), + }, + observed_version=probe.observed_version, + ) + + if clean_exit and not changed: + restored = _restore_worktree(repo, worktree, base["head"]) + 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": restored}, + 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"]) + 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": restored, + "external_objects_reconciled": True, + "network": False, + "external_cost": False, + }, + observed_version=probe.observed_version, + ) + + +def _supervise( + argv: tuple[str, ...], + *, + cwd: Path, + stdin: bytes | None, + timeout: int, + raw_log: Path, + raw_cap: int, + secret_values: set[str], +) -> dict[str, Any]: + env = _filtered_env() + # Capture for redaction only — never serialize into receipts. + for key, value in list(env.items()): + if SECRET_VALUE_RE.match(value.strip()) or "SECRET" in key.upper() or "TOKEN" in key.upper() or "KEY" in key.upper(): + if value: + secret_values.add(value) + + timed_out = False + exit_code: int | None = None + sig: int | None = None + pid: int | None = None + stdout_tail = _ByteTail(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. + secret_lookahead = max((len(s) for s in secret_values), default=0) + secret_lookahead = min(max(secret_lookahead, 64), 8192) + prefix_limit = max(raw_cap, 0) + secret_lookahead + raw_prefix = bytearray() + raw_written = 0 + prefix_sealed = False + + def _preexec() -> None: + os.setsid() + + try: + proc = subprocess.Popen( + list(argv), + cwd=str(cwd), + stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + preexec_fn=_preexec, + bufsize=0, + ) + except OSError as exc: + raw_log.write_bytes(_cap_bytes(str(exc).encode("utf-8", "replace"), raw_cap)) + return { + "pid": None, + "exit_code": 127, + "timed_out": False, + "signal": None, + "stdout_tail": "", + "stderr_tail": "", + "summary_source": str(exc), + } + + assert proc.stdout is not None and proc.stderr is not None + pid = proc.pid + + def _set_nonblocking(fh: Any) -> None: + fd = fh.fileno() + flags = fcntl.fcntl(fd, fcntl.F_GETFL) + fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) + + _set_nonblocking(proc.stdout) + _set_nonblocking(proc.stderr) + stdin_fh = proc.stdin + stdin_view: memoryview | None = None + stdin_offset = 0 + if stdin is not None and stdin_fh is not None: + _set_nonblocking(stdin_fh) + stdin_view = memoryview(stdin) + elif stdin_fh is not None: + try: + stdin_fh.close() + except OSError: + pass + stdin_fh = None + + streams = {proc.stdout: stdout_tail, proc.stderr: stderr_tail} + deadline = time.monotonic() + max(timeout, 1) + grace_deadline: float | None = None + + with raw_log.open("wb") as raw_fh: + + def _seal_raw_prefix() -> None: + nonlocal raw_written, prefix_sealed, raw_prefix + if prefix_sealed: + return + prefix_sealed = True + if not raw_prefix: + return + redacted = _redact_bytes(bytes(raw_prefix), secret_values) + raw_prefix.clear() + if raw_cap > 0: + take = redacted[:raw_cap] + if take: + raw_fh.write(take) + raw_written = len(take) + raw_fh.flush() + + def _emit_raw(chunk: bytes) -> None: + nonlocal raw_prefix + if not chunk: + return + if prefix_sealed: + # Cap reached earlier; keep draining but discard excess. + return + remaining = prefix_limit - len(raw_prefix) + if remaining <= 0: + _seal_raw_prefix() + return + if len(chunk) <= remaining: + raw_prefix.extend(chunk) + if len(raw_prefix) >= prefix_limit: + _seal_raw_prefix() + return + raw_prefix.extend(chunk[:remaining]) + _seal_raw_prefix() + + def _drain_fd(fh: Any, tail: _ByteTail) -> bool: + """Read one chunk. True = keep watching; False = EOF/error (drop stream).""" + try: + chunk = os.read(fh.fileno(), 65536) + except BlockingIOError: + return True + except OSError: + return False + if not chunk: + return False + tail.extend(chunk) + _emit_raw(chunk) + return True + + def _drain_fd_available(fh: Any, tail: _ByteTail) -> None: + """Drain until EOF, error, or would-block (final post-exit sweep).""" + while True: + try: + chunk = os.read(fh.fileno(), 65536) + except BlockingIOError: + return + except OSError: + return + if not chunk: + return + tail.extend(chunk) + _emit_raw(chunk) + + def _write_stdin() -> bool: + """Write a nonblocking stdin slice. Return False when stdin is done/closed.""" + nonlocal stdin_fh, stdin_offset + if stdin_fh is None or stdin_view is None: + return False + if stdin_offset >= len(stdin_view): + try: + stdin_fh.close() + except OSError: + pass + stdin_fh = None + return False + try: + written = os.write( + stdin_fh.fileno(), + stdin_view[stdin_offset : stdin_offset + 65536], + ) + except BrokenPipeError: + try: + stdin_fh.close() + except OSError: + pass + stdin_fh = None + return False + except BlockingIOError: + return True + except OSError: + try: + stdin_fh.close() + except OSError: + pass + stdin_fh = None + return False + stdin_offset += written + if stdin_offset >= len(stdin_view): + try: + stdin_fh.close() + except OSError: + pass + stdin_fh = None + return False + return True + + while streams or proc.poll() is None or stdin_fh is not None: + now = time.monotonic() + if grace_deadline is None and now >= deadline: + timed_out = True + _kill_process_group(proc.pid) + grace_deadline = now + 5 + if stdin_fh is not None: + try: + stdin_fh.close() + except OSError: + pass + stdin_fh = None + if grace_deadline is not None and now >= grace_deadline: + break + wait = 0.2 + if grace_deadline is None: + wait = max(0.0, min(wait, deadline - now)) + else: + wait = max(0.0, min(wait, grace_deadline - now)) + readers = list(streams.keys()) + writers: list[Any] = [] + if stdin_fh is not None and stdin_view is not None and stdin_offset < len(stdin_view): + writers.append(stdin_fh) + if not readers and not writers: + # Process still alive but pipes already closed; wait for exit/timeout. + time.sleep(min(wait, 0.05)) + continue + readable, writable, _ = select.select(readers, writers, [], wait) + for fh in writable: + if fh is stdin_fh: + _write_stdin() + for fh in readable: + alive = _drain_fd(fh, streams[fh]) + if not alive: + streams.pop(fh, None) + try: + fh.close() + except OSError: + pass + if proc.poll() is not None and not readable and stdin_fh is None: + # Final drain after exit when select reports idle pipes. + for fh in list(streams.keys()): + _drain_fd_available(fh, streams[fh]) + streams.pop(fh, None) + try: + fh.close() + except OSError: + pass + break + + _seal_raw_prefix() + + if proc.poll() is None: + timed_out = True + _kill_process_group(proc.pid) + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + pass + exit_code = proc.returncode + + if exit_code is not None and exit_code < 0: + sig = -exit_code + exit_code = 128 + sig + + stdout_text = stdout_tail.text() + stderr_text = stderr_tail.text() + summary_source = _redact_text(stdout_text[-2000:], secret_values) + return { + "pid": pid, + "exit_code": exit_code, + "timed_out": timed_out, + "signal": sig, + "stdout_tail": stdout_text, + "stderr_tail": stderr_text, + "summary_source": summary_source, + } + + +class _ByteTail: + """Bounded in-memory byte tail used while draining child output.""" + + def __init__(self, limit: int) -> None: + self.limit = max(1, limit) + self._data = bytearray() + + def extend(self, chunk: bytes) -> None: + if not chunk: + return + self._data.extend(chunk) + if len(self._data) > self.limit: + self._data = self._data[-self.limit :] + + def text(self) -> str: + return bytes(self._data).decode("utf-8", "replace") + + +def _kill_process_group(pid: int) -> None: + try: + os.killpg(pid, signal.SIGKILL) + except ProcessLookupError: + pass + except PermissionError: + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + + +def _filtered_env() -> dict[str, str]: + out: dict[str, str] = {} + for key, value in os.environ.items(): + if key in ENV_ALLOW_EXACT or key.startswith(ENV_ALLOW_PREFIXES): + out[key] = value + # Ensure PATH/HOME exist for executables. + if "PATH" not in out and "PATH" in os.environ: + out["PATH"] = os.environ["PATH"] + if "HOME" not in out and "HOME" in os.environ: + out["HOME"] = os.environ["HOME"] + return out + + +def _inventory_changes(worktree: Path) -> dict[str, Any]: + """Inventory tracked, staged, untracked, deleted, renamed, mode, and binary changes. + + Staged index entries are collected explicitly so they cannot be silently dropped. + """ + paths: list[str] = [] + details: list[dict[str, Any]] = [] + + def _consume_name_status(text: str, *, prefix: str) -> None: + for line in text.splitlines(): + if not line.strip(): + continue + parts = line.split("\t") + status = parts[0] + if status.startswith("R") and len(parts) >= 3: + paths.append(parts[2]) + details.append( + { + "status": f"{prefix}renamed", + "from": parts[1], + "path": parts[2], + } + ) + elif len(parts) >= 2: + paths.append(parts[1]) + details.append({"status": f"{prefix}{status[:1]}", "path": parts[1]}) + + # Explicit staged inventory (index vs HEAD) — fail-visible, never silent. + staged = _git( + worktree, + "diff", + "--cached", + "--name-status", + "--find-renames", + check=False, + ) + if staged.returncode != 0: + raise RuntimeError(f"staged inventory failed: {staged.stderr.strip()}") + _consume_name_status(staged.stdout, prefix="staged:") + + # Working tree vs HEAD (covers unstaged tracked modifications / deletions). + named = _git( + worktree, + "diff", + "--name-status", + "--find-renames", + "HEAD", + check=False, + ) + if named.returncode != 0: + raise RuntimeError(f"worktree inventory failed: {named.stderr.strip()}") + _consume_name_status(named.stdout, prefix="") + + # Untracked (including binary); ignore ignored files. + untracked = _git( + worktree, + "ls-files", + "--others", + "--exclude-standard", + check=False, + ) + if untracked.returncode != 0: + raise RuntimeError(f"untracked inventory failed: {untracked.stderr.strip()}") + for line in untracked.stdout.splitlines(): + path = line.strip() + if not path: + continue + if path.startswith(".git/") or "/.git/" in path: + continue + paths.append(path) + details.append({"status": "untracked", "path": path}) + + # Executable mode bits among tracked diffs. + mode = _git(worktree, "diff", "--summary", "HEAD", check=False) + if mode.returncode != 0: + raise RuntimeError(f"mode inventory failed: {mode.stderr.strip()}") + for line in mode.stdout.splitlines(): + if "mode change" in line or "change from" in line: + details.append({"status": "mode", "raw": line.strip()}) + + # Deduplicate while preserving order. + seen: set[str] = set() + ordered: list[str] = [] + for p in paths: + if p not in seen: + seen.add(p) + ordered.append(p) + return {"paths": ordered, "details": details} + +def _path_allowed(path: str, allowed: tuple[str, ...]) -> bool: + norm = path.replace("\\", "/").lstrip("./") + if ".." in Path(norm).parts: + return False + for rule in allowed: + rule_n = rule.replace("\\", "/").lstrip("./") + if norm == rule_n or norm.startswith(rule_n.rstrip("/") + "/"): + return True + # Exact file rule: also allow if rule is a prefix directory. + if rule_n.endswith("/") and norm.startswith(rule_n): + return True + return False + + +def _git_admin_snapshot(worktree: Path, repo: Path) -> dict[str, str]: + """Snapshot Git administration state before/after worker launch.""" + head = _git(worktree, "rev-parse", "HEAD").stdout.strip() + index_tree = _git(worktree, "write-tree").stdout.strip() + staged_proc = _git( + worktree, + "diff", + "--cached", + "--name-status", + "--find-renames", + check=False, + ) + if staged_proc.returncode != 0: + raise RuntimeError(f"admin staged snapshot failed: {staged_proc.stderr.strip()}") + staged = staged_proc.stdout + local_config_proc = _git(repo, "config", "--local", "--list", check=False) + if local_config_proc.returncode != 0: + raise RuntimeError( + f"admin local-config snapshot failed: {local_config_proc.stderr.strip()}" + ) + local_config = "\n".join( + sorted(line for line in local_config_proc.stdout.splitlines() if line) + ) + refs_proc = _git( + repo, + "for-each-ref", + "--format=%(refname) %(objectname)", + check=False, + ) + if refs_proc.returncode != 0: + raise RuntimeError(f"admin refs snapshot failed: {refs_proc.stderr.strip()}") + refs = "\n".join(sorted(line for line in refs_proc.stdout.splitlines() if line)) + return { + "head": head, + "index_tree": index_tree, + "staged": staged, + "local_config": local_config, + "refs": refs, + } + + +def _git_admin_delta(before: dict[str, str], after: dict[str, str]) -> list[str]: + return _snapshot_delta(before, after) + + +def _snapshot_delta(before: dict[str, str], after: dict[str, str]) -> list[str]: + changed: list[str] = [] + for key in sorted(set(before) | set(after)): + if before.get(key) != after.get(key): + changed.append(key) + return changed + + +def _canonical_checkout_snapshot(repo: Path) -> dict[str, str]: + """Snapshot the canonical checkout identity (not the attempt worktree).""" + repo = repo.resolve() + head = _git(repo, "rev-parse", "HEAD").stdout.strip() + branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip() + index_tree = _git(repo, "write-tree").stdout.strip() + status = _git(repo, "status", "--porcelain=v2", "-uall").stdout + git_dir = _git(repo, "rev-parse", "--absolute-git-dir").stdout.strip() + common = _git( + repo, "rev-parse", "--path-format=absolute", "--git-common-dir" + ).stdout.strip() + worktrees = _git(repo, "worktree", "list", "--porcelain").stdout + return { + "head": head, + "branch": branch, + "index_tree": index_tree, + "status": status, + "git_dir": git_dir, + "common_dir": common, + "worktree_list": worktrees, + } + + +def _worktree_pointer_snapshot( + worktree: Path, *, require_git: bool = True +) -> dict[str, str]: + """Snapshot the detached worktree `.git` pointer / admin identity.""" + git_path = worktree / ".git" + if git_path.is_symlink(): + kind = "symlink" + try: + content = os.readlink(git_path) + except OSError as exc: + raise RuntimeError(f"unable to read worktree .git symlink: {exc}") from exc + elif git_path.is_file(): + kind = "file" + try: + content = git_path.read_text(encoding="utf-8") + except OSError as exc: + raise RuntimeError(f"unable to read worktree .git file: {exc}") from exc + elif git_path.is_dir(): + kind = "dir" + content = "dir" + else: + kind = "missing" + content = "" + snap = {"kind": kind, "content": content, "abs_git": "", "common_dir": ""} + if require_git: + snap["abs_git"] = _git( + worktree, "rev-parse", "--absolute-git-dir" + ).stdout.strip() + snap["common_dir"] = _git( + worktree, "rev-parse", "--path-format=absolute", "--git-common-dir" + ).stdout.strip() + return snap + + +def _external_side_effects_unproven( + unit: UnitPacket, worker_result: dict[str, Any] | None +) -> bool: + """Nonzero/aborted workers cannot be failed-clean when external effects are open. + + failed-clean requires exact restore plus affirmative evidence that no authorized + external object remains unknown. Units with network/external_cost, or without a + trustworthy structured terminal receipt proving reconciliation, stay unknown. + """ + 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()) + return True + except (ValueError, OSError): + return False + + +def _path_symlink_escapes(root: Path, rel: str) -> bool: + """True when any existing ancestor/target of rel resolves outside root.""" + root = root.resolve() + cur = root + for part in Path(rel).parts: + if part in ("", "."): + continue + if part == "..": + return True + nxt = cur / part + if nxt.is_symlink() or nxt.exists(): + try: + resolved = nxt.resolve() + except OSError: + return True + if not _is_within_root(resolved, root): + return True + cur = resolved if nxt.is_symlink() else nxt + else: + # Remaining path does not exist yet; parent must already be inside. + return not _is_within_root(cur, root) + return False + + +def _validate_allowed_path_symlinks(root: Path, paths: tuple[str, ...]) -> None: + for rel in paths: + if _path_symlink_escapes(root, rel): + raise UnitPacketError( + f"allowed path escapes worktree via symlink: {rel!r}" + ) + + +def _create_transport_commit( + *, + repo: Path, + worktree: Path, + base_head: str, + run_id: str, + attempt_id: str, + inventory: dict[str, Any], +) -> str: + """Stage inside the isolated worktree only; create base-parented transport commit.""" + # Stage all tracked/untracked changes in the worktree. + _git(worktree, "add", "-A", "--", ".") + # Include deletions. + _git(worktree, "add", "-u", check=False) + tree = _git(worktree, "write-tree").stdout.strip() + msg = f"agent-exec transport {run_id}/{attempt_id}" + commit = _git( + worktree, + "commit-tree", + tree, + "-p", + base_head, + "-m", + msg, + ).stdout.strip() + ref = f"refs/agent-exec/{run_id}/{attempt_id}" + # Write private ref into the shared object store via the canonical repo. + # commit-tree leaves HEAD/index of both checkouts untouched; no reset needed. + _git(repo, "update-ref", ref, commit) + return ref + + +def _restore_worktree(repo: Path, worktree: Path, base_head: str) -> bool: + 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: + try: + _git(repo, "worktree", "remove", "--force", str(worktree), check=False) + except Exception: + pass + if worktree.exists(): + shutil.rmtree(worktree, ignore_errors=True) + + +def _load_worker_result( + result_path: Path, stdout_tail: str +) -> dict[str, Any] | 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 + + +def _base_receipt( + *, + run_id: str, + repo: Path, + role: str, + unit_id: str | None, + config: ExecutionConfig | None, + terminal_status: str, + reason: str | None, + attempts: list[AttemptRecord], + secret_values: set[str], +) -> dict[str, Any]: + attempt_dicts = [] + for a in attempts: + attempt_dicts.append( + { + "attempt_id": a.attempt_id, + "profile": a.profile, + "adapter": a.adapter, + "terminal_status": a.terminal_status, + "reason": a.reason, + "worktree": a.worktree, + "pid": a.pid, + "exit_code": a.exit_code, + "timed_out": a.timed_out, + "signal": a.signal, + "changed_paths": a.changed_paths, + "inventory_paths": a.inventory_paths or a.changed_paths, + "raw_log": a.raw_log, + "summary": a.summary, + "cleanup": a.cleanup, + "workspace_preserved": a.workspace_preserved, + "transport_ref": a.transport_ref, + "observed_version": a.observed_version, + } + ) + return { + "schema": RECEIPT_SCHEMA, + "run_id": run_id, + "repo": str(repo), + "role": role, + "unit_id": unit_id, + "terminal_status": terminal_status, + "reason": reason, + "attempts": attempt_dicts, + "config_summary": _config_summary(config), + "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + + +def _config_summary(config: ExecutionConfig | None) -> dict[str, Any] | None: + if config is None: + return None + return { + "schema": config.schema, + "permission": config.execution.permission, + "fallback_to_host": config.execution.fallback_to_host, + "timeout_seconds": config.execution.timeout_seconds, + "summary_bytes": config.execution.summary_bytes, + "raw_output_bytes": config.execution.raw_output_bytes, + "roles": {k: list(v) for k, v in config.roles.items()}, + "profiles": { + name: { + "adapter": p.adapter, + "executable": p.executable, + "model": p.model, + "enabled": p.enabled, + } + for name, p in config.profiles.items() + }, + } + + +def _write_receipt( + path: Path, + receipt: dict[str, Any], + *, + secret_values: set[str], + summary_bytes: int, +) -> None: + if "summary" in receipt and isinstance(receipt["summary"], str): + receipt["summary"] = _bounded_summary( + receipt["summary"], summary_bytes, secret_values + ) + text = json.dumps(receipt, indent=2, sort_keys=True) + "\n" + text = _redact_text(text, secret_values) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _build_run_summary( + terminal: str, attempts: list[AttemptRecord], reason: str | None +) -> str: + parts = [f"terminal={terminal}"] + if reason: + parts.append(f"reason={reason}") + parts.append(f"attempts={len(attempts)}") + for a in attempts: + parts.append(f"{a.profile}:{a.terminal_status}") + return "; ".join(parts) + + +def _exit_for_terminal(terminal: str) -> int: + return { + "success": EXIT_SUCCESS, + "host-fallback-authorized": EXIT_HOST_FALLBACK, + "invalid-config": EXIT_INVALID_CONFIG, + "unavailable": EXIT_UNAVAILABLE_EXHAUSTION, + "failed-clean": EXIT_UNAVAILABLE_EXHAUSTION, + "failed-unknown": EXIT_BLOCKED, + "blocked": EXIT_BLOCKED, + "scope-expansion": EXIT_SCOPE_EXPANSION, + "cancelled": EXIT_CANCELLED, + }.get(terminal, EXIT_BLOCKED) + + +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, + ) + + +def _file_digest(path: Path) -> str: + try: + return hashlib.sha256(path.read_bytes()).hexdigest() + except OSError: + return "" + + +def _collect_secret_values(out: set[str]) -> None: + for key, value in os.environ.items(): + if not value: + continue + if SECRET_VALUE_RE.match(value.strip()): + out.add(value) + if re.search(r"(SECRET|TOKEN|PASSWORD|API_KEY|CREDENTIAL)", key, re.I): + out.add(value) + + +def _redact_text(text: str, secrets: set[str]) -> str: + out = text + for secret in sorted(secrets, key=len, reverse=True): + if secret and secret in out: + out = out.replace(secret, "***") + out = SECRET_VALUE_RE.sub("***", out) + return out + + +def _redact_bytes(data: bytes, secrets: set[str]) -> bytes: + try: + text = data.decode("utf-8", "replace") + except Exception: + return data[: len(data)] + return _redact_text(text, secrets).encode("utf-8", "replace") + + +def _bounded_summary(text: str, limit: int, secrets: set[str]) -> str: + text = _redact_text(text or "", secrets) + encoded = text.encode("utf-8") + if len(encoded) <= limit: + return text + return encoded[:limit].decode("utf-8", "ignore") + + +def _cap_bytes(data: bytes, limit: int) -> bytes: + if len(data) <= limit: + return data + return data[:limit] + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py new file mode 100644 index 00000000..8827f69d --- /dev/null +++ b/tests/test_agent_exec.py @@ -0,0 +1,1372 @@ +"""End-to-end execution controller contract for agent-exec (U3).""" +from __future__ import annotations + +import json +import os +import shutil +import stat +import subprocess +import tempfile +import textwrap +import time +import unittest +import uuid +from pathlib import Path +from unittest import mock + +from scripts import agent_exec + + +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, + ) + + +def _canonical_snapshot(repo: Path) -> dict[str, str]: + head = _git(repo, "rev-parse", "HEAD").stdout.strip() + tree = _git(repo, "rev-parse", "HEAD^{tree}").stdout.strip() + status = _git(repo, "status", "--porcelain=v2", "--branch").stdout + index = _git(repo, "write-tree").stdout.strip() + return {"head": head, "tree": tree, "status": status, "index": index} + + +def _write_config( + repo: Path, + *, + profiles: dict[str, dict[str, object]], + role_order: list[str], + timeout_seconds: int = 30, + summary_bytes: int = 256, + raw_output_bytes: int = 1024, +) -> None: + profile_lines: list[str] = [] + custom_lines: list[str] = [] + for name, body in profiles.items(): + profile_lines.append(f" {name}:") + profile_lines.append(f" adapter: {body['adapter']}") + profile_lines.append(f" executable: {body['executable']}") + profile_lines.append(f" model: {body.get('model', 'auto')}") + profile_lines.append(f" enabled: {str(body.get('enabled', True)).lower()}") + if body["adapter"] == "custom": + custom_lines.append(f" {name}:") + custom_lines.append(' base_argv: ["${workspace}", "${unit_packet}", "${result_path}"]') + custom_lines.append(" run_everything_argv: []") + custom_lines.append(" workspace_argv: []") + custom_lines.append(" model_argv: []") + custom_lines.append(" prompt_transport: stdin") + custom_lines.append(" output_format: text") + order = ", ".join(role_order) + first = role_order[0] + block = textwrap.dedent( + f"""\ + schema: agent-execution/v1 + + execution: + fresh_process_per_unit: true + permission: run-everything + fallback_to_host: all-unavailable + max_parallel: 1 + timeout_seconds: {timeout_seconds} + summary_bytes: {summary_bytes} + raw_output_bytes: {raw_output_bytes} + + roles: + implementation: [{order}] + repair: [{first}] + test: [{first}] + review: [{first}] + + profiles: + """ + ) + block += "\n".join(profile_lines) + "\n" + if custom_lines: + block += "\ncustom_adapters:\n" + "\n".join(custom_lines) + "\n" + (repo / "AGENT-EXECUTION.md").write_text( + "# Fixture\n\n```agent-execution\n" + block + "```\n", + encoding="utf-8", + ) + _git(repo, "add", "AGENT-EXECUTION.md") + _git(repo, "commit", "-m", "add agent-execution config") + + +def _write_unit( + path: Path, + *, + unit_id: str = "u-test", + goal: str = "Create marker file", + allowed_paths: list[str] | None = None, + timeout_seconds: int = 20, +) -> Path: + packet = { + "unit_id": unit_id, + "goal": goal, + "allowed_paths": allowed_paths or ["src/marker.txt"], + "verification": ["test -f src/marker.txt"], + "constraints": ["leave uncommitted"], + "network": False, + "external_cost": False, + "timeout_seconds": timeout_seconds, + } + path.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8") + return path + + +def _write_fake_worker( + bindir: Path, + name: str, + *, + body: str, + help_text: str = "custom worker\n", +) -> Path: + path = bindir / name + path.write_text( + "#!/bin/sh\n" + 'if [ "$1" = "--version" ] || [ "$1" = "-v" ] || [ "$1" = "-V" ]; then\n' + ' echo "0.0.1-test"\n' + " exit 0\n" + "fi\n" + 'if [ "$1" = "--help" ] || [ "$1" = "-h" ]; then\n' + f" cat <<'HELP'\n{help_text}HELP\n" + " exit 0\n" + "fi\n" + + body, + encoding="utf-8", + ) + path.chmod(path.stat().st_mode | stat.S_IEXEC) + return path + + +class AgentExecControllerTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-ctrl-") + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.bindir = self.root / "bin" + self.bindir.mkdir() + self.repo = self.root / "repo" + self.repo.mkdir() + self.receipts = self.root / "receipts" + self.receipts.mkdir() + self._old_path = os.environ.get("PATH", "") + os.environ["PATH"] = f"{self.bindir}{os.pathsep}{self._old_path}" + self.addCleanup(self._restore_path) + self._init_repo() + + def _restore_path(self) -> None: + os.environ["PATH"] = self._old_path + + def _init_repo(self) -> None: + _git(self.repo, "init") + _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") + (self.repo / "src").mkdir() + (self.repo / "src" / ".gitkeep").write_text("", encoding="utf-8") + _git(self.repo, "add", "README.md", "src/.gitkeep") + _git(self.repo, "commit", "-m", "init") + _git(self.repo, "checkout", "-b", "feature/agent-exec-test") + self.base_snap = _canonical_snapshot(self.repo) + + def _run_cli(self, *args: str, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + cmd_env = os.environ.copy() + if env: + cmd_env.update(env) + cmd_env["PYTHONDONTWRITEBYTECODE"] = "1" + workspace = Path(__file__).resolve().parents[1] + existing = cmd_env.get("PYTHONPATH", "") + cmd_env["PYTHONPATH"] = ( + f"{workspace}{os.pathsep}{existing}" if existing else str(workspace) + ) + return subprocess.run( + ["python3", "-m", "scripts.agent_exec", *args], + cwd=str(self.root), + capture_output=True, + text=True, + env=cmd_env, + ) + + def _assert_canonical_untouched(self) -> None: + self.assertEqual(_canonical_snapshot(self.repo), self.base_snap) + + def _configure(self, **kwargs: object) -> None: + _write_config(self.repo, **kwargs) # type: ignore[arg-type] + self.base_snap = _canonical_snapshot(self.repo) + + def test_validate_does_not_contact_provider(self) -> None: + marker = self.root / "provider-contacted" + _write_fake_worker( + self.bindir, + "cursor-agent", + body=( + f'echo contacted > "{marker}"\n' + "exit 1\n" + ), + help_text=textwrap.dedent( + """\ + Usage: agent [options] + -p, --print + --force + --sandbox + --trust + --workspace + --model + --output-format + """ + ), + ) + self._configure( + profiles={ + "cursor": { + "adapter": "cursor", + "executable": "cursor-agent", + "model": "auto", + } + }, + role_order=["cursor"], + ) + proc = self._run_cli("validate", "--repo", str(self.repo)) + self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout) + self.assertFalse(marker.exists()) + self._assert_canonical_untouched() + + 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", + body=textwrap.dedent( + f"""\ + workspace="$1" + result="$3" + echo "$$" > "{meta}/b.pid" + echo "$workspace" > "{meta}/b.wt" + 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" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "a": {"adapter": "custom", "executable": "worker-a"}, + "b": {"adapter": "custom", "executable": "worker-b"}, + }, + role_order=["a", "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()}/")) + self._assert_canonical_untouched() + + def test_missing_profiles_fall_through_in_declared_order(self) -> None: + log = self.root / "order.log" + log.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "second", + body=textwrap.dedent( + f"""\ + echo second >> "{log}" + workspace="$1" + result="$3" + 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":[],"scope_expansion":null}}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "missing": {"adapter": "custom", "executable": "does-not-exist-xyz"}, + "second": {"adapter": "custom", "executable": "second"}, + }, + role_order=["missing", "second"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "order.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")) + names = [a["profile"] for a in data["attempts"]] + self.assertEqual(names, ["missing", "second"]) + self.assertEqual(data["attempts"][0]["terminal_status"], "unavailable") + self.assertEqual(log.read_text(encoding="utf-8").strip(), "second") + self._assert_canonical_untouched() + + def test_all_unavailable_authorizes_host_fallback_once(self) -> None: + self._configure( + profiles={ + "m1": {"adapter": "custom", "executable": "gone-one"}, + "m2": {"adapter": "custom", "executable": "gone-two"}, + }, + role_order=["m1", "m2"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "fallback.json" + before = list(self.repo.rglob("*")) + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + self.assertEqual( + proc.returncode, agent_exec.EXIT_HOST_FALLBACK, proc.stderr + proc.stdout + ) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "host-fallback-authorized") + self.assertEqual(len(data["attempts"]), 2) + self.assertTrue(all(a["terminal_status"] == "unavailable" for a in data["attempts"])) + self.assertEqual(data.get("host_fallback_authorizations"), 1) + # Controller must not implement the unit itself. + after = list(self.repo.rglob("*")) + self.assertEqual(before, after) + self.assertFalse((self.repo / "src" / "marker.txt").exists()) + self.assertIn("allowed_paths", data["host_scope"]) + self._assert_canonical_untouched() + + def test_failed_clean_can_continue_after_exact_restore(self) -> None: + _write_fake_worker( + self.bindir, + "dirty", + body=textwrap.dedent( + """\ + workspace="$1" + mkdir -p "$workspace/src" + printf 'tmp\\n' > "$workspace/src/marker.txt" + exit 9 + """ + ), + ) + _write_fake_worker( + 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" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "dirty": {"adapter": "custom", "executable": "dirty"}, + "ok": {"adapter": "custom", "executable": "ok"}, + }, + role_order=["dirty", "ok"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "clean.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["attempts"][0]["terminal_status"], "failed-clean") + self.assertTrue(data["attempts"][0]["cleanup"]["exact_restore"]) + self.assertEqual(data["terminal_status"], "success") + self._assert_canonical_untouched() + + def test_failed_unknown_blocks_fallback(self) -> None: + launched = self.root / "launched.log" + launched.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "unknown", + body=textwrap.dedent( + """\ + workspace="$1" + mkdir -p "$workspace/src" + printf 'x\\n' > "$workspace/src/marker.txt" + # Abnormal termination / disconnect: cannot prove absence of side effects. + kill -KILL $$ + """ + ), + ) + _write_fake_worker( + self.bindir, + "next", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "unknown": {"adapter": "custom", "executable": "unknown"}, + "next": {"adapter": "custom", "executable": "next"}, + }, + role_order=["unknown", "next"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "unknown.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_BLOCKED, proc.stderr + proc.stdout) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(len(data["attempts"]), 1) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self.assertNotEqual(data["terminal_status"], "host-fallback-authorized") + self._assert_canonical_untouched() + + def test_scope_expansion_preserves_workspace_and_blocks(self) -> None: + launched = self.root / "scope-next.log" + launched.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "scope", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + printf 'bad\\n' > "$workspace/README.md" + printf '%s\\n' '{"terminal_status":"completed","summary":"expanded","changed_files":["README.md"],"evidence":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + _write_fake_worker( + self.bindir, + "next", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "scope": {"adapter": "custom", "executable": "scope"}, + "next": {"adapter": "custom", "executable": "next"}, + }, + role_order=["scope", "next"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "scope.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_SCOPE_EXPANSION, proc.stderr + proc.stdout + ) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "scope-expansion") + self.assertEqual(len(data["attempts"]), 1) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + attempt = data["attempts"][0] + self.assertTrue(Path(attempt["worktree"]).exists()) + self.assertTrue(attempt.get("workspace_preserved")) + self._assert_canonical_untouched() + + def test_worker_commit_or_head_move_is_rejected(self) -> None: + """Worker HEAD/commit, staged index, and config/ref mutations are fail-stop.""" + cases = { + "commit": textwrap.dedent( + """\ + workspace="$1" + mkdir -p "$workspace/src" + printf 'c\\n' > "$workspace/src/marker.txt" + git -C "$workspace" add src/marker.txt + git -C "$workspace" -c user.email=t@e -c user.name=t commit -m 'worker commit' + exit 0 + """ + ), + "staged": textwrap.dedent( + """\ + workspace="$1" + mkdir -p "$workspace/src" + printf 'staged\\n' > "$workspace/src/marker.txt" + git -C "$workspace" add src/marker.txt + exit 0 + """ + ), + "config_ref": textwrap.dedent( + """\ + workspace="$1" + git -C "$workspace" config --local agentexec.probe 1 + git -C "$workspace" update-ref refs/agent-exec-probe/tmp HEAD + exit 0 + """ + ), + } + for label, body in cases.items(): + with self.subTest(case=label): + launched = self.root / f"commit-next-{label}.log" + launched.write_text("", encoding="utf-8") + worker = f"committer-{label}" + _write_fake_worker(self.bindir, worker, body=body) + _write_fake_worker( + self.bindir, + f"next-{label}", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "committer": {"adapter": "custom", "executable": worker}, + "next": { + "adapter": "custom", + "executable": f"next-{label}", + }, + }, + role_order=["committer", "next"], + ) + self.base_snap = _canonical_snapshot(self.repo) + unit = _write_unit(self.root / f"unit-{label}.json") + receipt = self.receipts / f"commit-{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_BLOCKED, + proc.stderr + proc.stdout, + ) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(len(data["attempts"]), 1) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self.assertTrue(data["attempts"][0].get("workspace_preserved")) + self._assert_canonical_untouched() + + def test_untracked_and_binary_files_enter_transport_inventory(self) -> None: + _write_fake_worker( + self.bindir, + "binwriter", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + mkdir -p "$workspace/src" + printf 'text\\n' > "$workspace/src/marker.txt" + printf 'bin\\0data' > "$workspace/src/blob.bin" + printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt","src/blob.bin"],"evidence":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"binwriter": {"adapter": "custom", "executable": "binwriter"}}, + role_order=["binwriter"], + ) + unit = _write_unit( + self.root / "unit.json", + allowed_paths=["src/marker.txt", "src/blob.bin"], + ) + receipt = self.receipts / "binary.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")) + changed = set(data["changed_paths"]) + self.assertIn("src/marker.txt", changed) + self.assertIn("src/blob.bin", changed) + transport = data["transport"] + self.assertTrue(transport["ref"].startswith("refs/agent-exec/")) + self._assert_canonical_untouched() + + def test_timeout_kills_process_group_and_records_receipt(self) -> None: + child_pid_file = self.root / "child.pid" + _write_fake_worker( + self.bindir, + "sleeper", + body=textwrap.dedent( + f"""\ + # Start a child in the same process group + (sleep 120 & echo $! > "{child_pid_file}"; wait) + """ + ), + ) + self._configure( + profiles={"sleeper": {"adapter": "custom", "executable": "sleeper"}}, + role_order=["sleeper"], + timeout_seconds=2, + ) + unit = _write_unit(self.root / "unit.json", timeout_seconds=2) + receipt = self.receipts / "timeout.json" + t0 = time.time() + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + elapsed = time.time() - t0 + self.assertLess(elapsed, 15) + self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertTrue(data["attempts"][0].get("timed_out")) + if child_pid_file.exists(): + child = int(child_pid_file.read_text(encoding="utf-8").strip()) + # Child must not still be alive. + still = subprocess.run( + ["kill", "-0", str(child)], + check=False, + capture_output=True, + ) + self.assertNotEqual(still.returncode, 0) + self._assert_canonical_untouched() + + def test_summary_and_raw_log_caps_are_enforced(self) -> None: + _write_fake_worker( + self.bindir, + "noisy", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + python3 - <<'PY' + import sys + chunk = b"A" * (512 * 1024) + sys.stdout.buffer.write(chunk) + sys.stdout.buffer.flush() + sys.stderr.buffer.write(b"B" * (512 * 1024)) + sys.stderr.buffer.flush() + PY + mkdir -p "$workspace/src" + printf 'ok\\n' > "$workspace/src/marker.txt" + printf '%s\\n' '{"terminal_status":"completed","summary":"'"$(python3 -c 'print("S"*2000)')"'","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"noisy": {"adapter": "custom", "executable": "noisy"}}, + role_order=["noisy"], + summary_bytes=64, + raw_output_bytes=128, + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "caps.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")) + summary = data["summary"] + self.assertLessEqual(len(summary.encode("utf-8")), 64) + attempt = data["attempts"][0] + raw_path = Path(attempt["raw_log"]) + self.assertTrue(raw_path.is_file()) + self.assertLessEqual(raw_path.stat().st_size, 128) + cleanup = attempt.get("cleanup") or {} + self.assertLessEqual( + int(cleanup.get("stdout_tail_bytes", agent_exec.OUTPUT_TAIL_BYTES)), + agent_exec.OUTPUT_TAIL_BYTES, + ) + self.assertLessEqual( + int(cleanup.get("stderr_tail_bytes", agent_exec.OUTPUT_TAIL_BYTES)), + agent_exec.OUTPUT_TAIL_BYTES, + ) + self._assert_canonical_untouched() + + def test_receipt_redacts_environment_secrets(self) -> None: + secret = "sk-testdeadbeefdeadbeef123456" + _write_fake_worker( + self.bindir, + "leaky", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + echo "token=$AGENT_EXEC_TEST_SECRET" + 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":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"leaky": {"adapter": "custom", "executable": "leaky"}}, + role_order=["leaky"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "redact.json" + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + env={"AGENT_EXEC_TEST_SECRET": secret}, + ) + self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) + text = receipt.read_text(encoding="utf-8") + self.assertNotIn(secret, text) + data = json.loads(text) + self.assertIn("***", json.dumps(data.get("attempts", [{}])[0].get("summary", "***"))) + self._assert_canonical_untouched() + + def test_success_creates_base_parented_transport_without_touching_canonical_head( + self, + ) -> None: + _write_fake_worker( + self.bindir, + "ok", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + mkdir -p "$workspace/src" + printf 'shipped\\n' > "$workspace/src/marker.txt" + printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"ok": {"adapter": "custom", "executable": "ok"}}, + role_order=["ok"], + ) + unit = _write_unit(self.root / "unit.json") + receipt = self.receipts / "transport.json" + before = _canonical_snapshot(self.repo) + 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) + after = _canonical_snapshot(self.repo) + self.assertEqual(before, after) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "success") + transport = data["transport"] + ref = transport["ref"] + self.assertRegex(ref, r"^refs/agent-exec/[^/]+/[^/]+$") + parent = _git(self.repo, "rev-parse", f"{ref}^").stdout.strip() + self.assertEqual(parent, before["head"]) + commit = _git(self.repo, "rev-parse", ref).stdout.strip() + self.assertNotEqual(commit, before["head"]) + # Canonical branch HEAD unchanged + branch_head = _git(self.repo, "rev-parse", "HEAD").stdout.strip() + self.assertEqual(branch_head, before["head"]) + attempt = data["attempts"][0] + self.assertTrue(attempt.get("workspace_preserved")) + cleanup = attempt.get("cleanup") or {} + self.assertFalse(cleanup.get("exact_restore")) + self.assertTrue(cleanup.get("transport_created")) + + def test_canonical_checkout_mutation_is_failed_unknown(self) -> None: + """Worker mutating canonical files/index while keeping worktree clean is blocked.""" + launched = self.root / "canon-next.log" + launched.write_text("", encoding="utf-8") + # Resolve the main checkout from the linked worktree .git pointer. + _write_fake_worker( + self.bindir, + "canonmut", + body=textwrap.dedent( + """\ + workspace="$1" + common=$(git -C "$workspace" rev-parse --path-format=absolute --git-common-dir) + canonical=$(dirname "$common") + printf 'canonical-mutated\\n' >> "$canonical/README.md" + git -C "$canonical" add README.md + # Detached worktree left clean on purpose. + exit 0 + """ + ), + ) + _write_fake_worker( + self.bindir, + "canonnext", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "canonmut": {"adapter": "custom", "executable": "canonmut"}, + "canonnext": {"adapter": "custom", "executable": "canonnext"}, + }, + role_order=["canonmut", "canonnext"], + ) + unit = _write_unit(self.root / "unit-canon.json") + receipt = self.receipts / "canon.json" + before = _canonical_snapshot(self.repo) + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(len(data["attempts"]), 1) + attempt = data["attempts"][0] + self.assertTrue(attempt.get("workspace_preserved")) + self.assertIn("canonical", (attempt.get("reason") or "").lower()) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self.assertEqual(data.get("host_fallback_authorizations", 0), 0) + # Mutation is retained for reconciliation; canonical is intentionally dirty. + after = _canonical_snapshot(self.repo) + self.assertNotEqual(after["status"], before["status"]) + + def test_corrupt_worktree_git_pointer_yields_failed_unknown_receipt(self) -> None: + """Corrupted worktree .git pointer must produce a stable blocked receipt.""" + _write_fake_worker( + self.bindir, + "corruptgit", + body=textwrap.dedent( + """\ + workspace="$1" + printf 'not-a-valid-gitdir\\n' > "$workspace/.git" + exit 0 + """ + ), + ) + self._configure( + profiles={"corruptgit": {"adapter": "custom", "executable": "corruptgit"}}, + role_order=["corruptgit"], + ) + unit = _write_unit(self.root / "unit-corrupt.json") + receipt = self.receipts / "corrupt.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_BLOCKED, proc.stderr + proc.stdout) + self.assertTrue(receipt.is_file()) + data = json.loads(receipt.read_text(encoding="utf-8")) + 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.assertTrue(attempt.get("workspace_preserved")) + self.assertFalse((attempt.get("cleanup") or {}).get("exact_restore")) + reason = (attempt.get("reason") or "").lower() + self.assertTrue( + "inspection" in reason or "pointer" in reason or ".git" in reason, + attempt.get("reason"), + ) + self.assertNotIn("Traceback", proc.stderr) + self._assert_canonical_untouched() + + def test_networked_nonzero_no_file_change_is_failed_unknown(self) -> None: + """network=true nonzero exit with no file changes cannot be failed-clean.""" + launched = self.root / "net-next.log" + launched.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "netfail", + body=textwrap.dedent( + """\ + # No worktree writes; nonzero exit after possible external contact. + exit 11 + """ + ), + ) + _write_fake_worker( + self.bindir, + "netnext", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "netfail": {"adapter": "custom", "executable": "netfail"}, + "netnext": {"adapter": "custom", "executable": "netnext"}, + }, + role_order=["netfail", "netnext"], + ) + unit = _write_unit(self.root / "unit-net.json") + packet = json.loads(unit.read_text(encoding="utf-8")) + packet["network"] = True + packet["external_cost"] = False + unit.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8") + receipt = self.receipts / "netfail.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_BLOCKED, proc.stderr + proc.stdout) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(len(data["attempts"]), 1) + self.assertEqual(data["attempts"][0]["terminal_status"], "failed-unknown") + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self.assertNotEqual(data["terminal_status"], "host-fallback-authorized") + self.assertEqual(data.get("host_fallback_authorizations", 0), 0) + self._assert_canonical_untouched() + + def test_git_admin_staged_config_ref_mutation_is_failed_unknown(self) -> None: + """Staged index plus local config/ref mutation preserves workspace and stops.""" + launched = self.root / "admin-next.log" + launched.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "adminmut", + body=textwrap.dedent( + """\ + workspace="$1" + mkdir -p "$workspace/src" + printf 'idx\\n' > "$workspace/src/marker.txt" + git -C "$workspace" add src/marker.txt + git -C "$workspace" config --local agentexec.mutation yes + git -C "$workspace" update-ref refs/temporary/agent-exec-admin HEAD + exit 0 + """ + ), + ) + _write_fake_worker( + self.bindir, + "adminnext", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "adminmut": {"adapter": "custom", "executable": "adminmut"}, + "adminnext": {"adapter": "custom", "executable": "adminnext"}, + }, + role_order=["adminmut", "adminnext"], + ) + unit = _write_unit(self.root / "unit-admin.json") + receipt = self.receipts / "admin.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_BLOCKED, proc.stderr + proc.stdout) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(len(data["attempts"]), 1) + attempt = data["attempts"][0] + self.assertTrue(attempt.get("workspace_preserved")) + inventory = attempt.get("inventory_paths") or attempt.get("changed_paths") + self.assertIn("src/marker.txt", inventory) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self.assertEqual(data.get("host_fallback_authorizations", 0), 0) + self._assert_canonical_untouched() + + def test_symlink_allowed_path_escape_rejects_without_external_write(self) -> None: + """Allowed path through symlink escaping the worktree cannot touch outside.""" + outside = self.root / "outside-target" + outside.mkdir() + victim = outside / "victim.txt" + victim.write_text("untouched\n", encoding="utf-8") + link = self.repo / "src" / "escape" + link.symlink_to(outside, target_is_directory=True) + _git(self.repo, "add", "src/escape") + _git(self.repo, "commit", "-m", "add escape symlink") + self.base_snap = _canonical_snapshot(self.repo) + + launched = self.root / "symlink-next.log" + launched.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "symworker", + body=textwrap.dedent( + """\ + workspace="$1" + # If launch were allowed, attempt to write through the symlink. + printf 'pwned\\n' > "$workspace/src/escape/victim.txt" + exit 0 + """ + ), + ) + _write_fake_worker( + self.bindir, + "symnext", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "symworker": {"adapter": "custom", "executable": "symworker"}, + "symnext": {"adapter": "custom", "executable": "symnext"}, + }, + role_order=["symworker", "symnext"], + ) + unit = _write_unit( + self.root / "unit-symlink.json", + allowed_paths=["src/escape/victim.txt"], + ) + receipt = self.receipts / "symlink.json" + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + self.assertNotEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) + self.assertNotEqual(proc.returncode, agent_exec.EXIT_HOST_FALLBACK) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertNotEqual(data["terminal_status"], "success") + self.assertNotEqual(data["terminal_status"], "host-fallback-authorized") + self.assertEqual(victim.read_text(encoding="utf-8"), "untouched\n") + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self._assert_canonical_untouched() + + def test_secret_crossing_raw_log_flush_boundary_is_redacted(self) -> None: + """Exact env secrets must not leak when split across raw-log drain/flush edges.""" + secret = "SECRETBOUNDARYTOKEN" + _write_fake_worker( + self.bindir, + "boundary", + body=textwrap.dedent( + f"""\ + workspace="$1" + result="$3" + python3 - <<'PY' + import os, sys + secret = os.environ["AGENT_EXEC_TEST_SECRET"] + sys.stdout.buffer.write(b"A" * 120 + secret.encode() + b"Z" * 60) + sys.stdout.buffer.flush() + PY + 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":[],"scope_expansion":null}}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"boundary": {"adapter": "custom", "executable": "boundary"}}, + role_order=["boundary"], + raw_output_bytes=4096, + ) + unit = _write_unit(self.root / "unit-boundary.json") + receipt = self.receipts / "boundary-redact.json" + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + env={"AGENT_EXEC_TEST_SECRET": secret}, + ) + self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) + receipt_text = receipt.read_text(encoding="utf-8") + self.assertNotIn(secret, receipt_text) + data = json.loads(receipt_text) + raw_path = Path(data["attempts"][0]["raw_log"]) + raw_bytes = raw_path.read_bytes() + self.assertNotIn(secret.encode("utf-8"), raw_bytes) + self.assertNotIn(secret, raw_bytes.decode("utf-8", "replace")) + self._assert_canonical_untouched() + + def test_large_stdin_with_early_stdout_does_not_deadlock(self) -> None: + """Stdin must be select-driven so early stdout beyond pipe capacity cannot hang.""" + # Stay under adapter MAX_UNIT_PACKET_BYTES (262144) while exceeding pipe capacity. + pad = "P" * (128 * 1024) + _write_fake_worker( + self.bindir, + "pipefill", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + # Use -c so process stdin (unit packet) remains readable by Python. + python3 -c 'import sys; sys.stdout.buffer.write(b"X"*(256*1024)); sys.stdout.buffer.flush(); sys.stdin.buffer.read()' + 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":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"pipefill": {"adapter": "custom", "executable": "pipefill"}}, + role_order=["pipefill"], + timeout_seconds=15, + raw_output_bytes=1024, + ) + unit = _write_unit( + self.root / "unit-pipefill.json", + timeout_seconds=15, + ) + # Inflate the stdin-delivered unit packet well beyond pipe capacity. + packet = json.loads(unit.read_text(encoding="utf-8")) + packet["constraints"] = ["leave uncommitted", pad] + unit.write_text(json.dumps(packet) + "\n", encoding="utf-8") + self.assertGreater(unit.stat().st_size, 65536) + self.assertLessEqual(unit.stat().st_size, 262144) + receipt = self.receipts / "pipefill.json" + started = time.monotonic() + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + elapsed = time.monotonic() - started + self.assertLessEqual(elapsed, 20.0, f"controller hung for {elapsed:.1f}s") + self.assertIn( + proc.returncode, + { + agent_exec.EXIT_SUCCESS, + agent_exec.EXIT_BLOCKED, + }, + proc.stderr + proc.stdout, + ) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertIn( + data["terminal_status"], + {"success", "failed-unknown", "blocked"}, + ) + if data["terminal_status"] == "success": + self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS) + self._assert_canonical_untouched() + + def test_transport_construction_failure_is_failed_unknown(self) -> None: + """Host transport staging/commit-tree/update-ref failure stays recoverable.""" + launched = self.root / "transport-next.log" + launched.write_text("", encoding="utf-8") + _write_fake_worker( + self.bindir, + "txok", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + 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":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + _write_fake_worker( + self.bindir, + "txnext", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "txok": {"adapter": "custom", "executable": "txok"}, + "txnext": {"adapter": "custom", "executable": "txnext"}, + }, + role_order=["txok", "txnext"], + ) + unit = _write_unit(self.root / "unit-txfail.json") + receipt = self.receipts / "txfail.json" + with mock.patch.object( + agent_exec, + "_create_transport_commit", + side_effect=subprocess.CalledProcessError( + 1, ["git", "commit-tree"], stderr="simulated transport failure" + ), + ): + 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") + self.assertEqual(len(data["attempts"]), 1) + attempt = data["attempts"][0] + self.assertEqual(attempt["terminal_status"], "failed-unknown") + self.assertTrue(attempt.get("workspace_preserved")) + cleanup = attempt.get("cleanup") or {} + self.assertFalse(cleanup.get("transport_created", True)) + self.assertIn("transport construction failed", attempt.get("reason") or data.get("reason") or "") + self.assertEqual(data.get("host_fallback_authorizations", 0), 0) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + self._assert_canonical_untouched() + + +if __name__ == "__main__": + unittest.main()