"""Recoverable agent-exec controller: validate, run, apply transports, admit provenance.""" 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 ( DEFAULT_IDLE_TIMEOUT_SECONDS, SECRET_VALUE_RE, ConfigError, ExecutionConfig, Profile, load_config, validate_config, ) RECEIPT_SCHEMA = "agent-exec-receipt/v1" HISTORY_SCHEMA = "agent-exec-history/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 _EMERGENCY_ROLES = frozenset({"implementation", "repair", "test"}) _SAFE_EXHAUST_STATUSES = frozenset({"unavailable", "failed-clean"}) # Transport commits must not depend on machine-global Git identity. TRANSPORT_AUTHOR_NAME = "agent-exec-transport" TRANSPORT_AUTHOR_EMAIL = "agent-exec-transport@localhost.invalid" 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 idle_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 attempt_ordinal: int = 1 emergency_fallback: bool = False 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) p_apply = sub.add_parser( "apply", help="Fast-forward the current branch to a success transport", ) p_apply.add_argument("--repo", required=True) p_apply.add_argument("--receipt", required=True) p_admit = sub.add_parser( "admit", help="Admit staged paths backed by a success transport or host exemption", ) p_admit.add_argument("--repo", 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), ) if args.command == "apply": return cmd_apply(Path(args.repo), Path(args.receipt)) if args.command == "admit": return cmd_admit(Path(args.repo)) 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 _apply_attempt_route( attempt: AttemptRecord, *, attempt_id: str, parent: str, terminal: str, reason: str | None, changed_paths: list[str], transport: dict[str, Any] | None, ) -> tuple[str, str, str | None, list[str], dict[str, Any] | None]: status = attempt.terminal_status if status == "success": return ( "stop", "success", None, list(attempt.changed_paths), { "ref": attempt.transport_ref, "parent": parent, "attempt_id": attempt_id, }, ) if status in _SAFE_EXHAUST_STATUSES: return ("continue", "unavailable", attempt.reason, changed_paths, transport) if status == "scope-expansion": return ( "stop", "scope-expansion", attempt.reason, list(attempt.changed_paths), transport, ) if status in {"failed-unknown", "blocked"}: return ("stop", status, attempt.reason, changed_paths, transport) if status == "cancelled": return ("stop", "cancelled", attempt.reason, changed_paths, transport) return ("next-profile", terminal, reason, changed_paths, transport) def cmd_run( repo: Path, *, role: str, unit_path: Path, receipt_path: Path, ) -> int: started_monotonic = time.monotonic() 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: receipt = _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, ) return _finish_run( repo=repo, receipt_path=receipt_path, receipt=receipt, secret_values=secret_values, summary_bytes=256, exit_code=EXIT_INVALID_CONFIG, started_monotonic=started_monotonic, ) 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: receipt = _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, ) return _finish_run( repo=repo, receipt_path=receipt_path, receipt=receipt, secret_values=secret_values, summary_bytes=cfg.execution.summary_bytes, exit_code=EXIT_INVALID_CONFIG, started_monotonic=started_monotonic, ) try: _validate_allowed_path_symlinks(repo, unit.allowed_paths) except UnitPacketError as exc: # Symlink escape is a safety failure: stop routing, never authorize fallback. receipt = _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, ) return _finish_run( repo=repo, receipt_path=receipt_path, receipt=receipt, secret_values=secret_values, summary_bytes=cfg.execution.summary_bytes, exit_code=EXIT_BLOCKED, started_monotonic=started_monotonic, ) 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 max_attempts = cfg.execution.max_attempts_per_profile for profile_name in profiles: if stop: break profile = cfg.profiles[profile_name] for attempt_ordinal in range(1, max_attempts + 1): 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, ) attempt.attempt_ordinal = attempt_ordinal attempt.emergency_fallback = False attempts.append(attempt) action, terminal, reason, changed_paths, transport = _apply_attempt_route( attempt, attempt_id=attempt_id, parent=base["head"], terminal=terminal, reason=reason, changed_paths=changed_paths, transport=transport, ) if action == "continue": continue stop = action == "stop" break exhausted_ok = ( not stop and attempts and all(a.terminal_status in _SAFE_EXHAUST_STATUSES for a in attempts) and len(attempts) == len(profiles) * max_attempts ) emergency_name = cfg.execution.emergency_profile if exhausted_ok and emergency_name and role in _EMERGENCY_ROLES: emergency_profile = cfg.profiles.get(emergency_name) if emergency_profile is None: terminal = "blocked" reason = f"emergency profile {emergency_name!r} is not configured" else: attempt_id = uuid.uuid4().hex[:12] attempt = _run_attempt( repo=repo, cfg=cfg, profile=emergency_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, ) attempt.attempt_ordinal = 1 attempt.emergency_fallback = True attempts.append(attempt) action, terminal, reason, changed_paths, transport = _apply_attempt_route( attempt, attempt_id=attempt_id, parent=base["head"], terminal=terminal, reason=reason, changed_paths=changed_paths, transport=transport, ) if action == "continue": terminal = "blocked" reason = attempt.reason stop = True if ( exhausted_ok and not stop and not emergency_name 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), } ) return _finish_run( repo=repo, receipt_path=receipt_path, receipt=receipt, secret_values=secret_values, summary_bytes=cfg.execution.summary_bytes, exit_code=_exit_for_terminal(terminal), started_monotonic=started_monotonic, ) 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 _create_private_attempt_clone(repo: Path, dest: Path, base_head: str) -> None: """Create a private local shared clone with isolated Git administration. Objects are reused read-only via ``git clone --shared`` alternates. The clone owns its own ``.git`` config, refs, index, HEAD, and worktree. Never contacts a remote. """ if dest.exists(): shutil.rmtree(dest) proc = subprocess.run( [ "git", "clone", "--shared", "--no-checkout", "--", str(repo.resolve()), str(dest), ], check=False, capture_output=True, text=True, ) if proc.returncode != 0: raise subprocess.CalledProcessError( proc.returncode, proc.args, proc.stdout, proc.stderr ) # Drop the auto-configured origin so the attempt cannot reach a remote URL. _git(dest, "remote", "remove", "origin", check=False) _git(dest, "checkout", "--detach", base_head) 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: _create_private_attempt_clone(repo, worktree, base["head"]) except subprocess.CalledProcessError as exc: err = (exc.stderr or str(exc)).strip() return AttemptRecord( attempt_id=attempt_id, profile=profile.name, adapter=profile.adapter, terminal_status="unavailable", reason=f"private clone create failed: {err}", cleanup={"exact_restore": True, "skipped": True}, observed_version=probe.observed_version, ) except OSError as exc: return AttemptRecord( attempt_id=attempt_id, profile=profile.name, adapter=profile.adapter, terminal_status="unavailable", reason=f"private clone create failed: {exc}", 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) idle_timeout = min(cfg.execution.idle_timeout_seconds, timeout) proc_meta = _supervise( invocation.argv, cwd=invocation.cwd, stdin=invocation.stdin, timeout=timeout, idle_timeout=idle_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"], idle_timed_out=bool(proc_meta.get("idle_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"], idle_timed_out=bool(proc_meta.get("idle_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.get("idle_timed_out") and not proc_meta["timed_out"]: # Idle expiry is not a proven-clean failure and must not continue routing. return _preserved( "failed-unknown", "idle timeout; process group terminated", cleanup_extra={ "reason": "idle timeout", "idle_timed_out": True, "timed_out": False, }, ) 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 # Worker fail-stop terminals dominate success and no-change classification. # A zero exit or an in-scope diff must not launder blocked/cancelled/unknown. worker_stop = _worker_fail_stop_status(worker_result) if worker_stop: return _preserved( worker_stop, f"worker terminal_status {worker_stop}", cleanup_extra={"reason": "worker fail-stop terminal"}, ) # 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: # 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, }, 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], idle_timeout: int | None = None, ) -> 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 idle_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, "idle_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} bounded_timeout = max(int(timeout), 1) if idle_timeout is None: idle_limit = min(DEFAULT_IDLE_TIMEOUT_SECONDS, bounded_timeout) else: idle_limit = min(max(int(idle_timeout), 1), bounded_timeout) started = time.monotonic() deadline = started + bounded_timeout idle_deadline = started + idle_limit grace_deadline: float | None = None with raw_log.open("wb") as raw_fh: def _note_progress() -> None: """Reset idle only after a successful stdin write or stdout/stderr read.""" nonlocal idle_deadline if grace_deadline is not None: return idle_deadline = time.monotonic() + idle_limit 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) _note_progress() 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 written > 0: _note_progress() 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 or now >= idle_deadline): # Total timeout stays the hard bound when both deadlines are due. if now >= deadline: timed_out = True else: idle_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, idle_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: if not idle_timed_out: 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, "idle_timed_out": idle_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 _WORKER_FAIL_STOP = frozenset({"blocked", "cancelled", "failed-unknown"}) def _worker_fail_stop_status(worker_result: dict[str, Any] | None) -> str | None: """Return a worker terminal that must fail-stop before success or no-change.""" if not isinstance(worker_result, dict): return None raw = worker_result.get("terminal_status") if not isinstance(raw, str): return None status = raw.strip().replace("_", "-") if status in _WORKER_FAIL_STOP: return status return None def _parse_name_status(text: str) -> list[tuple[str, tuple[str, ...]]]: """Parse git name-status text into ``(status, paths)`` records. Rename (``R``) and copy (``C``) records keep both the source and the destination. Keeping only the destination would hide deletion of an out-of-scope or non-exempt source that was moved into an allowed path. """ records: list[tuple[str, tuple[str, ...]]] = [] for line in text.splitlines(): if not line.strip(): continue parts = line.split("\t") status = parts[0] if status[:1] in {"R", "C"} and len(parts) >= 3: records.append((status, (parts[1], parts[2]))) elif len(parts) >= 2: records.append((status, (parts[1],))) return records 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 status, rec_paths in _parse_name_status(text): paths.extend(rec_paths) kind = status[:1] if kind in {"R", "C"} and len(rec_paths) == 2: label = "renamed" if kind == "R" else "copied" details.append( { "status": f"{prefix}{label}", "from": rec_paths[0], "path": rec_paths[1], } ) else: details.append({"status": f"{prefix}{kind}", "path": rec_paths[0]}) # 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 the private attempt clone's Git administration state. Canonical checkout identity is snapshotted separately. ``repo`` is unused now that each attempt owns private config/refs; kept for call-site stability. """ del repo 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(worktree, "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( worktree, "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)) abs_git = _git(worktree, "rev-parse", "--absolute-git-dir").stdout.strip() common_dir = _git( worktree, "rev-parse", "--path-format=absolute", "--git-common-dir" ).stdout.strip() return { "head": head, "index_tree": index_tree, "staged": staged, "local_config": local_config, "refs": refs, "abs_git": abs_git, "common_dir": common_dir, } 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 workspace). Protects HEAD, current branch, index, tracked/untracked status, and git/common-dir identity. Unrelated sibling branch refs, branch config, and worktree-list registration are intentionally excluded — worker Git is isolated in a private clone, so those host-side delivery signals are not worker mutation. """ 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() return { "head": head, "branch": branch, "index_tree": index_tree, "status": status, "git_dir": git_dir, "common_dir": common, } def _worktree_pointer_snapshot( worktree: Path, *, require_git: bool = True ) -> dict[str, str]: """Snapshot the private attempt `.git` directory / 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: """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()) 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: """Create a base-parented transport commit in the private clone and copy it back. Staging and ``commit-tree`` run only inside the attempt clone. Exact commit/tree/blob objects are fetched into the canonical repository, which receives solely the host-owned ``refs/agent-exec//`` ref. """ del inventory # Inventory already validated; staging uses the full private tree. # Stage all tracked/untracked changes in the private clone only. _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, "-c", f"user.name={TRANSPORT_AUTHOR_NAME}", "-c", f"user.email={TRANSPORT_AUTHOR_EMAIL}", "commit-tree", tree, "-p", base_head, "-m", msg, ).stdout.strip() ref = f"refs/agent-exec/{run_id}/{attempt_id}" # Park the commit on a private ref so a local fetch can name it. _git(worktree, "update-ref", ref, commit) # Copy objects into the canonical store and create only the host-owned ref. fetch = _git( repo, "fetch", "--no-tags", "--no-write-fetch-head", str(worktree.resolve()), f"+{ref}:{ref}", check=False, ) if fetch.returncode != 0: raise RuntimeError( f"transport object transfer failed: {fetch.stderr.strip() or fetch.stdout.strip()}" ) resolved = _git(repo, "rev-parse", ref).stdout.strip() if resolved != commit: raise RuntimeError( f"transport ref resolved to {resolved}, expected {commit}" ) parents = ( _git(repo, "rev-list", "--parents", "-n", "1", commit) .stdout.strip() .split() ) if len(parents) != 2 or parents[0] != commit or parents[1] != base_head: raise RuntimeError( f"transport commit parent check failed: {' '.join(parents)}" ) 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. 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, "attempt_ordinal": a.attempt_ordinal, "emergency_fallback": a.emergency_fallback, "terminal_status": a.terminal_status, "reason": a.reason, "worktree": a.worktree, "pid": a.pid, "exit_code": a.exit_code, "timed_out": a.timed_out, "idle_timed_out": a.idle_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, "idle_timeout_seconds": config.execution.idle_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 _finish_run( *, repo: Path, receipt_path: Path, receipt: dict[str, Any], secret_values: set[str], summary_bytes: int, exit_code: int, started_monotonic: float, ) -> int: """Persist one history line and the normal receipt without changing outcome.""" receipt["elapsed_seconds"] = round( max(0.0, time.monotonic() - started_monotonic), 3 ) history_path: Path | None = None try: history_path = _append_execution_history(repo, receipt, secret_values) receipt["history"] = { "path": str(history_path), "status": "recorded", } except Exception as exc: receipt["history"] = { "path": str(history_path) if history_path is not None else None, "status": "failed", "reason": _redact_text(str(exc), secret_values), } _write_receipt( receipt_path, receipt, secret_values=secret_values, summary_bytes=summary_bytes, ) return exit_code def _append_execution_history( repo: Path, receipt: dict[str, Any], secret_values: set[str], ) -> Path: common = _git( repo.resolve(), "rev-parse", "--path-format=absolute", "--git-common-dir", check=False, ) if common.returncode != 0 or not common.stdout.strip(): raise RuntimeError( "unable to resolve git common dir: " + (common.stderr.strip() or common.stdout.strip()) ) common_dir = Path(common.stdout.strip()) if not common_dir.is_absolute(): common_dir = (repo.resolve() / common_dir).resolve() history_dir = common_dir / "agent-exec" if history_dir.is_symlink(): raise RuntimeError("agent-exec history directory must not be a symlink") history_dir.mkdir(mode=0o700, parents=True, exist_ok=True) os.chmod(history_dir, 0o700) history_path = history_dir / "execution-history.jsonl" record = _execution_history_record(repo, receipt) line = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")) line = _redact_text(line, secret_values) + "\n" fd = os.open(history_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "a", encoding="utf-8", closefd=False) as handle: fcntl.flock(fd, fcntl.LOCK_EX) try: handle.write(line) handle.flush() os.fsync(fd) finally: fcntl.flock(fd, fcntl.LOCK_UN) finally: os.close(fd) return history_path def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str, Any]: config = receipt.get("config_summary") profiles = config.get("profiles", {}) if isinstance(config, dict) else {} attempt_records: list[dict[str, Any]] = [] for attempt in receipt.get("attempts", []): if not isinstance(attempt, dict): continue profile_name = attempt.get("profile") profile = profiles.get(profile_name, {}) if isinstance(profiles, dict) else {} requested = profile.get("model") if isinstance(profile, dict) else None attempt_records.append( { "attempt_id": attempt.get("attempt_id"), "profile": profile_name, "adapter": attempt.get("adapter"), "attempt_ordinal": attempt.get("attempt_ordinal"), "emergency_fallback": bool(attempt.get("emergency_fallback")), "observed_version": attempt.get("observed_version"), "model_requested": requested or "auto", "model_observed": "not-observable", "fresh_session": True, "session_id": "not-observable", "terminal_status": attempt.get("terminal_status"), "exit_code": attempt.get("exit_code"), "timed_out": bool(attempt.get("timed_out")), "idle_timed_out": bool(attempt.get("idle_timed_out")), "changed_paths": list(attempt.get("changed_paths") or []), "transport_ref": attempt.get("transport_ref"), } ) transport = receipt.get("transport") safe_transport = None if isinstance(transport, dict): safe_transport = { "ref": transport.get("ref"), "parent": transport.get("parent"), } terminal = receipt.get("terminal_status") return { "schema": HISTORY_SCHEMA, "recorded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "record_id": receipt.get("run_id"), "run_id": receipt.get("run_id"), "repo": str(repo.resolve()), "unit_id": receipt.get("unit_id"), "role": receipt.get("role"), "stage": receipt.get("role"), "config_digest": receipt.get("config_digest"), "elapsed_seconds": receipt.get("elapsed_seconds"), "terminal_status": terminal, "reason": receipt.get("reason"), "failure_category": None if terminal == "success" else terminal, "changed_paths": list(receipt.get("changed_paths") or []), "transport": safe_transport, "host_fallback_authorizations": receipt.get( "host_fallback_authorizations", 0 ), "attempts": attempt_records, "verification_result": "NOT_RUN", "review_result": "NOT_RUN", "final_result": "NOT_RUN", "tokens": "not-observable", "cost": "not-observable", } 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] # Host may commit task cards and superpowers docs without a worker transport. _HOST_MANAGED_PREFIXES = ( ".task/", "docs/superpowers/specs/", "docs/superpowers/plans/", ) _EMBEDDED_SECRET_RE = re.compile(r"(?i)(?:bearer\s+\S+|sk-[A-Za-z0-9_\-]{8,})") def _reject(message: str, secrets: set[str], code: int = EXIT_BLOCKED) -> int: print(_redact_text(message, secrets), file=sys.stderr) return code def _secret_set(*texts: str) -> set[str]: found: set[str] = set() _collect_secret_values(found) for text in texts: found.update(_EMBEDDED_SECRET_RE.findall(text)) return found def _normalize_repo_path(path: str) -> str: """Normalize a repo-relative path without eating dotted names such as ``.task``.""" norm = path.replace("\\", "/").strip() while norm.startswith("./"): norm = norm[2:] return norm def _host_managed(path: str) -> bool: norm = _normalize_repo_path(path) if not norm or ".." in Path(norm).parts: return False return any(norm.startswith(prefix) for prefix in _HOST_MANAGED_PREFIXES) def _non_exempt(paths: list[str]) -> set[str]: return {_normalize_repo_path(path) for path in paths if not _host_managed(path)} def _same_repo(repo: Path, raw: object) -> bool: if not isinstance(raw, str) or not raw.strip(): return False candidate = Path(raw) if not candidate.is_absolute(): candidate = repo / candidate try: return candidate.resolve() == repo.resolve() except OSError: return False def _resolve_commit(repo: Path, rev: str) -> str | None: if not rev or rev.startswith("-") or "\n" in rev or "\x00" in rev: return None proc = _git(repo, "rev-parse", "--verify", "--quiet", f"{rev}^{{commit}}", check=False) if proc.returncode != 0: return None resolved = proc.stdout.strip() return resolved or None def _commit_parents(repo: Path, commit: str) -> list[str] | None: proc = _git(repo, "rev-list", "--parents", "-n", "1", commit, check=False) if proc.returncode != 0: return None parts = proc.stdout.split() if not parts or parts[0] != commit: return None return parts[1:] def _name_status_paths(text: str) -> list[str]: paths: list[str] = [] for _status, rec_paths in _parse_name_status(text): paths.extend(rec_paths) return paths def _diff_paths(repo: Path, parent: str, commit: str) -> list[str]: proc = _git( repo, "diff", "--name-status", "--find-renames", parent, commit, check=False, ) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or "transport diff failed") return _name_status_paths(proc.stdout) def _staged_paths(repo: Path) -> list[str]: proc = _git( repo, "diff", "--cached", "--name-status", "--find-renames", check=False, ) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or "staged diff failed") return _name_status_paths(proc.stdout) def _index_matches_commit(repo: Path, commit: str, paths: set[str]) -> bool: """True when each named index path has the same content as commit.""" if not paths: return True proc = _git( repo, "diff", "--cached", "--quiet", commit, "--", *sorted(paths), check=False, ) return proc.returncode == 0 def _valid_transport_ref(ref: object) -> str | None: if not isinstance(ref, str) or not ref.startswith("refs/agent-exec/"): return None if ".." in ref.split("/") or ref.endswith("/"): return None return ref def _history_file(repo: Path) -> Path | None: common = _git( repo, "rev-parse", "--path-format=absolute", "--git-common-dir", check=False, ) if common.returncode != 0 or not common.stdout.strip(): raise RuntimeError(common.stderr.strip() or "unable to resolve git common dir") common_dir = Path(common.stdout.strip()) if not common_dir.is_absolute(): common_dir = (repo / common_dir).resolve() history_dir = common_dir / "agent-exec" history_path = history_dir / "execution-history.jsonl" if history_dir.is_symlink() or history_path.is_symlink(): return None return history_path def _load_history_records(repo: Path) -> list[dict[str, Any]]: history_path = _history_file(repo) if history_path is None or not history_path.is_file(): return [] records: list[dict[str, Any]] = [] for line in history_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: data = json.loads(line) except json.JSONDecodeError: continue if isinstance(data, dict): records.append(data) return records def _transport_covers( repo: Path, record: dict[str, Any], *, head: str, required: set[str], ) -> bool: if record.get("schema") != HISTORY_SCHEMA: return False if record.get("terminal_status") != "success": return False if not _same_repo(repo, record.get("repo")): return False transport = record.get("transport") if not isinstance(transport, dict): return False ref = _valid_transport_ref(transport.get("ref")) parent = transport.get("parent") if ref is None or not isinstance(parent, str): return False if _resolve_commit(repo, parent) != head: return False commit = _resolve_commit(repo, ref) if commit is None or _commit_parents(repo, 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, commit) if _non_exempt(changed) != required or _non_exempt(diff_paths) != required: return False return _index_matches_commit(repo, commit, required) def cmd_apply(repo: Path, receipt_path: Path) -> int: """Fast-forward the clean current branch onto one success transport.""" repo = repo.resolve() text = "" try: text = receipt_path.read_text(encoding="utf-8") receipt = json.loads(text) except (OSError, UnicodeError, json.JSONDecodeError) as exc: return _reject( f"apply: invalid receipt: {exc}", _secret_set(text), EXIT_INVALID_CONFIG, ) secrets = _secret_set(text) if not isinstance(receipt, dict) or receipt.get("schema") != RECEIPT_SCHEMA: return _reject("apply: invalid receipt", secrets, EXIT_INVALID_CONFIG) if not _same_repo(repo, receipt.get("repo")): return _reject("apply: receipt repo mismatch", secrets) if receipt.get("terminal_status") != "success": message = "apply: non-success receipt" reason = receipt.get("reason") if isinstance(reason, str) and reason: message = f"{message}: {reason}" return _reject(message, secrets) base = receipt.get("base") base_head = base.get("head") if isinstance(base, dict) else None if not isinstance(base_head, str): return _reject("apply: stale HEAD", secrets) resolved_base = _resolve_commit(repo, base_head) head = _resolve_commit(repo, "HEAD") status = _git( repo, "status", "--porcelain", "--untracked-files=all", check=False ) if status.returncode != 0: return _reject( f"apply: dirty worktree: {status.stderr.strip()}", secrets, ) if status.stdout.strip(): return _reject("apply: dirty worktree", secrets) branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD", check=False) if branch.returncode != 0 or branch.stdout.strip() in {"", "HEAD"}: return _reject("apply: detached HEAD", secrets) if head is None or resolved_base is None or head != resolved_base: return _reject("apply: stale HEAD", secrets) transport = receipt.get("transport") if not isinstance(transport, dict): return _reject("apply: missing transport", secrets) ref = _valid_transport_ref(transport.get("ref")) parent = transport.get("parent") if ref is None or not isinstance(parent, str) or not parent: return _reject("apply: missing transport", secrets) commit = _resolve_commit(repo, ref) if commit is None: return _reject("apply: missing transport", secrets) if ( _resolve_commit(repo, parent) != resolved_base or _commit_parents(repo, commit) != [resolved_base] ): return _reject("apply: mismatched transport", secrets) merge = _git(repo, "merge", "--ff-only", "--quiet", commit, check=False) if merge.returncode != 0: return _reject( "apply: fast-forward failed: " + (merge.stderr.strip() or merge.stdout.strip()), secrets, ) print(f"apply: fast-forwarded {branch.stdout.strip()} {head}..{commit}") return EXIT_SUCCESS def cmd_admit(repo: Path) -> int: """Accept staged host-managed paths, or paths exactly covered by one transport.""" repo = repo.resolve() secrets = _secret_set() try: staged = _staged_paths(repo) required = _non_exempt(staged) if not required: print("admit: ok") return EXIT_SUCCESS head = _resolve_commit(repo, "HEAD") if head is None: return _reject("admit: no provenance", 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) for record in _load_history_records(repo): if _transport_covers(repo, record, head=head, required=required): print("admit: ok") return EXIT_SUCCESS except Exception as exc: return _reject(f"admit: no provenance: {exc}", secrets) listed = ", ".join(sorted(required)) return _reject(f"admit: no provenance for {listed}", secrets) if __name__ == "__main__": sys.exit(main())