"""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 ( 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 # 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) 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: 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) 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 in {"unavailable", "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 break 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) * max_attempts ) 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), } ) 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 # 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], 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 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 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: """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: """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] if __name__ == "__main__": sys.exit(main())