Agent-System-Dev/scripts/agent_exec.py

"""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


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

    for profile_name in profiles:
        if stop:
            break
        profile = cfg.profiles[profile_name]
        attempt_id = uuid.uuid4().hex[:12]
        attempt = _run_attempt(
            repo=repo,
            cfg=cfg,
            profile=profile,
            unit=unit,
            unit_path=unit_path,
            base=base,
            run_id=run_id,
            attempt_id=attempt_id,
            run_dir=run_dir,
            secret_values=secret_values,
        )
        attempts.append(attempt)

        if attempt.terminal_status == "success":
            terminal = "success"
            reason = None
            changed_paths = list(attempt.changed_paths)
            transport = {
                "ref": attempt.transport_ref,
                "parent": base["head"],
                "attempt_id": attempt_id,
            }
            stop = True
            break
        if attempt.terminal_status == "unavailable":
            terminal = "unavailable"
            reason = attempt.reason
            continue
        if attempt.terminal_status == "failed-clean":
            terminal = "unavailable"
            reason = attempt.reason
            continue
        if attempt.terminal_status == "scope-expansion":
            terminal = "scope-expansion"
            reason = attempt.reason
            changed_paths = list(attempt.changed_paths)
            stop = True
            break
        if attempt.terminal_status in {"failed-unknown", "blocked"}:
            terminal = attempt.terminal_status
            reason = attempt.reason
            stop = True
            break
        if attempt.terminal_status == "cancelled":
            terminal = "cancelled"
            reason = attempt.reason
            stop = True
            break

    exhausted_ok = (
        not stop
        and attempts
        and all(a.terminal_status in {"unavailable", "failed-clean"} for a in attempts)
        and len(attempts) == len(profiles)
    )
    if exhausted_ok and cfg.execution.fallback_to_host == "all-unavailable":
        terminal = "host-fallback-authorized"
        host_fallback_authorizations = 1
        host_scope = {
            "allowed_paths": list(unit.allowed_paths),
            "unit_id": unit.unit_id,
            "goal": unit.goal,
            "verification": list(unit.verification),
            "constraints": list(unit.constraints),
            "network": unit.network,
            "external_cost": unit.external_cost,
        }
        reason = "all configured profiles exhausted with unavailable or failed-clean"

    summary = _bounded_summary(
        _build_run_summary(terminal, attempts, reason),
        cfg.execution.summary_bytes,
        secret_values,
    )
    receipt = _base_receipt(
        run_id=run_id,
        repo=repo,
        role=role,
        unit_id=unit.unit_id,
        config=cfg,
        terminal_status=terminal,
        reason=reason,
        attempts=attempts,
        secret_values=secret_values,
    )
    receipt.update(
        {
            "unit_digest": unit.digest,
            "config_digest": config_digest,
            "base": {"head": base["head"], "tree": base["tree"], "branch": base["branch"]},
            "summary": summary,
            "changed_paths": changed_paths,
            "transport": transport,
            "host_scope": host_scope,
            "host_fallback_authorizations": host_fallback_authorizations,
            "run_dir": str(run_dir),
            "private_root_mode": oct(run_dir.parent.stat().st_mode & 0o777),
        }
    )
    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/<run>/<attempt>`` 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,
                "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"),
                "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())