"""End-to-end execution controller contract for agent-exec (U3).

U5 adds an opt-in real CLI probe. Default discovery does not contact providers.
Set RUN_REAL_AGENT_EXEC=1 and AGENT_EXEC_REAL_PROFILE to exactly one profile.
"""
from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import shlex
import shutil
import stat
import subprocess
import tempfile
import textwrap
import threading
import time
import unittest
import uuid
from pathlib import Path
from unittest import mock

from scripts import agent_exec
from scripts.agent_exec_config import NORMAL_ROLE_ADAPTERS


def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", "-C", str(repo), *args],
        check=check,
        capture_output=True,
        text=True,
    )


def _canonical_snapshot(repo: Path) -> dict[str, str]:
    head = _git(repo, "rev-parse", "HEAD").stdout.strip()
    tree = _git(repo, "rev-parse", "HEAD^{tree}").stdout.strip()
    status = _git(repo, "status", "--porcelain=v2", "--branch").stdout
    index = _git(repo, "write-tree").stdout.strip()
    return {"head": head, "tree": tree, "status": status, "index": index}


def _pid_alive(pid: int) -> bool:
    """Return true only for a running, non-zombie process."""
    try:
        text = Path(f"/proc/{pid}/stat").read_text(
            encoding="utf-8", errors="replace"
        )
    except FileNotFoundError:
        return False
    rpar = text.rfind(")")
    if rpar != -1 and rpar + 2 < len(text) and text[rpar + 2] == "Z":
        return False
    return (
        subprocess.run(
            ["kill", "-0", str(pid)], check=False, capture_output=True
        ).returncode
        == 0
    )


_NORMAL_ADAPTER_ORDER = ("cursor", "antigravity", "pi")


def _holder_name(adapter: str) -> str:
    return f"{adapter}_holder"


def _route_normal_roles(
    profiles: dict[str, dict[str, object]],
    orders: list[list[str]],
) -> dict[str, dict[str, object]]:
    """Place fixture profiles behind the cursor/antigravity/pi adapters.

    A ``fake`` profile is a ``_write_fake_worker`` executable; it takes the
    next normal adapter not already enabled elsewhere. Any adapter a normal
    role still lacks gets an enabled holder whose executable is absent, so it
    is a safe ``unavailable`` after the configured profiles.
    """
    routed = {name: dict(body) for name, body in profiles.items()}
    used = {
        body["adapter"]
        for body in routed.values()
        if body["adapter"] in _NORMAL_ADAPTER_ORDER and body.get("enabled", True)
    }
    for body in routed.values():
        if body["adapter"] != "fake":
            continue
        free = [adapter for adapter in _NORMAL_ADAPTER_ORDER if adapter not in used]
        if not free:
            raise ValueError("fixture has more fake profiles than normal adapters")
        body["adapter"] = free[0]
        used.add(free[0])
    for order in orders:
        present = {
            routed[name]["adapter"]
            for name in order
            if routed[name].get("enabled", True)
        }
        for adapter in _NORMAL_ADAPTER_ORDER:
            if adapter in present:
                continue
            holder = _holder_name(adapter)
            routed.setdefault(
                holder, {"adapter": adapter, "executable": f"{adapter}-holder-absent"}
            )
            order.append(holder)
    return routed


def _write_config(
    repo: Path,
    *,
    profiles: dict[str, dict[str, object]],
    role_order: list[str],
    review_order: list[str] | None = None,
    repair_test_order: list[str] | None = None,
    timeout_seconds: int = 30,
    idle_timeout_seconds: int | None = None,
    max_attempts_per_profile: int | None = None,
    emergency_profile: str | None = None,
    summary_bytes: int = 256,
    raw_output_bytes: int = 1024,
) -> None:
    implementation = list(role_order)
    repair_test = list(repair_test_order if repair_test_order is not None else role_order)
    repair = list(repair_test)
    profiles = _route_normal_roles(profiles, [implementation, repair, repair_test])
    review_names = review_order if review_order is not None else implementation[:1]
    profile_lines: list[str] = []
    custom_lines: list[str] = []
    for name, body in profiles.items():
        profile_lines.append(f"  {name}:")
        profile_lines.append(f"    adapter: {body['adapter']}")
        profile_lines.append(f"    executable: {body['executable']}")
        profile_lines.append(f"    model: {body.get('model', 'auto')}")
        profile_lines.append(f"    enabled: {str(body.get('enabled', True)).lower()}")
        if body["adapter"] == "custom":
            custom_lines.append(f"  {name}:")
            custom_lines.append('    base_argv: ["${workspace}", "${unit_packet}", "${result_path}"]')
            custom_lines.append("    run_everything_argv: []")
            custom_lines.append("    workspace_argv: []")
            custom_lines.append("    model_argv: []")
            custom_lines.append("    prompt_transport: stdin")
            custom_lines.append("    output_format: text")
    implementation_line = ", ".join(implementation)
    repair_line = ", ".join(repair)
    test_line = ", ".join(repair_test)
    review_line = ", ".join(review_names)
    idle_line = (
        f"          idle_timeout_seconds: {idle_timeout_seconds}\n"
        if idle_timeout_seconds is not None
        else ""
    )
    attempts_line = (
        f"          max_attempts_per_profile: {max_attempts_per_profile}\n"
        if max_attempts_per_profile is not None
        else ""
    )
    emergency_line = (
        f"          emergency_profile: {emergency_profile}\n"
        if emergency_profile is not None
        else ""
    )
    block = textwrap.dedent(
        f"""\
        schema: agent-execution/v1

        execution:
          fresh_process_per_unit: true
          permission: run-everything
          fallback_to_host: all-unavailable
          max_parallel: 1
          timeout_seconds: {timeout_seconds}
{idle_line}{attempts_line}{emergency_line}          summary_bytes: {summary_bytes}
          raw_output_bytes: {raw_output_bytes}

        roles:
          implementation: [{implementation_line}]
          repair: [{repair_line}]
          test: [{test_line}]
          review: [{review_line}]

        profiles:
        """
    )
    block += "\n".join(profile_lines) + "\n"
    if custom_lines:
        block += "\ncustom_adapters:\n" + "\n".join(custom_lines) + "\n"
    (repo / "AGENT-EXECUTION.md").write_text(
        "# Fixture\n\n```agent-execution\n" + block + "```\n",
        encoding="utf-8",
    )
    _git(repo, "add", "AGENT-EXECUTION.md")
    _git(repo, "commit", "-m", "add agent-execution config")


def _write_unit(
    path: Path,
    *,
    unit_id: str = "u-test",
    goal: str = "Create marker file",
    allowed_paths: list[str] | None = None,
    timeout_seconds: int = 20,
) -> Path:
    packet = {
        "unit_id": unit_id,
        "goal": goal,
        "allowed_paths": allowed_paths or ["src/marker.txt"],
        "verification": ["test -f src/marker.txt"],
        "constraints": ["leave uncommitted"],
        "network": False,
        "external_cost": False,
        "timeout_seconds": timeout_seconds,
    }
    path.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8")
    return path


_FAKE_ENVELOPE_PY = textwrap.dedent(
    """\
    import json, sys
    adapter, path = sys.argv[1], sys.argv[2]
    with open(path, encoding="utf-8") as handle:
        text = handle.read()
    if text:
        if adapter == "cursor":
            event = {"type": "result", "subtype": "success", "is_error": False, "result": text}
        elif adapter == "antigravity":
            event = {"event": "result", "result": {"response": text}}
        else:
            event = {
                "type": "message_end",
                "message": {
                    "role": "assistant",
                    "content": [{"type": "text", "text": text}],
                    "stopReason": "stop",
                },
            }
        sys.stdout.write("\\n" + json.dumps(event) + "\\n")
    """
)


def _fake_adapter_prelude(scratch: Path) -> str:
    """Run a ``$1``=workspace/``$3``=result body behind cursor, agy, or PI argv.

    Text the body writes to ``$3`` becomes that adapter's final envelope on
    exit, so terminal decoding still goes through the real adapter path.
    """
    return textwrap.dedent(
        f"""\
        _fake_adapter=""
        for _fake_arg in "$@"; do
          case "$_fake_arg" in
            --force) _fake_adapter=cursor ;;
            --dangerously-skip-permissions) _fake_adapter=antigravity ;;
            --approve) _fake_adapter=pi ;;
          esac
        done
        if [ -n "$_fake_adapter" ]; then
          _fake_result=$(mktemp {shlex.quote(str(scratch))}/fake-terminal.XXXXXX) || exit 70
          _fake_emit() {{
            _fake_code=$?
            python3 -c {shlex.quote(_FAKE_ENVELOPE_PY)} "$_fake_adapter" "$_fake_result"
            rm -f "$_fake_result"
            exit "$_fake_code"
          }}
          trap _fake_emit EXIT
          set -- "$(pwd -P)" "" "$_fake_result"
        fi
        """
    )


def _write_fake_worker(
    bindir: Path,
    name: str,
    *,
    body: str,
    help_text: str | None = None,
    native: bool = False,
) -> Path:
    """Write a fake CLI.

    By default the body uses the ``$1``/``$3`` convention behind any normal
    adapter. ``native=True`` leaves argv and stdout exactly as the adapter
    delivers them.
    """
    if help_text is None:
        help_text = "custom worker\n" if native else _normal_help()
    prelude = "" if native else _fake_adapter_prelude(bindir.parent)
    path = bindir / name
    path.write_text(
        "#!/bin/sh\n"
        'if [ "$1" = "--version" ] || [ "$1" = "-v" ] || [ "$1" = "-V" ]; then\n'
        '  echo "0.0.1-test"\n'
        "  exit 0\n"
        "fi\n"
        'if [ "$1" = "--help" ] || [ "$1" = "-h" ]; then\n'
        f"  cat <<'HELP'\n{help_text}HELP\n"
        "  exit 0\n"
        "fi\n"
        + prelude
        + body,
        encoding="utf-8",
    )
    path.chmod(path.stat().st_mode | stat.S_IEXEC)
    return path


def _normal_help() -> str:
    from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS

    markers = {
        marker
        for adapter in _NORMAL_ADAPTER_ORDER
        for marker in _REQUIRED_HELP_MARKERS[adapter]
    }
    return "\n".join(sorted(markers)) + "\n"


def _codex_help() -> str:
    from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS

    return "\n".join(_REQUIRED_HELP_MARKERS["codex"]) + "\n"


def _pi_help() -> str:
    from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS

    return "\n".join(_REQUIRED_HELP_MARKERS["pi"]) + "\n"


def _antigravity_help() -> str:
    from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS

    return "\n".join(_REQUIRED_HELP_MARKERS["antigravity"]) + "\n"


def _terminal_json(status: str = "completed") -> str:
    return json.dumps(
        {
            "terminal_status": status,
            "summary": "done",
            "changed_files": ["src/marker.txt"],
            "evidence": ["wrote marker"],
            "scope_expansion": None,
        },
        separators=(",", ":"),
    )


def _agy_stream_events(final_text: str, *, is_error: bool = False) -> list[str]:
    """Representative Antigravity ``--output-format stream-json`` lines."""
    conv = "b97cb31f-a6bd-40bf-aca8-aaeaa2e1999a"
    result: dict[str, object] = {
        "conversation_id": conv,
        "response": final_text,
        "num_turns": 2,
        "duration_ms": 10,
    }
    if is_error:
        result["is_error"] = True
        result["error"] = "model error"
    events: list[dict[str, object]] = [
        {"event": "init", "conversation_id": conv, "init": {"cwd": "/w", "tools": ["run_command"]}},
        {
            "event": "step_update",
            "step_update": {"conversation_id": conv, "step_index": 0, "state": "DONE", "step_type": "user_input"},
        },
        {
            "event": "step_update",
            "step_update": {"conversation_id": conv, "step_index": 1, "state": "DONE", "step_type": "agent_response"},
        },
        {"event": "result", "conversation_id": conv, "result": result},
    ]
    return [json.dumps(event, separators=(",", ":")) for event in events]


def _pi_json_events(
    final_text: str,
    *,
    prompt_text: str = "unit",
    stop_reason: str = "stop",
    agent_end_padding: int = 0,
) -> list[str]:
    """Representative PI ``--mode json`` lines (see pi docs/json.md)."""
    user = {"role": "user", "content": prompt_text, "timestamp": 1}
    assistant = {
        "role": "assistant",
        "content": [
            {"type": "thinking", "thinking": "plan"},
            {"type": "text", "text": final_text},
        ],
        "stopReason": stop_reason,
    }
    padding = {"role": "toolResult", "content": [{"type": "text", "text": "x" * agent_end_padding}]}
    events: list[dict[str, object]] = [
        {"type": "session", "version": 3, "id": "s", "timestamp": "2026-10-15T00:00:00.000Z", "cwd": "/w"},
        {"type": "agent_start"},
        {"type": "turn_start"},
        {"type": "message_start", "message": user},
        {"type": "message_end", "message": user},
        {"type": "message_start", "message": {"role": "assistant", "content": [], "stopReason": "pending"}},
        {
            "type": "message_update",
            "usage": {"input": 1, "output": 1},
            "assistantMessageEvent": {"type": "text_delta", "contentIndex": 1, "delta": final_text},
        },
        {"type": "message_end", "message": assistant},
        {"type": "turn_end", "message": assistant, "toolResults": []},
        {"type": "agent_end", "messages": [user, padding, assistant], "willRetry": False},
        {"type": "agent_settled"},
    ]
    return [json.dumps(event, separators=(",", ":")) for event in events]


class AgentExecControllerTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-ctrl-")
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.bindir = self.root / "bin"
        self.bindir.mkdir()
        self.repo = self.root / "repo"
        self.repo.mkdir()
        self.receipts = self.root / "receipts"
        self.receipts.mkdir()
        self._old_path = os.environ.get("PATH", "")
        os.environ["PATH"] = f"{self.bindir}{os.pathsep}{self._old_path}"
        self.addCleanup(self._restore_path)
        self._init_repo()

    def _restore_path(self) -> None:
        os.environ["PATH"] = self._old_path

    def _init_repo(self) -> None:
        _git(self.repo, "init")
        _git(self.repo, "config", "user.email", "agent-exec@example.com")
        _git(self.repo, "config", "user.name", "Agent Exec")
        (self.repo / "README.md").write_text("fixture\n", encoding="utf-8")
        (self.repo / "src").mkdir()
        (self.repo / "src" / ".gitkeep").write_text("", encoding="utf-8")
        _git(self.repo, "add", "README.md", "src/.gitkeep")
        _git(self.repo, "commit", "-m", "init")
        _git(self.repo, "checkout", "-b", "feature/agent-exec-test")
        self.base_snap = _canonical_snapshot(self.repo)

    def _run_cli(self, *args: str, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
        cmd_env = os.environ.copy()
        if env:
            cmd_env.update(env)
        cmd_env["PYTHONDONTWRITEBYTECODE"] = "1"
        workspace = Path(__file__).resolve().parents[1]
        existing = cmd_env.get("PYTHONPATH", "")
        cmd_env["PYTHONPATH"] = (
            f"{workspace}{os.pathsep}{existing}" if existing else str(workspace)
        )
        return subprocess.run(
            ["python3", "-m", "scripts.agent_exec", *args],
            cwd=str(self.root),
            capture_output=True,
            text=True,
            env=cmd_env,
        )

    def _assert_canonical_untouched(self) -> None:
        self.assertEqual(_canonical_snapshot(self.repo), self.base_snap)

    def _configure(self, **kwargs: object) -> None:
        _write_config(self.repo, **kwargs)  # type: ignore[arg-type]
        self.base_snap = _canonical_snapshot(self.repo)

    def test_validate_does_not_contact_provider(self) -> None:
        marker = self.root / "provider-contacted"
        _write_fake_worker(
            self.bindir,
            "cursor-agent",
            body=(
                f'echo contacted > "{marker}"\n'
                "exit 1\n"
            ),
            help_text=textwrap.dedent(
                """\
                Usage: agent [options]
                  -p, --print
                  --force
                  --sandbox
                  --trust
                  --workspace
                  --model
                  --output-format
                """
            ),
        )
        self._configure(
            profiles={
                "cursor": {
                    "adapter": "cursor",
                    "executable": "cursor-agent",
                    "model": "auto",
                }
            },
            role_order=["cursor"],
        )
        proc = self._run_cli("validate", "--repo", str(self.repo))
        self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout)
        self.assertFalse(marker.exists())
        self._assert_canonical_untouched()

    def test_run_uses_fresh_process_and_worktree_per_attempt(self) -> None:
        meta = self.root / "meta"
        meta.mkdir()
        _write_fake_worker(
            self.bindir,
            "worker-b",
            body=textwrap.dedent(
                f"""\
                workspace="$1"
                result="$3"
                echo "$$" >> "{meta}/pids"
                echo "$workspace" >> "{meta}/wts"
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":["wrote marker"],"scope_expansion":null}}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"b": {"adapter": "fake", "executable": "worker-b"}},
            role_order=["b"],
        )
        loaded = agent_exec.load_config(self.repo)
        self.assertEqual(loaded.profiles["b"].adapter, "cursor")
        for role in ("implementation", "repair", "test"):
            self.assertEqual(loaded.roles[role][0], "b")
            adapters = [loaded.profiles[name].adapter for name in loaded.roles[role]]
            self.assertEqual(sorted(adapters), sorted(NORMAL_ROLE_ADAPTERS))
        unit = _write_unit(self.root / "unit.json")
        for label in ("first", "second"):
            receipt = self.receipts / f"fresh-{label}.json"
            proc = self._run_cli(
                "run",
                "--repo",
                str(self.repo),
                "--role",
                "implementation",
                "--unit",
                str(unit),
                "--receipt",
                str(receipt),
            )
            self.assertEqual(
                proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout
            )
            data = json.loads(receipt.read_text(encoding="utf-8"))
            self.assertEqual(data["schema"], "agent-exec-receipt/v1")
            self.assertEqual(data["terminal_status"], "success")
        pids = (meta / "pids").read_text(encoding="utf-8").split()
        wts = (meta / "wts").read_text(encoding="utf-8").split()
        self.assertEqual(len(pids), 2)
        self.assertNotEqual(pids[0], pids[1])
        self.assertNotEqual(wts[0], wts[1])
        for wt in wts:
            self.assertTrue(wt.startswith(f"/tmp/agent-exec-{os.getuid()}/"))
        self._assert_canonical_untouched()

    def test_missing_profiles_fall_through_in_declared_order(self) -> None:
        log = self.root / "order.log"
        log.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "second",
            body=textwrap.dedent(
                f"""\
                echo second >> "{log}"
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "missing": {"adapter": "fake", "executable": "does-not-exist-xyz"},
                "second": {"adapter": "fake", "executable": "second"},
            },
            role_order=["missing", "second"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "order.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        names = [a["profile"] for a in data["attempts"]]
        self.assertEqual(names, ["missing", "second"])
        self.assertEqual(data["attempts"][0]["terminal_status"], "unavailable")
        self.assertEqual(log.read_text(encoding="utf-8").strip(), "second")
        self._assert_canonical_untouched()

    def test_all_unavailable_exhaustion_does_not_authorize_host(self) -> None:
        self._configure(
            profiles={
                "m1": {"adapter": "fake", "executable": "gone-one"},
                "m2": {"adapter": "fake", "executable": "gone-two"},
            },
            role_order=["m1", "m2"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "fallback.json"
        before = list(self.repo.rglob("*"))
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(
            proc.returncode,
            agent_exec.EXIT_UNAVAILABLE_EXHAUSTION,
            proc.stderr + proc.stdout,
        )
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "unavailable")
        self.assertNotEqual(data["terminal_status"], "host-fallback-authorized")
        self.assertEqual(
            [a["profile"] for a in data["attempts"]], ["m1", "m2", _holder_name("pi")]
        )
        self.assertTrue(all(a["terminal_status"] == "unavailable" for a in data["attempts"]))
        self.assertEqual(data.get("host_fallback_authorizations"), 0)
        # Controller must not implement the unit itself.
        history_root = self.repo / ".git" / "agent-exec"
        after = [
            path
            for path in self.repo.rglob("*")
            if path != history_root and history_root not in path.parents
        ]
        self.assertEqual(before, after)
        self.assertEqual(data["history"]["status"], "recorded")
        self.assertTrue((history_root / "execution-history.jsonl").is_file())
        self.assertFalse((self.repo / "src" / "marker.txt").exists())
        self.assertIsNone(data.get("host_scope"))
        self._assert_canonical_untouched()

    def test_legacy_exhaustion_without_emergency_does_not_authorize_host(self) -> None:
        self._configure(
            profiles={
                "m1": {"adapter": "fake", "executable": "gone-one"},
                "m2": {"adapter": "fake", "executable": "gone-two"},
            },
            role_order=["m1", "m2"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "no-host.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(
            proc.returncode,
            agent_exec.EXIT_UNAVAILABLE_EXHAUSTION,
            proc.stderr + proc.stdout,
        )
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "unavailable")
        self.assertNotEqual(data["terminal_status"], "host-fallback-authorized")
        self.assertEqual(data.get("host_fallback_authorizations"), 0)
        self.assertIsNone(data.get("host_scope"))
        self.assertFalse((self.repo / "src" / "marker.txt").exists())
        self._assert_canonical_untouched()

    def test_declared_offline_launched_failure_is_failed_unknown_and_stops(self) -> None:
        """network=false/external_cost=false are declarations, not isolation proof."""
        launched = self.root / "declared-offline-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "dirty",
            body=textwrap.dedent(
                """\
                workspace="$1"
                mkdir -p "$workspace/src"
                printf 'tmp\\n' > "$workspace/src/marker.txt"
                exit 9
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "ok",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "dirty": {"adapter": "fake", "executable": "dirty"},
                "ok": {"adapter": "fake", "executable": "ok"},
            },
            role_order=["dirty", "ok"],
        )
        unit = _write_unit(self.root / "unit.json")
        packet = json.loads(unit.read_text(encoding="utf-8"))
        self.assertFalse(packet["network"])
        self.assertFalse(packet["external_cost"])
        receipt = self.receipts / "declared-offline.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        attempt = data["attempts"][0]
        self.assertEqual(attempt["terminal_status"], "failed-unknown")
        self.assertFalse(attempt["cleanup"]["exact_restore"])
        self.assertTrue(attempt["workspace_preserved"])
        self.assertFalse(attempt["cleanup"]["declared_network"])
        self.assertFalse(attempt["cleanup"]["declared_external_cost"])
        self.assertTrue(
            (Path(attempt["worktree"]) / "src" / "marker.txt").is_file()
        )
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self._assert_canonical_untouched()

    def test_safe_unavailable_retries_profile_before_next_success(self) -> None:
        launches = self.root / "profile-launches.log"
        launches.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "thenok",
            body=textwrap.dedent(
                f"""\
                printf 'thenok\\n' >> "{launches}"
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'final\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "retryme": {"adapter": "fake", "executable": "retryme-missing"},
                "thenok": {"adapter": "fake", "executable": "thenok"},
            },
            role_order=["retryme", "thenok"],
            max_attempts_per_profile=3,
        )
        loaded = agent_exec.load_config(self.repo)
        self.assertEqual(loaded.execution.max_attempts_per_profile, 3)
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "retry-ordinal.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        observed = [
            line for line in launches.read_text(encoding="utf-8").splitlines() if line
        ]
        self.assertEqual(observed, ["thenok"])
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(
            [
                (
                    attempt["profile"],
                    attempt.get("attempt_ordinal"),
                    attempt["terminal_status"],
                )
                for attempt in data["attempts"]
            ],
            [
                ("retryme", 1, "unavailable"),
                ("retryme", 2, "unavailable"),
                ("retryme", 3, "unavailable"),
                ("thenok", 1, "success"),
            ],
        )
        self.assertTrue(
            all(attempt["cleanup"]["exact_restore"] for attempt in data["attempts"][:3])
        )
        self._assert_canonical_untouched()

    def test_emergency_profile_runs_once_after_regular_safe_exhaustion(self) -> None:
        launches = self.root / "emergency-launches.log"
        launches.write_text("", encoding="utf-8")
        # Installed but missing required capability flags: each attempt is a
        # safe pre-launch `unavailable`, which is what may route onward.
        normal = (
            ("cursor", "cursor-regular", "incompatible\n"),
            ("antigravity", "agy-regular", "incompatible\n"),
            ("pi", "pi-regular", "incompatible\n"),
        )
        for adapter, executable, help_text in normal:
            _write_fake_worker(
                self.bindir,
                executable,
                help_text=help_text,
                native=True,
                body=textwrap.dedent(
                    f"""\
                    printf '{adapter}\\n' >> "{launches}"
                    exit 9
                    """
                ),
            )
        _write_fake_worker(
            self.bindir,
            "codex",
            help_text=_codex_help(),
            native=True,
            body=textwrap.dedent(
                f"""\
                if [ "$1" = "exec" ] && [ "$2" = "--help" ]; then
                  exec "$0" --help
                fi
                printf 'codex\\n' >> "{launches}"
                result=""
                prev=""
                for arg in "$@"; do
                  if [ "$prev" = "--output-last-message" ]; then
                    result="$arg"
                  fi
                  prev="$arg"
                done
                if [ -z "$result" ]; then
                  echo "missing --output-last-message" >&2
                  exit 2
                fi
                mkdir -p src
                printf 'final\\n' > src/marker.txt
                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}}' > "$result"
                exit 0
                """
            ),
        )
        order = [adapter for adapter, _, _ in normal]
        profiles: dict[str, dict[str, object]] = {
            adapter: {"adapter": adapter, "executable": executable, "model": "auto"}
            for adapter, executable, _ in normal
        }
        profiles["codex"] = {"adapter": "codex", "executable": "codex", "model": "auto"}
        self._configure(
            profiles=profiles,
            role_order=order,
            repair_test_order=order,
            max_attempts_per_profile=3,
            emergency_profile="codex",
        )
        loaded = agent_exec.load_config(self.repo)
        self.assertEqual(loaded.execution.max_attempts_per_profile, 3)
        self.assertEqual(loaded.execution.emergency_profile, "codex")
        self.assertIn("codex", loaded.profiles)
        for role in ("implementation", "repair", "test"):
            self.assertEqual(loaded.roles[role], tuple(order))
        for adapter in order:
            self.assertEqual(loaded.profiles[adapter].adapter, adapter)
        self.assertEqual(loaded.profiles["codex"].adapter, "codex")
        for role_names in loaded.roles.values():
            self.assertNotIn("codex", role_names)
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "emergency-fallback.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        data = json.loads(receipt.read_text(encoding="utf-8"))
        recorded = [
            (
                attempt["profile"],
                attempt.get("attempt_ordinal"),
                attempt["terminal_status"],
                attempt["emergency_fallback"],
            )
            for attempt in data["attempts"]
        ]
        self.assertEqual(
            recorded,
            [
                (adapter, ordinal, "unavailable", False)
                for adapter in order
                for ordinal in (1, 2, 3)
            ]
            + [("codex", 1, "success", True)],
        )
        self.assertTrue(recorded[-1][3])
        self.assertFalse(any(flag for _, _, _, flag in recorded[:-1]))
        self.assertEqual(data.get("host_fallback_authorizations"), 0)
        observed = [
            line for line in launches.read_text(encoding="utf-8").splitlines() if line
        ]
        self.assertEqual(observed, ["codex"])
        self.assertEqual(data["attempts"][-1]["summary"], "done")
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        self._assert_canonical_untouched()

    def test_failed_unknown_blocks_fallback(self) -> None:
        launched = self.root / "launched.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "unknown",
            body=textwrap.dedent(
                """\
                workspace="$1"
                mkdir -p "$workspace/src"
                printf 'x\\n' > "$workspace/src/marker.txt"
                # Abnormal termination / disconnect: cannot prove absence of side effects.
                kill -KILL $$
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "next",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "unknown": {"adapter": "fake", "executable": "unknown"},
                "next": {"adapter": "fake", "executable": "next"},
            },
            role_order=["unknown", "next"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "unknown.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertNotEqual(data["terminal_status"], "host-fallback-authorized")
        self._assert_canonical_untouched()

    def test_scope_expansion_preserves_workspace_and_blocks(self) -> None:
        launched = self.root / "scope-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "scope",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                printf 'bad\\n' > "$workspace/README.md"
                printf '%s\\n' '{"terminal_status":"completed","summary":"expanded","changed_files":["README.md"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "next",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "scope": {"adapter": "fake", "executable": "scope"},
                "next": {"adapter": "fake", "executable": "next"},
            },
            role_order=["scope", "next"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "scope.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(
            proc.returncode, agent_exec.EXIT_SCOPE_EXPANSION, proc.stderr + proc.stdout
        )
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "scope-expansion")
        self.assertEqual(len(data["attempts"]), 1)
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        attempt = data["attempts"][0]
        self.assertTrue(Path(attempt["worktree"]).exists())
        self.assertTrue(attempt.get("workspace_preserved"))
        self._assert_canonical_untouched()

    def test_worker_commit_or_head_move_is_rejected(self) -> None:
        """Worker HEAD/commit, staged index, and config/ref mutations are fail-stop."""
        cases = {
            "commit": textwrap.dedent(
                """\
                workspace="$1"
                mkdir -p "$workspace/src"
                printf 'c\\n' > "$workspace/src/marker.txt"
                git -C "$workspace" add src/marker.txt
                git -C "$workspace" -c user.email=t@e -c user.name=t commit -m 'worker commit'
                exit 0
                """
            ),
            "staged": textwrap.dedent(
                """\
                workspace="$1"
                mkdir -p "$workspace/src"
                printf 'staged\\n' > "$workspace/src/marker.txt"
                git -C "$workspace" add src/marker.txt
                exit 0
                """
            ),
            "config_ref": textwrap.dedent(
                """\
                workspace="$1"
                git -C "$workspace" config --local agentexec.probe 1
                git -C "$workspace" update-ref refs/agent-exec-probe/tmp HEAD
                exit 0
                """
            ),
        }
        for label, body in cases.items():
            with self.subTest(case=label):
                launched = self.root / f"commit-next-{label}.log"
                launched.write_text("", encoding="utf-8")
                worker = f"committer-{label}"
                _write_fake_worker(self.bindir, worker, body=body)
                _write_fake_worker(
                    self.bindir,
                    f"next-{label}",
                    body=textwrap.dedent(
                        f"""\
                        echo next >> "{launched}"
                        exit 0
                        """
                    ),
                )
                self._configure(
                    profiles={
                        "committer": {"adapter": "fake", "executable": worker},
                        "next": {
                            "adapter": "fake",
                            "executable": f"next-{label}",
                        },
                    },
                    role_order=["committer", "next"],
                )
                self.base_snap = _canonical_snapshot(self.repo)
                unit = _write_unit(self.root / f"unit-{label}.json")
                receipt = self.receipts / f"commit-{label}.json"
                proc = self._run_cli(
                    "run",
                    "--repo",
                    str(self.repo),
                    "--role",
                    "implementation",
                    "--unit",
                    str(unit),
                    "--receipt",
                    str(receipt),
                )
                self.assertEqual(
                    proc.returncode,
                    agent_exec.EXIT_BLOCKED,
                    proc.stderr + proc.stdout,
                )
                data = json.loads(receipt.read_text(encoding="utf-8"))
                self.assertEqual(data["terminal_status"], "failed-unknown")
                self.assertEqual(len(data["attempts"]), 1)
                self.assertEqual(launched.read_text(encoding="utf-8"), "")
                self.assertTrue(data["attempts"][0].get("workspace_preserved"))
                self._assert_canonical_untouched()

    def test_untracked_and_binary_files_enter_transport_inventory(self) -> None:
        _write_fake_worker(
            self.bindir,
            "binwriter",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'text\\n' > "$workspace/src/marker.txt"
                printf 'bin\\0data' > "$workspace/src/blob.bin"
                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt","src/blob.bin"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"binwriter": {"adapter": "fake", "executable": "binwriter"}},
            role_order=["binwriter"],
        )
        unit = _write_unit(
            self.root / "unit.json",
            allowed_paths=["src/marker.txt", "src/blob.bin"],
        )
        receipt = self.receipts / "binary.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        changed = set(data["changed_paths"])
        self.assertIn("src/marker.txt", changed)
        self.assertIn("src/blob.bin", changed)
        transport = data["transport"]
        self.assertTrue(transport["ref"].startswith("refs/agent-exec/"))
        self._assert_canonical_untouched()

    def test_timeout_kills_process_group_and_records_receipt(self) -> None:
        child_pid_file = self.root / "child.pid"
        _write_fake_worker(
            self.bindir,
            "sleeper",
            body=textwrap.dedent(
                f"""\
                # Start a child in the same process group
                (sleep 120 & echo $! > "{child_pid_file}"; wait)
                """
            ),
        )
        self._configure(
            profiles={"sleeper": {"adapter": "fake", "executable": "sleeper"}},
            role_order=["sleeper"],
            timeout_seconds=2,
        )
        unit = _write_unit(self.root / "unit.json", timeout_seconds=2)
        receipt = self.receipts / "timeout.json"
        t0 = time.time()
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        elapsed = time.time() - t0
        self.assertLess(elapsed, 15)
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertTrue(data["attempts"][0].get("timed_out"))
        self.assertFalse(data["attempts"][0].get("idle_timed_out"))
        self.assertIn("timed out", data["reason"])
        self.assertNotIn("idle timeout", data["reason"])
        if child_pid_file.exists():
            child = int(child_pid_file.read_text(encoding="utf-8").strip())
            # A zombie is terminated even if a container PID 1 has not reaped it.
            self.assertFalse(_pid_alive(child))
        self._assert_canonical_untouched()

    def test_idle_timeout_kills_silent_worker_before_total(self) -> None:
        child_pid_file = self.root / "idle-child.pid"
        launched = self.root / "idlenext.log"
        launched.write_text("", encoding="utf-8")
        secret = "sk-idletimeoutdeadbeef123456"
        _write_fake_worker(
            self.bindir,
            "silent",
            body=textwrap.dedent(
                f"""\
                # No stdout/stderr. Child stays in this process group.
                (sleep 120 & echo $! > "{child_pid_file}"; wait)
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "idlenext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "silent": {"adapter": "fake", "executable": "silent"},
                "idlenext": {"adapter": "fake", "executable": "idlenext"},
            },
            role_order=["silent", "idlenext"],
            timeout_seconds=40,
            idle_timeout_seconds=2,
        )
        unit = _write_unit(self.root / "unit.json", timeout_seconds=40)
        receipt = self.receipts / "idle.json"
        t0 = time.time()
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
            env={"AGENT_EXEC_TEST_SECRET": secret},
        )
        elapsed = time.time() - t0
        self.assertLess(elapsed, 20)
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        text = receipt.read_text(encoding="utf-8")
        self.assertNotIn(secret, text)
        data = json.loads(text)
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(data["reason"], "idle timeout; process group terminated")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self.assertNotEqual(data["terminal_status"], "host-fallback-authorized")
        self.assertEqual(len(data["attempts"]), 1)
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        attempt = data["attempts"][0]
        self.assertEqual(attempt["terminal_status"], "failed-unknown")
        self.assertTrue(attempt["idle_timed_out"])
        self.assertFalse(attempt["timed_out"])
        self.assertTrue(attempt["workspace_preserved"])
        self.assertTrue(Path(attempt["worktree"]).is_dir())
        cleanup = attempt["cleanup"]
        self.assertFalse(cleanup["exact_restore"])
        self.assertEqual(cleanup["reason"], "idle timeout")
        self.assertTrue(child_pid_file.is_file())
        child = int(child_pid_file.read_text(encoding="utf-8").strip())
        # A zombie is terminated even if a container PID 1 has not reaped it.
        self.assertFalse(_pid_alive(child))
        self._assert_canonical_untouched()

    def test_progress_keeps_worker_alive_past_one_idle_interval(self) -> None:
        _write_fake_worker(
            self.bindir,
            "paced",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                python3 - <<'PY'
                import sys, time
                end = time.monotonic() + 5.0
                while time.monotonic() < end:
                    sys.stdout.buffer.write(b"tick\\n")
                    sys.stdout.buffer.flush()
                    time.sleep(0.5)
                PY
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"progress","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"paced": {"adapter": "fake", "executable": "paced"}},
            role_order=["paced"],
            timeout_seconds=40,
            idle_timeout_seconds=2,
        )
        unit = _write_unit(self.root / "unit.json", timeout_seconds=40)
        receipt = self.receipts / "progress.json"
        t0 = time.time()
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        elapsed = time.time() - t0
        self.assertGreaterEqual(elapsed, 5)
        self.assertLess(elapsed, 25)
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "success")
        self.assertIn("src/marker.txt", data["changed_paths"])
        attempt = data["attempts"][0]
        self.assertFalse(attempt["idle_timed_out"])
        self.assertFalse(attempt["timed_out"])
        raw = Path(attempt["raw_log"]).read_text(encoding="utf-8")
        self.assertGreaterEqual(raw.count("tick"), 8)
        self._assert_canonical_untouched()

    def test_stdin_and_stderr_progress_resets_idle_deadline(self) -> None:
        code = "\n".join(
            [
                "import sys, time",
                "end = time.monotonic() + 3.0",
                "while time.monotonic() < end:",
                "    data = sys.stdin.buffer.read(4096)",
                "    sys.stderr.buffer.write(b'e\\n')",
                "    sys.stderr.buffer.flush()",
                "    time.sleep(0.25)",
                "    if not data:",
                "        continue",
            ]
        )
        raw_log = self.root / "paced-raw.log"
        t0 = time.monotonic()
        meta = agent_exec._supervise(
            ("python3", "-c", code),
            cwd=self.root,
            stdin=b"x" * 200_000,
            timeout=8,
            idle_timeout=1,
            raw_log=raw_log,
            raw_cap=1024,
            secret_values=set(),
        )
        elapsed = time.monotonic() - t0
        self.assertEqual(meta["exit_code"], 0, meta)
        self.assertFalse(meta["idle_timed_out"])
        self.assertFalse(meta["timed_out"])
        self.assertGreaterEqual(elapsed, 2.5)
        self.assertLess(elapsed, 8)

    def test_summary_and_raw_log_caps_are_enforced(self) -> None:
        _write_fake_worker(
            self.bindir,
            "noisy",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                python3 - <<'PY'
                import sys
                chunk = b"A" * (512 * 1024)
                sys.stdout.buffer.write(chunk)
                sys.stdout.buffer.flush()
                sys.stderr.buffer.write(b"B" * (512 * 1024))
                sys.stderr.buffer.flush()
                PY
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"'"$(python3 -c 'print("S"*2000)')"'","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"noisy": {"adapter": "fake", "executable": "noisy"}},
            role_order=["noisy"],
            summary_bytes=64,
            raw_output_bytes=128,
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "caps.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        summary = data["summary"]
        self.assertLessEqual(len(summary.encode("utf-8")), 64)
        attempt = data["attempts"][0]
        raw_path = Path(attempt["raw_log"])
        self.assertTrue(raw_path.is_file())
        self.assertLessEqual(raw_path.stat().st_size, 128)
        cleanup = attempt.get("cleanup") or {}
        self.assertLessEqual(
            int(cleanup.get("stdout_tail_bytes", agent_exec.OUTPUT_TAIL_BYTES)),
            agent_exec.OUTPUT_TAIL_BYTES,
        )
        self.assertLessEqual(
            int(cleanup.get("stderr_tail_bytes", agent_exec.OUTPUT_TAIL_BYTES)),
            agent_exec.OUTPUT_TAIL_BYTES,
        )
        self._assert_canonical_untouched()

    def test_receipt_redacts_environment_secrets(self) -> None:
        secret = "sk-testdeadbeefdeadbeef123456"
        _write_fake_worker(
            self.bindir,
            "leaky",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                echo "token=$AGENT_EXEC_TEST_SECRET"
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"leaky": {"adapter": "fake", "executable": "leaky"}},
            role_order=["leaky"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "redact.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
            env={"AGENT_EXEC_TEST_SECRET": secret},
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        text = receipt.read_text(encoding="utf-8")
        self.assertNotIn(secret, text)
        data = json.loads(text)
        self.assertIn("***", json.dumps(data.get("attempts", [{}])[0].get("summary", "***")))
        self._assert_canonical_untouched()

    def test_success_creates_base_parented_transport_without_touching_canonical_head(
        self,
    ) -> None:
        _write_fake_worker(
            self.bindir,
            "ok",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'shipped\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"ok": {"adapter": "fake", "executable": "ok"}},
            role_order=["ok"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt = self.receipts / "transport.json"
        before = _canonical_snapshot(self.repo)
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        after = _canonical_snapshot(self.repo)
        self.assertEqual(before, after)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "success")
        transport = data["transport"]
        ref = transport["ref"]
        self.assertRegex(ref, r"^refs/agent-exec/[^/]+/[^/]+$")
        parent = _git(self.repo, "rev-parse", f"{ref}^").stdout.strip()
        self.assertEqual(parent, before["head"])
        commit = _git(self.repo, "rev-parse", ref).stdout.strip()
        self.assertNotEqual(commit, before["head"])
        # Canonical branch HEAD unchanged
        branch_head = _git(self.repo, "rev-parse", "HEAD").stdout.strip()
        self.assertEqual(branch_head, before["head"])
        attempt = data["attempts"][0]
        self.assertTrue(attempt.get("workspace_preserved"))
        cleanup = attempt.get("cleanup") or {}
        self.assertFalse(cleanup.get("exact_restore"))
        self.assertTrue(cleanup.get("transport_created"))

    def test_canonical_checkout_mutation_is_failed_unknown(self) -> None:
        """Worker mutating canonical files/index while keeping private clone clean is blocked."""
        launched = self.root / "canon-next.log"
        launched.write_text("", encoding="utf-8")
        # Resolve the canonical checkout via the shared-clone object alternates.
        _write_fake_worker(
            self.bindir,
            "canonmut",
            body=textwrap.dedent(
                """\
                workspace="$1"
                alternates="$workspace/.git/objects/info/alternates"
                objdir=$(head -n1 "$alternates")
                canonical=$(dirname "$(dirname "$objdir")")
                printf 'canonical-mutated\\n' >> "$canonical/README.md"
                git -C "$canonical" add README.md
                # Private clone left clean on purpose.
                exit 0
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "canonnext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "canonmut": {"adapter": "fake", "executable": "canonmut"},
                "canonnext": {"adapter": "fake", "executable": "canonnext"},
            },
            role_order=["canonmut", "canonnext"],
        )
        unit = _write_unit(self.root / "unit-canon.json")
        receipt = self.receipts / "canon.json"
        before = _canonical_snapshot(self.repo)
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        attempt = data["attempts"][0]
        self.assertTrue(attempt.get("workspace_preserved"))
        self.assertIn("canonical", (attempt.get("reason") or "").lower())
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        # Mutation is retained for reconciliation; canonical is intentionally dirty.
        after = _canonical_snapshot(self.repo)
        self.assertNotEqual(after["status"], before["status"])

    def test_corrupt_worktree_git_pointer_yields_failed_unknown_receipt(self) -> None:
        """Corrupted private-clone .git directory must produce a stable blocked receipt."""
        _write_fake_worker(
            self.bindir,
            "corruptgit",
            body=textwrap.dedent(
                """\
                workspace="$1"
                rm -rf "$workspace/.git"
                printf 'not-a-valid-gitdir\\n' > "$workspace/.git"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"corruptgit": {"adapter": "fake", "executable": "corruptgit"}},
            role_order=["corruptgit"],
        )
        unit = _write_unit(self.root / "unit-corrupt.json")
        receipt = self.receipts / "corrupt.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        self.assertTrue(receipt.is_file())
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        attempt = data["attempts"][0]
        self.assertEqual(attempt["terminal_status"], "failed-unknown")
        self.assertTrue(attempt.get("workspace_preserved"))
        self.assertFalse((attempt.get("cleanup") or {}).get("exact_restore"))
        reason = (attempt.get("reason") or "").lower()
        self.assertTrue(
            "inspection" in reason or "pointer" in reason or ".git" in reason,
            attempt.get("reason"),
        )
        self.assertNotIn("Traceback", proc.stderr)
        self._assert_canonical_untouched()

    def test_networked_nonzero_no_file_change_is_failed_unknown(self) -> None:
        """network=true nonzero exit with no file changes cannot be failed-clean."""
        launched = self.root / "net-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "netfail",
            body=textwrap.dedent(
                """\
                # No worktree writes; nonzero exit after possible external contact.
                exit 11
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "netnext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "netfail": {"adapter": "fake", "executable": "netfail"},
                "netnext": {"adapter": "fake", "executable": "netnext"},
            },
            role_order=["netfail", "netnext"],
        )
        unit = _write_unit(self.root / "unit-net.json")
        packet = json.loads(unit.read_text(encoding="utf-8"))
        packet["network"] = True
        packet["external_cost"] = False
        unit.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8")
        receipt = self.receipts / "netfail.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        self.assertEqual(data["attempts"][0]["terminal_status"], "failed-unknown")
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertNotEqual(data["terminal_status"], "host-fallback-authorized")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self._assert_canonical_untouched()

    def test_git_admin_staged_config_ref_mutation_is_failed_unknown(self) -> None:
        """Staged index plus local config/ref mutation preserves workspace and stops."""
        launched = self.root / "admin-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "adminmut",
            body=textwrap.dedent(
                """\
                workspace="$1"
                mkdir -p "$workspace/src"
                printf 'idx\\n' > "$workspace/src/marker.txt"
                git -C "$workspace" add src/marker.txt
                git -C "$workspace" config --local agentexec.mutation yes
                git -C "$workspace" update-ref refs/temporary/agent-exec-admin HEAD
                exit 0
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "adminnext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "adminmut": {"adapter": "fake", "executable": "adminmut"},
                "adminnext": {"adapter": "fake", "executable": "adminnext"},
            },
            role_order=["adminmut", "adminnext"],
        )
        unit = _write_unit(self.root / "unit-admin.json")
        receipt = self.receipts / "admin.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        attempt = data["attempts"][0]
        self.assertTrue(attempt.get("workspace_preserved"))
        inventory = attempt.get("inventory_paths") or attempt.get("changed_paths")
        self.assertIn("src/marker.txt", inventory)
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self._assert_canonical_untouched()

    def test_sibling_canonical_ref_config_mutation_during_private_clone_succeeds(
        self,
    ) -> None:
        """Host sibling branch/ref/config churn must not fail-stop a private clone."""
        ready = self.root / "sibling-ready"
        proceed = self.root / "sibling-proceed"
        for path in (ready, proceed):
            if path.exists():
                path.unlink()

        _write_fake_worker(
            self.bindir,
            "sibworker",
            body=textwrap.dedent(
                f"""\
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf 'ready\\n' > "{ready}"
                # Wait for the host-side sibling mutation before finishing.
                i=0
                while [ ! -f "{proceed}" ]; do
                  i=$((i + 1))
                  if [ "$i" -gt 400 ]; then
                    echo "timed out waiting for host sibling mutation" >&2
                    exit 99
                  fi
                  sleep 0.025
                done
                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":["wrote marker"],"scope_expansion":null}}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"sibworker": {"adapter": "fake", "executable": "sibworker"}},
            role_order=["sibworker"],
        )
        unit = _write_unit(self.root / "unit-sibling.json")
        receipt = self.receipts / "sibling.json"
        before = _canonical_snapshot(self.repo)
        sibling_ref = "refs/heads/feature/host-sibling-delivery"
        sibling_head = before["head"]

        def _host_sibling_mutation() -> None:
            deadline = time.time() + 10
            while not ready.exists():
                if time.time() > deadline:
                    return
                time.sleep(0.025)
            _git(self.repo, "update-ref", sibling_ref, sibling_head)
            _git(
                self.repo,
                "config",
                "--local",
                "branch.feature/host-sibling-delivery.remote",
                "origin",
            )
            _git(
                self.repo,
                "config",
                "--local",
                "branch.feature/host-sibling-delivery.merge",
                sibling_ref,
            )
            proceed.write_text("go\n", encoding="utf-8")

        host_thread = threading.Thread(target=_host_sibling_mutation, daemon=True)
        host_thread.start()
        try:
            proc = self._run_cli(
                "run",
                "--repo",
                str(self.repo),
                "--role",
                "implementation",
                "--unit",
                str(unit),
                "--receipt",
                str(receipt),
            )
        finally:
            host_thread.join(timeout=15)

        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "success")
        attempt = data["attempts"][0]
        worktree = Path(attempt["worktree"])
        self.assertTrue(worktree.is_dir())
        self.assertTrue((worktree / ".git").is_dir())
        self.assertTrue(ready.exists())
        self.assertTrue(proceed.exists())
        self.assertEqual(
            _git(self.repo, "rev-parse", sibling_ref).stdout.strip(), sibling_head
        )

        porcelain = _git(self.repo, "worktree", "list", "--porcelain").stdout
        self.assertNotIn(str(worktree.resolve()), porcelain)
        canon_git = _git(self.repo, "rev-parse", "--absolute-git-dir").stdout.strip()
        worker_git = _git(worktree, "rev-parse", "--absolute-git-dir").stdout.strip()
        self.assertNotEqual(Path(canon_git).resolve(), Path(worker_git).resolve())
        canon_common = _git(
            self.repo, "rev-parse", "--path-format=absolute", "--git-common-dir"
        ).stdout.strip()
        worker_common = _git(
            worktree, "rev-parse", "--path-format=absolute", "--git-common-dir"
        ).stdout.strip()
        self.assertNotEqual(Path(canon_common).resolve(), Path(worker_common).resolve())

        transport = data["transport"]
        ref = transport["ref"]
        self.assertRegex(ref, r"^refs/agent-exec/[^/]+/[^/]+$")
        parent = _git(self.repo, "rev-parse", f"{ref}^").stdout.strip()
        self.assertEqual(parent, before["head"])
        # Canonical HEAD/index identity preserved; sibling delivery side effects are allowed.
        after = _canonical_snapshot(self.repo)
        self.assertEqual(after["head"], before["head"])
        self.assertEqual(after["index"], before["index"])
        branch_head = _git(self.repo, "rev-parse", "HEAD").stdout.strip()
        self.assertEqual(branch_head, before["head"])

    def test_symlink_allowed_path_escape_rejects_without_external_write(self) -> None:
        """Allowed path through symlink escaping the worktree cannot touch outside."""
        outside = self.root / "outside-target"
        outside.mkdir()
        victim = outside / "victim.txt"
        victim.write_text("untouched\n", encoding="utf-8")
        link = self.repo / "src" / "escape"
        link.symlink_to(outside, target_is_directory=True)
        _git(self.repo, "add", "src/escape")
        _git(self.repo, "commit", "-m", "add escape symlink")
        self.base_snap = _canonical_snapshot(self.repo)

        launched = self.root / "symlink-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "symworker",
            body=textwrap.dedent(
                """\
                workspace="$1"
                # If launch were allowed, attempt to write through the symlink.
                printf 'pwned\\n' > "$workspace/src/escape/victim.txt"
                exit 0
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "symnext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "symworker": {"adapter": "fake", "executable": "symworker"},
                "symnext": {"adapter": "fake", "executable": "symnext"},
            },
            role_order=["symworker", "symnext"],
        )
        unit = _write_unit(
            self.root / "unit-symlink.json",
            allowed_paths=["src/escape/victim.txt"],
        )
        receipt = self.receipts / "symlink.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertNotEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        self.assertNotEqual(proc.returncode, agent_exec.EXIT_HOST_FALLBACK)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertNotEqual(data["terminal_status"], "success")
        self.assertNotEqual(data["terminal_status"], "host-fallback-authorized")
        self.assertEqual(victim.read_text(encoding="utf-8"), "untouched\n")
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self._assert_canonical_untouched()

    def test_secret_crossing_raw_log_flush_boundary_is_redacted(self) -> None:
        """Exact env secrets must not leak when split across raw-log drain/flush edges."""
        secret = "SECRETBOUNDARYTOKEN"
        _write_fake_worker(
            self.bindir,
            "boundary",
            body=textwrap.dedent(
                f"""\
                workspace="$1"
                result="$3"
                python3 - <<'PY'
                import os, sys
                secret = os.environ["AGENT_EXEC_TEST_SECRET"]
                sys.stdout.buffer.write(b"A" * 120 + secret.encode() + b"Z" * 60)
                sys.stdout.buffer.flush()
                PY
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"boundary": {"adapter": "fake", "executable": "boundary"}},
            role_order=["boundary"],
            raw_output_bytes=4096,
        )
        unit = _write_unit(self.root / "unit-boundary.json")
        receipt = self.receipts / "boundary-redact.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
            env={"AGENT_EXEC_TEST_SECRET": secret},
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        receipt_text = receipt.read_text(encoding="utf-8")
        self.assertNotIn(secret, receipt_text)
        data = json.loads(receipt_text)
        raw_path = Path(data["attempts"][0]["raw_log"])
        raw_bytes = raw_path.read_bytes()
        self.assertNotIn(secret.encode("utf-8"), raw_bytes)
        self.assertNotIn(secret, raw_bytes.decode("utf-8", "replace"))
        self._assert_canonical_untouched()

    def test_large_stdin_with_early_stdout_does_not_deadlock(self) -> None:
        """Stdin must be select-driven so early stdout beyond pipe capacity cannot hang."""
        # Stay under adapter MAX_UNIT_PACKET_BYTES (262144) while exceeding pipe capacity.
        # Only a review-role custom adapter delivers the packet on stdin.
        pad = "P" * (128 * 1024)
        _write_fake_worker(
            self.bindir,
            "pipefill",
            native=True,
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                # Use -c so process stdin (unit packet) remains readable by Python.
                python3 -c 'import sys; sys.stdout.buffer.write(b"X"*(256*1024)); sys.stdout.buffer.flush(); sys.stdin.buffer.read()'
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={"pipefill": {"adapter": "custom", "executable": "pipefill"}},
            role_order=[],
            review_order=["pipefill"],
            timeout_seconds=15,
            raw_output_bytes=1024,
        )
        loaded = agent_exec.load_config(self.repo)
        self.assertEqual(loaded.custom_adapters["pipefill"].prompt_transport, "stdin")
        self.assertNotIn("pipefill", loaded.roles["implementation"])
        unit = _write_unit(
            self.root / "unit-pipefill.json",
            timeout_seconds=15,
        )
        # Inflate the stdin-delivered unit packet well beyond pipe capacity.
        packet = json.loads(unit.read_text(encoding="utf-8"))
        packet["constraints"] = ["leave uncommitted", pad]
        unit.write_text(json.dumps(packet) + "\n", encoding="utf-8")
        self.assertGreater(unit.stat().st_size, 65536)
        self.assertLessEqual(unit.stat().st_size, 262144)
        receipt = self.receipts / "pipefill.json"
        started = time.monotonic()
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "review",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        elapsed = time.monotonic() - started
        self.assertLessEqual(elapsed, 20.0, f"controller hung for {elapsed:.1f}s")
        self.assertIn(
            proc.returncode,
            {
                agent_exec.EXIT_SUCCESS,
                agent_exec.EXIT_BLOCKED,
            },
            proc.stderr + proc.stdout,
        )
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertIn(
            data["terminal_status"],
            {"success", "failed-unknown", "blocked"},
        )
        if data["terminal_status"] == "success":
            self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS)
        self._assert_canonical_untouched()

    def test_transport_construction_failure_is_failed_unknown(self) -> None:
        """Host transport staging/commit-tree/update-ref failure stays recoverable."""
        launched = self.root / "transport-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "txok",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "txnext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "txok": {"adapter": "fake", "executable": "txok"},
                "txnext": {"adapter": "fake", "executable": "txnext"},
            },
            role_order=["txok", "txnext"],
        )
        unit = _write_unit(self.root / "unit-txfail.json")
        receipt = self.receipts / "txfail.json"
        with mock.patch.object(
            agent_exec,
            "_create_transport_commit",
            side_effect=subprocess.CalledProcessError(
                1, ["git", "commit-tree"], stderr="simulated transport failure"
            ),
        ):
            code = agent_exec.cmd_run(
                self.repo,
                role="implementation",
                unit_path=unit,
                receipt_path=receipt,
            )
        self.assertEqual(code, agent_exec.EXIT_BLOCKED)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        attempt = data["attempts"][0]
        self.assertEqual(attempt["terminal_status"], "failed-unknown")
        self.assertTrue(attempt.get("workspace_preserved"))
        cleanup = attempt.get("cleanup") or {}
        self.assertFalse(cleanup.get("transport_created", True))
        self.assertIn("transport construction failed", attempt.get("reason") or data.get("reason") or "")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self._assert_canonical_untouched()

    def test_transport_commit_does_not_require_configured_git_identity(self) -> None:
        base_head = _git(self.repo, "rev-parse", "HEAD").stdout.strip()
        _git(self.repo, "config", "--unset-all", "user.name")
        _git(self.repo, "config", "--unset-all", "user.email")
        worktree = self.root / "no-identity-worktree"
        agent_exec._create_private_attempt_clone(self.repo, worktree, base_head)
        (worktree / "src" / "marker.txt").write_text("ok\n", encoding="utf-8")

        clean_git_env = {
            "PATH": os.environ["PATH"],
            "GIT_CONFIG_GLOBAL": "/dev/null",
            "GIT_CONFIG_NOSYSTEM": "1",
        }
        with mock.patch.dict(os.environ, clean_git_env, clear=True):
            ref = agent_exec._create_transport_commit(
                repo=self.repo,
                worktree=worktree,
                base_head=base_head,
                run_id="no-identity",
                attempt_id="attempt-1",
                inventory={"paths": ["src/marker.txt"]},
            )

        identity = _git(
            self.repo, "log", "-1", "--format=%an <%ae>|%cn <%ce>", ref
        ).stdout.strip()
        expected = (
            f"{agent_exec.TRANSPORT_AUTHOR_NAME} "
            f"<{agent_exec.TRANSPORT_AUTHOR_EMAIL}>"
        )
        self.assertEqual(identity, f"{expected}|{expected}")

    def test_rename_into_allowed_destination_cannot_hide_source(self) -> None:
        """Rename parsing must keep an out-of-scope source and its destination."""
        launched = self.root / "rename-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(
            self.bindir,
            "renamer",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                git -C "$workspace" mv README.md src/marker.txt
                printf '%s\\n' '{"terminal_status":"completed","summary":"moved","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        _write_fake_worker(
            self.bindir,
            "renamenext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "renamer": {"adapter": "fake", "executable": "renamer"},
                "renamenext": {"adapter": "fake", "executable": "renamenext"},
            },
            role_order=["renamer", "renamenext"],
        )
        unit = _write_unit(self.root / "unit-rename.json")
        receipt = self.receipts / "rename.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertNotEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertNotEqual(data["terminal_status"], "success")
        self.assertEqual(len(data["attempts"]), 1)
        attempt = data["attempts"][0]
        paths = attempt.get("inventory_paths") or []
        self.assertIn("README.md", paths)
        self.assertIn("src/marker.txt", paths)
        self.assertIn("README.md", attempt.get("changed_paths") or [])
        self.assertIn("src/marker.txt", attempt.get("changed_paths") or [])
        self.assertFalse(agent_exec._path_allowed("README.md", ("src/marker.txt",)))
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self._assert_canonical_untouched()

    def test_launched_failure_without_diff_is_failed_unknown_and_stops(self) -> None:
        """No diff and declared-offline packets still do not prove external cleanliness."""
        cases = {
            "clean-exit": "exit 0\n",
            "nonzero-exit": "exit 7\n",
        }
        for label, body in cases.items():
            with self.subTest(case=label):
                launched = self.root / f"restore-next-{label}.log"
                launched.write_text("", encoding="utf-8")
                _write_fake_worker(self.bindir, f"restorer-{label}", body=body)
                _write_fake_worker(
                    self.bindir,
                    f"restorenext-{label}",
                    body=textwrap.dedent(
                        f"""\
                        echo next >> "{launched}"
                        exit 0
                        """
                    ),
                )
                self._configure(
                    profiles={
                        "restorer": {
                            "adapter": "fake",
                            "executable": f"restorer-{label}",
                        },
                        "restorenext": {
                            "adapter": "fake",
                            "executable": f"restorenext-{label}",
                        },
                    },
                    role_order=["restorer", "restorenext"],
                )
                unit = _write_unit(self.root / f"unit-restore-{label}.json")
                receipt = self.receipts / f"restore-{label}.json"
                code = agent_exec.cmd_run(
                    self.repo,
                    role="implementation",
                    unit_path=unit,
                    receipt_path=receipt,
                )
                self.assertEqual(code, agent_exec.EXIT_BLOCKED)
                data = json.loads(receipt.read_text(encoding="utf-8"))
                self.assertEqual(data["terminal_status"], "failed-unknown")
                self.assertEqual(len(data["attempts"]), 1)
                attempt = data["attempts"][0]
                self.assertEqual(attempt["terminal_status"], "failed-unknown")
                self.assertFalse((attempt.get("cleanup") or {}).get("exact_restore"))
                self.assertEqual(launched.read_text(encoding="utf-8"), "")
                self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
                self._assert_canonical_untouched()

    def test_clean_exit_without_diff_and_network_is_failed_unknown(self) -> None:
        """Authorized network with a clean no-diff exit is not failed-clean."""
        launched = self.root / "net-clean-next.log"
        launched.write_text("", encoding="utf-8")
        _write_fake_worker(self.bindir, "netclean", body="exit 0\n")
        _write_fake_worker(
            self.bindir,
            "netcleannext",
            body=textwrap.dedent(
                f"""\
                echo next >> "{launched}"
                exit 0
                """
            ),
        )
        self._configure(
            profiles={
                "netclean": {"adapter": "fake", "executable": "netclean"},
                "netcleannext": {"adapter": "fake", "executable": "netcleannext"},
            },
            role_order=["netclean", "netcleannext"],
        )
        unit = _write_unit(self.root / "unit-net-clean.json")
        packet = json.loads(unit.read_text(encoding="utf-8"))
        packet["network"] = True
        packet["external_cost"] = False
        unit.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8")
        receipt = self.receipts / "net-clean.json"
        proc = self._run_cli(
            "run",
            "--repo",
            str(self.repo),
            "--role",
            "implementation",
            "--unit",
            str(unit),
            "--receipt",
            str(receipt),
        )
        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
        data = json.loads(receipt.read_text(encoding="utf-8"))
        self.assertEqual(data["terminal_status"], "failed-unknown")
        self.assertEqual(len(data["attempts"]), 1)
        self.assertEqual(data["attempts"][0]["terminal_status"], "failed-unknown")
        self.assertEqual(launched.read_text(encoding="utf-8"), "")
        self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
        self._assert_canonical_untouched()

    def test_worker_fail_stop_terminal_beats_success_and_no_change(self) -> None:
        """blocked, cancelled, and failed-unknown stop before success or no-change."""
        cases = (
            ("blocked", agent_exec.EXIT_BLOCKED, True),
            ("cancelled", agent_exec.EXIT_CANCELLED, True),
            ("failed-unknown", agent_exec.EXIT_BLOCKED, True),
            ("cancelled", agent_exec.EXIT_CANCELLED, False),
        )
        for status, exit_code, write_file in cases:
            label = f"{status}-{'files' if write_file else 'nodiff'}"
            with self.subTest(case=label):
                launched = self.root / f"failstop-{label}.log"
                launched.write_text("", encoding="utf-8")
                write = ""
                if write_file:
                    write = (
                        'mkdir -p "$workspace/src"\n'
                        "printf 'x\\n' > \"$workspace/src/marker.txt\"\n"
                    )
                _write_fake_worker(
                    self.bindir,
                    f"failstop-{label}",
                    body=textwrap.dedent(
                        f"""\
                        workspace="$1"
                        result="$3"
                        {write}printf '%s\\n' '{{"terminal_status":"{status}","summary":"stop","changed_files":[],"evidence":[],"scope_expansion":null}}' > "$result"
                        exit 0
                        """
                    ),
                )
                _write_fake_worker(
                    self.bindir,
                    f"failstop-next-{label}",
                    body=textwrap.dedent(
                        f"""\
                        echo next >> "{launched}"
                        exit 0
                        """
                    ),
                )
                self._configure(
                    profiles={
                        "failstop": {
                            "adapter": "fake",
                            "executable": f"failstop-{label}",
                        },
                        "failnext": {
                            "adapter": "fake",
                            "executable": f"failstop-next-{label}",
                        },
                    },
                    role_order=["failstop", "failnext"],
                )
                unit = _write_unit(self.root / f"unit-failstop-{label}.json")
                receipt = self.receipts / f"failstop-{label}.json"
                proc = self._run_cli(
                    "run",
                    "--repo",
                    str(self.repo),
                    "--role",
                    "implementation",
                    "--unit",
                    str(unit),
                    "--receipt",
                    str(receipt),
                )
                self.assertEqual(proc.returncode, exit_code, proc.stderr + proc.stdout)
                data = json.loads(receipt.read_text(encoding="utf-8"))
                self.assertEqual(data["terminal_status"], status)
                self.assertEqual(len(data["attempts"]), 1)
                attempt = data["attempts"][0]
                self.assertEqual(attempt["terminal_status"], status)
                if write_file:
                    self.assertIn("src/marker.txt", attempt.get("changed_paths") or [])
                self.assertEqual(launched.read_text(encoding="utf-8"), "")
                self.assertEqual(data.get("host_fallback_authorizations", 0), 0)
                self._assert_canonical_untouched()

    def test_builtin_event_envelopes_gate_success(self) -> None:
        """Real Cursor/agy/PI output envelopes decide the terminal; prose fails closed."""
        prose = "All done, the marker is written."
        cases = (
            ("cursor", _cursor_help(), _cursor_stream_events("ok\n" + _terminal_json()), "success"),
            ("cursor", _cursor_help(), _cursor_stream_events(prose), "failed-unknown"),
            ("cursor", _cursor_help(), _cursor_stream_events(_terminal_json("blocked")), "blocked"),
            (
                "cursor",
                _cursor_help(),
                _cursor_stream_events(_terminal_json(), is_error=True),
                "failed-unknown",
            ),
            ("antigravity", _antigravity_help(), _agy_stream_events("ok " + _terminal_json()), "success"),
            ("antigravity", _antigravity_help(), _agy_stream_events(prose), "failed-unknown"),
            (
                "pi",
                _pi_help(),
                _pi_json_events(
                    "ok\n" + _terminal_json() + "\n",
                    agent_end_padding=2 * agent_exec.TERMINAL_LINE_MAX_BYTES,
                ),
                "success",
            ),
            ("pi", _pi_help(), _pi_json_events(prose, prompt_text=_terminal_json()), "failed-unknown"),
            ("pi", _pi_help(), [], "failed-unknown"),
        )
        expected_exit = {
            "success": agent_exec.EXIT_SUCCESS,
            "failed-unknown": agent_exec.EXIT_BLOCKED,
            "blocked": agent_exec.EXIT_BLOCKED,
        }
        for index, (adapter, help_text, events, expected) in enumerate(cases):
            label = f"{adapter}-{index}-{expected}"
            with self.subTest(case=label):
                events_path = self.root / f"events-{label}.jsonl"
                events_path.write_text(
                    "".join(line + "\n" for line in events), encoding="utf-8"
                )
                executable = f"envelope-{label}"
                _write_fake_worker(
                    self.bindir,
                    executable,
                    help_text=help_text,
                    native=True,
                    body=textwrap.dedent(
                        f"""\
                        mkdir -p src
                        printf 'ok\\n' > src/marker.txt
                        cat "{events_path}"
                        exit 0
                        """
                    ),
                )
                self._configure(
                    profiles={
                        adapter: {"adapter": adapter, "executable": executable}
                    },
                    role_order=[adapter],
                )
                unit = _write_unit(self.root / f"unit-{label}.json")
                receipt = self.receipts / f"envelope-{label}.json"
                code = agent_exec.cmd_run(
                    self.repo,
                    role="implementation",
                    unit_path=unit,
                    receipt_path=receipt,
                )
                data = json.loads(receipt.read_text(encoding="utf-8"))
                self.assertEqual(data["terminal_status"], expected, data.get("reason"))
                self.assertEqual(code, expected_exit[expected])
                self.assertEqual(len(data["attempts"]), 1)
                attempt = data["attempts"][0]
                self.assertEqual(attempt["exit_code"], 0)
                self.assertEqual(attempt["changed_paths"], ["src/marker.txt"])
                if expected == "success":
                    self.assertTrue(attempt["transport_ref"])
                else:
                    self.assertIsNone(attempt.get("transport_ref"))
                    self.assertTrue(attempt["workspace_preserved"])
                if expected == "failed-unknown":
                    self.assertIn("no valid completed terminal record", attempt["reason"])
                self._assert_canonical_untouched()


class TerminalRecordDecodingTests(unittest.TestCase):
    """Adapter-specific terminal decoding over representative real envelopes."""

    def _decode(
        self,
        adapter: str,
        lines: list[str | None],
        *,
        stdout_tail: str | None = None,
        result_text: str | None = None,
    ) -> dict[str, object] | None:
        from scripts.agent_exec_adapters import decode_terminal_record

        tail = stdout_tail
        if tail is None:
            tail = "".join(line + "\n" for line in lines if line is not None)
        return decode_terminal_record(
            adapter, event_lines=lines, stdout_tail=tail, result_text=result_text
        )

    def test_cursor_stream_json_result_is_decoded(self) -> None:
        lines = _cursor_stream_events("Done.\n" + _terminal_json() + "\n  \n")
        record = self._decode("cursor", lines)
        assert record is not None
        self.assertEqual(record["terminal_status"], "completed")
        self.assertEqual(record["changed_files"], ["src/marker.txt"])

    def test_terminal_record_must_be_sole_final_content(self) -> None:
        invalid = json.dumps({"terminal_status": "done", "summary": "x"})
        texts = {
            "trailing-prose": "Done.\n" + _terminal_json() + "\nThanks.",
            "trailing-invalid-object": _terminal_json() + "\n" + invalid,
            "trailing-non-terminal-object": _terminal_json() + ' {"note":"x"}',
            "truncated-terminal-object": _terminal_json() + '\n{"terminal_status":',
            "two-records": _terminal_json() + "\n" + _terminal_json(),
            "conflicting-records": _terminal_json() + "\n" + _terminal_json("blocked"),
        }
        for name, text in texts.items():
            for adapter, lines in (
                ("cursor", _cursor_stream_events(text)),
                ("antigravity", _agy_stream_events(text)),
                ("pi", _pi_json_events(text)),
            ):
                with self.subTest(case=name, adapter=adapter):
                    self.assertIsNone(self._decode(adapter, lines))
            with self.subTest(case=name, adapter="custom"):
                self.assertIsNone(
                    self._decode("custom", [], stdout_tail="", result_text=text)
                )

    def test_cursor_user_echo_and_error_result_are_not_records(self) -> None:
        lines = _cursor_stream_events("prose only")
        echo = json.loads(lines[1])
        echo["message"]["content"][0]["text"] = _terminal_json()
        lines[1] = json.dumps(echo)
        self.assertIsNone(self._decode("cursor", lines))
        self.assertIsNone(
            self._decode("cursor", _cursor_stream_events(_terminal_json(), is_error=True))
        )
        self.assertIsNone(self._decode("cursor", _cursor_stream_events(_terminal_json())[:-1]))

    def test_cursor_blocked_record_is_decoded_as_fail_stop(self) -> None:
        record = self._decode("cursor", _cursor_stream_events(_terminal_json("blocked")))
        assert record is not None
        self.assertEqual(record["terminal_status"], "blocked")

    def test_agy_stream_json_and_json_results_are_decoded(self) -> None:
        record = self._decode("antigravity", _agy_stream_events("ok " + _terminal_json()))
        assert record is not None
        self.assertEqual(record["terminal_status"], "completed")
        single = json.dumps(
            {"conversation_id": "c", "response": "ok\n" + _terminal_json(), "num_turns": 1},
            indent=2,
        )
        record = self._decode("antigravity", [], stdout_tail=single)
        assert record is not None
        self.assertEqual(record["terminal_status"], "completed")
        self.assertIsNone(
            self._decode("antigravity", _agy_stream_events(_terminal_json(), is_error=True))
        )
        self.assertIsNone(self._decode("antigravity", _agy_stream_events(_terminal_json())[:-1]))

    def test_pi_json_final_assistant_message_is_decoded(self) -> None:
        lines = _pi_json_events("ok\n" + _terminal_json())
        record = self._decode("pi", lines)
        assert record is not None
        self.assertEqual(record["terminal_status"], "completed")
        self.assertIsNone(
            self._decode("pi", _pi_json_events("prose", prompt_text=_terminal_json()))
        )
        self.assertIsNone(
            self._decode("pi", _pi_json_events(_terminal_json(), stop_reason="error"))
        )

    def test_prose_missing_or_invalid_terminal_fails_closed(self) -> None:
        for adapter, lines in (
            ("cursor", _cursor_stream_events("finished, no record")),
            ("antigravity", _agy_stream_events("finished, no record")),
            ("pi", _pi_json_events("finished, no record")),
            ("cursor", []),
            ("antigravity", []),
            ("pi", []),
        ):
            with self.subTest(adapter=adapter, lines=len(lines)):
                self.assertIsNone(self._decode(adapter, lines))
        invalid = json.dumps({"terminal_status": "done", "summary": "x"})
        self.assertIsNone(self._decode("cursor", _cursor_stream_events(invalid)))
        missing_keys = json.dumps({"terminal_status": "completed", "summary": "x"})
        self.assertIsNone(self._decode("cursor", _cursor_stream_events(missing_keys)))

    def test_raw_stdout_record_is_not_an_envelope_for_builtins(self) -> None:
        bare = [_terminal_json()]
        for adapter in ("cursor", "antigravity", "pi"):
            with self.subTest(adapter=adapter):
                self.assertIsNone(self._decode(adapter, bare))
        record = self._decode("custom", [], stdout_tail="", result_text=_terminal_json())
        assert record is not None
        self.assertEqual(record["terminal_status"], "completed")

    def test_prompt_states_schema_without_being_a_record(self) -> None:
        from scripts import agent_exec_adapters as adapters

        prompt = adapters._build_bounded_unit_prompt('{"unit_id":"u"}')
        self.assertIn(
            json.dumps(adapters.TERMINAL_RECORD_SCHEMA, separators=(",", ":")), prompt
        )
        for status in adapters.TERMINAL_STATUSES:
            self.assertIn(status, prompt)
        self.assertIsNone(adapters.extract_terminal_record(prompt))

    def test_stdout_tail_keeps_final_envelope_behind_oversized_event(self) -> None:
        big = 2 * agent_exec.TERMINAL_LINE_MAX_BYTES
        stream = "".join(
            line + "\n"
            for line in _pi_json_events("ok " + _terminal_json(), agent_end_padding=big)
        ).encode("utf-8")
        tail = agent_exec._StdoutTail(agent_exec.OUTPUT_TAIL_BYTES)
        for start in range(0, len(stream), 4093):
            tail.extend(stream[start : start + 4093])
        self.assertNotIn(b"message_end", tail.text().encode("utf-8"))
        record = self._decode("pi", tail.terminal_lines(), stdout_tail=tail.text())
        assert record is not None
        self.assertEqual(record["terminal_status"], "completed")

        oversized = _cursor_stream_events("x" * big + _terminal_json())
        tail = agent_exec._StdoutTail(agent_exec.OUTPUT_TAIL_BYTES)
        tail.extend("".join(line + "\n" for line in oversized).encode("utf-8"))
        lines = tail.terminal_lines()
        self.assertIsNone(lines[-1])
        self.assertIsNone(self._decode("cursor", lines, stdout_tail=tail.text()))


class AgentExecHistoryTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-history-")
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.bindir = self.root / "bin"
        self.bindir.mkdir()
        self.repo = self.root / "repo"
        self.repo.mkdir()
        self.receipts = self.root / "receipts"
        self.receipts.mkdir()
        self.old_path = os.environ.get("PATH", "")
        os.environ["PATH"] = f"{self.bindir}{os.pathsep}{self.old_path}"
        self.addCleanup(self._restore_path)
        _git(self.repo, "init")
        _git(self.repo, "config", "user.email", "history@example.com")
        _git(self.repo, "config", "user.name", "History Test")
        (self.repo / "README.md").write_text("fixture\n", encoding="utf-8")
        (self.repo / "src").mkdir()
        (self.repo / "src" / ".gitkeep").write_text("", encoding="utf-8")
        _git(self.repo, "add", "README.md", "src/.gitkeep")
        _git(self.repo, "commit", "-m", "init")
        _git(self.repo, "checkout", "-b", "feature/history-test")

    def _restore_path(self) -> None:
        os.environ["PATH"] = self.old_path

    @property
    def history_path(self) -> Path:
        return self.repo / ".git" / "agent-exec" / "execution-history.jsonl"

    def test_history_appends_safe_success_and_fallback_records(self) -> None:
        secret = "sk-historysecretdeadbeef123456"
        _write_fake_worker(
            self.bindir,
            "history-writer",
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'ok\n' > "$workspace/src/marker.txt"
                printf '%s\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        _write_config(
            self.repo,
            profiles={
                "writer": {
                    "adapter": "fake",
                    "executable": "history-writer",
                }
            },
            role_order=["writer"],
        )
        unit = _write_unit(self.root / "success.json", unit_id=f"unit-{secret}")
        receipt_path = self.receipts / "success.json"
        with mock.patch.dict(os.environ, {"AGENT_EXEC_TEST_SECRET": secret}):
            code = agent_exec.cmd_run(
                self.repo,
                role="implementation",
                unit_path=unit,
                receipt_path=receipt_path,
            )
        self.assertEqual(code, agent_exec.EXIT_SUCCESS)
        receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
        self.assertEqual(receipt["history"]["status"], "recorded")
        self.assertEqual(Path(receipt["history"]["path"]), self.history_path)
        self.assertEqual(stat.S_IMODE(self.history_path.parent.stat().st_mode), 0o700)
        self.assertEqual(stat.S_IMODE(self.history_path.stat().st_mode), 0o600)

        _write_config(
            self.repo,
            profiles={
                "missing": {
                    "adapter": "fake",
                    "executable": "missing-history-cli",
                }
            },
            role_order=["missing"],
        )
        fallback_unit = _write_unit(self.root / "fallback.json", unit_id="unit-fallback")
        fallback_receipt = self.receipts / "fallback.json"
        fallback_code = agent_exec.cmd_run(
            self.repo,
            role="implementation",
            unit_path=fallback_unit,
            receipt_path=fallback_receipt,
        )
        self.assertEqual(fallback_code, agent_exec.EXIT_UNAVAILABLE_EXHAUSTION)

        text = self.history_path.read_text(encoding="utf-8")
        self.assertNotIn(secret, text)
        self.assertNotIn("raw_log", text)
        self.assertNotIn('"summary"', text)
        records = [json.loads(line) for line in text.splitlines()]
        self.assertEqual(len(records), 2)
        success, fallback = records
        self.assertEqual(success["schema"], "agent-exec-history/v1")
        self.assertEqual(success["terminal_status"], "success")
        self.assertEqual(success["verification_result"], "NOT_RUN")
        self.assertEqual(success["review_result"], "NOT_RUN")
        self.assertEqual(success["final_result"], "NOT_RUN")
        self.assertEqual(success["attempts"][0]["fresh_session"], True)
        self.assertEqual(success["attempts"][0]["session_id"], "not-observable")
        self.assertEqual(success["attempts"][0]["model_requested"], "auto")
        self.assertEqual(fallback["terminal_status"], "unavailable")
        self.assertEqual(fallback["host_fallback_authorizations"], 0)

    def test_history_failure_does_not_change_run_outcome(self) -> None:
        _write_config(
            self.repo,
            profiles={
                "missing": {
                    "adapter": "fake",
                    "executable": "missing-history-cli",
                }
            },
            role_order=["missing"],
        )
        unit = _write_unit(self.root / "unit.json")
        receipt_path = self.receipts / "failed-history.json"
        secret = "sk-historyfailuredeadbeef123456"
        with mock.patch.dict(os.environ, {"AGENT_EXEC_TEST_SECRET": secret}), mock.patch.object(
            agent_exec,
            "_append_execution_history",
            side_effect=OSError(f"history blocked {secret}"),
        ):
            code = agent_exec.cmd_run(
                self.repo,
                role="implementation",
                unit_path=unit,
                receipt_path=receipt_path,
            )
        self.assertEqual(code, agent_exec.EXIT_UNAVAILABLE_EXHAUSTION)
        receipt_text = receipt_path.read_text(encoding="utf-8")
        self.assertNotIn(secret, receipt_text)
        receipt = json.loads(receipt_text)
        self.assertEqual(receipt["terminal_status"], "unavailable")
        self.assertEqual(receipt.get("host_fallback_authorizations"), 0)
        self.assertEqual(receipt["history"]["status"], "failed")

    def test_invalid_config_also_appends_one_history_record(self) -> None:
        (self.repo / "AGENT-EXECUTION.md").write_text("invalid\n", encoding="utf-8")
        _git(self.repo, "add", "AGENT-EXECUTION.md")
        _git(self.repo, "commit", "-m", "invalid config fixture")
        receipt_path = self.receipts / "invalid.json"
        code = agent_exec.cmd_run(
            self.repo,
            role="implementation",
            unit_path=self.root / "unused.json",
            receipt_path=receipt_path,
        )
        self.assertEqual(code, agent_exec.EXIT_INVALID_CONFIG)
        receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
        self.assertEqual(receipt["terminal_status"], "invalid-config")
        self.assertEqual(receipt["history"]["status"], "recorded")
        records = [
            json.loads(line)
            for line in self.history_path.read_text(encoding="utf-8").splitlines()
        ]
        self.assertEqual(len(records), 1)
        self.assertEqual(records[0]["terminal_status"], "invalid-config")
        self.assertEqual(records[0]["attempts"], [])

    def _run_marker_success(self, name: str) -> tuple[Path, dict[str, object]]:
        _write_fake_worker(
            self.bindir,
            name,
            body=textwrap.dedent(
                """\
                workspace="$1"
                result="$3"
                mkdir -p "$workspace/src"
                printf 'ok\\n' > "$workspace/src/marker.txt"
                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
                exit 0
                """
            ),
        )
        _write_config(
            self.repo,
            profiles={name: {"adapter": "fake", "executable": name}},
            role_order=[name],
        )
        unit = _write_unit(self.root / f"{name}.json", unit_id=f"unit-{name}")
        receipt_path = self.receipts / f"{name}.json"
        code = agent_exec.cmd_run(
            self.repo,
            role="implementation",
            unit_path=unit,
            receipt_path=receipt_path,
        )
        receipt_text = (
            receipt_path.read_text(encoding="utf-8") if receipt_path.is_file() else ""
        )
        self.assertEqual(code, agent_exec.EXIT_SUCCESS, receipt_text)
        return receipt_path, json.loads(receipt_text)

    def test_success_history_is_mode_correct_and_signed(self) -> None:
        receipt_path, receipt = self._run_marker_success("signedhistory")
        key_path = self.history_path.parent / "attestation.key"
        self.assertTrue(key_path.is_file())
        self.assertFalse(key_path.is_symlink())
        self.assertFalse(key_path.parent.is_symlink())
        self.assertEqual(stat.S_IMODE(key_path.parent.stat().st_mode), 0o700)
        self.assertEqual(stat.S_IMODE(key_path.stat().st_mode), 0o600)
        key = key_path.read_bytes()
        self.assertEqual(len(key), 32)
        self.assertNotIn(key, receipt_path.read_bytes())
        self.assertNotIn(key, self.history_path.read_bytes())

        records = [
            json.loads(line)
            for line in self.history_path.read_text(encoding="utf-8").splitlines()
        ]
        self.assertEqual(len(records), 1)
        record = records[0]
        transport = record["transport"]
        ref = transport["ref"]
        commit = _git(self.repo, "rev-parse", ref).stdout.strip()
        self.assertEqual(record["terminal_status"], "success")
        self.assertEqual(record["run_id"], receipt["run_id"])
        self.assertEqual(transport["parent"], receipt["transport"]["parent"])
        self.assertEqual(transport["commit"], commit)
        self.assertEqual(receipt["transport"]["commit"], commit)
        self.assertEqual(record["changed_paths"], receipt["changed_paths"])
        body = {field: value for field, value in record.items() if field != "attestation"}
        payload = json.dumps(
            body, ensure_ascii=False, sort_keys=True, separators=(",", ":")
        ).encode("utf-8")
        self.assertEqual(
            hmac.new(key, payload, hashlib.sha256).hexdigest(),
            record["attestation"],
        )
        identity = receipt["history"]
        self.assertEqual(identity["status"], "recorded")
        self.assertEqual(identity["run_id"], receipt["run_id"])
        self.assertEqual(identity["record_id"], record["record_id"])
        self.assertEqual(identity["local_integrity_mac"], record["attestation"])
        self.assertNotIn("attestation", identity)
        self.assertNotIn(key.hex(), json.dumps(identity))

    def test_attestation_key_symlink_is_not_followed(self) -> None:
        history_dir = self.history_path.parent
        history_dir.mkdir(mode=0o700)
        outside = self.root / "outside-key"
        outside.write_bytes(b"A" * 32)
        key_path = history_dir / "attestation.key"
        key_path.symlink_to(outside)
        receipt_path, receipt = self._run_marker_success("symlinkkey")
        self.assertTrue(key_path.is_symlink())
        self.assertEqual(outside.read_bytes(), b"A" * 32)
        self.assertEqual(receipt["history"]["status"], "failed")
        self.assertNotIn(b"A" * 32, receipt_path.read_bytes())

    def test_attestation_key_first_run_race_shares_one_key(self) -> None:
        found: list[bytes] = []
        errors: list[BaseException] = []
        barrier = threading.Barrier(8)

        def load() -> None:
            try:
                barrier.wait(timeout=5)
                found.append(agent_exec._ensure_local_integrity_key(self.repo))
            except BaseException as exc:  # noqa: BLE001 - race failures are the assertion
                errors.append(exc)

        threads = [threading.Thread(target=load) for _ in range(8)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(timeout=10)
            self.assertFalse(thread.is_alive())
        self.assertEqual(errors, [])
        self.assertEqual(len(found), 8)
        self.assertEqual(len({item for item in found}), 1)
        self.assertEqual(len(found[0]), 32)
        key_path = self.history_path.parent / "attestation.key"
        self.assertFalse(key_path.is_symlink())
        self.assertEqual(key_path.read_bytes(), found[0])
        self.assertEqual(stat.S_IMODE(key_path.stat().st_mode), 0o600)
        self.assertEqual(stat.S_IMODE(key_path.parent.stat().st_mode), 0o700)


# Opt-in real CLI probes (U5). Default unittest must not contact providers.
U5_PROFILES = (
    "cursor",
    "antigravity",
    "claude",
    "codex",
    "pi",
    "kimi",
    "opencode",
)
U5_EXECUTABLE = {
    "cursor": "cursor-agent",
    "antigravity": "agy",
    "claude": "claude",
    "codex": "codex",
    "pi": "pi",
    "kimi": "kimi",
    "opencode": "opencode",
}
# Normal roles need exactly one enabled cursor, antigravity, and PI profile and
# reject every other adapter. The probe runs its one selected profile alone
# from review; normal roles keep only uninvoked holders.
_PROBE_NORMAL_HOLDERS = [_holder_name(adapter) for adapter in _NORMAL_ADAPTER_ORDER]
U5_BRANCH = "feature/u5-real-probe"
U5_MARKER_REL = "src/u5-marker.txt"
U5_MARKER_BODY = "U5-REAL-PROBE-MARKER\n"
# Envelope adapters JSON-escape the final text, so quotes may be backslashed.
_MODEL_RECEIPT_RE = re.compile(
    r'\\?"model_receipt_status\\?"\s*:\s*\\?"(verified|not-observable|mismatch|unavailable)\\?"'
)


def real_probe_opt_in(env: os._Environ[str] | dict[str, str] | None = None) -> bool:
    """True only for the explicit host gate. Other truthy strings stay closed."""
    source = os.environ if env is None else env
    return source.get("RUN_REAL_AGENT_EXEC") == "1"


def require_single_real_profile(env: dict[str, str]) -> str:
    """Select exactly one installed-CLI profile. Reject lists and unknown names."""
    raw = env.get("AGENT_EXEC_REAL_PROFILE", "")
    if not raw or raw.strip() != raw:
        raise ValueError(
            "AGENT_EXEC_REAL_PROFILE must name exactly one profile"
        )
    if any(sep in raw for sep in (",", " ", "\t", "\n", ";", ":")):
        raise ValueError(
            "AGENT_EXEC_REAL_PROFILE must name exactly one profile"
        )
    if raw not in U5_PROFILES:
        raise ValueError(f"unknown real probe profile: {raw}")
    return raw


def _observe_model_receipt(attempt: dict[str, object]) -> str:
    """Record a model receipt only when the raw output actually carries one."""
    chunks = [str(attempt.get("summary") or "")]
    raw_log = attempt.get("raw_log")
    if isinstance(raw_log, str) and raw_log:
        path = Path(raw_log)
        if path.is_file():
            chunks.append(path.read_text(encoding="utf-8", errors="replace"))
    found = _MODEL_RECEIPT_RE.findall("\n".join(chunks))
    if not found:
        return "not-observable"
    if "mismatch" in found:
        return "mismatch"
    return found[-1]


def _probe_tool_bindir(extra: Path | None = None) -> tempfile.TemporaryDirectory[str]:
    """PATH containing only git, python3, and an optional fake CLI directory."""
    tool_dir = tempfile.TemporaryDirectory(prefix="u5-probe-path-")
    bindir = Path(tool_dir.name)
    if extra is not None:
        for entry in extra.iterdir():
            target = bindir / entry.name
            if entry.is_file():
                target.symlink_to(entry.resolve())
    for tool in ("git", "python3"):
        found = shutil.which(tool)
        if not found:
            tool_dir.cleanup()
            raise RuntimeError(f"required tool missing from PATH: {tool}")
        link = bindir / tool
        if not link.exists():
            link.symlink_to(found)
    return tool_dir


def _write_probe_double(
    bindir: Path,
    name: str,
    *,
    help_text: str,
    run_body: str,
) -> None:
    path = bindir / name
    # printf is a shell builtin. The offline PATH has no cat/mkdir.
    help_lines = help_text.splitlines() or [""]
    help_printf = "".join(
        f"  printf '%s\\n' {shlex.quote(line)}\n" for line in help_lines
    )
    path.write_text(
        "#!/bin/sh\n"
        'if [ "$1" = "--version" ] || [ "$1" = "-v" ] || [ "$1" = "-V" ]; then\n'
        '  echo "0.0.1-probe"\n'
        "  exit 0\n"
        "fi\n"
        'if [ "$1" = "--help" ] || [ "$1" = "-h" ]; then\n'
        + help_printf
        + "  exit 0\n"
        "fi\n"
        + run_body,
        encoding="utf-8",
    )
    path.chmod(path.stat().st_mode | stat.S_IEXEC)


def _cursor_help() -> str:
    from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS

    return "\n".join(_REQUIRED_HELP_MARKERS["cursor"]) + "\n"


def _marker_run_body(*, body: str, model_receipt: str | None) -> str:
    payload: dict[str, object] = {
        "terminal_status": "completed",
        "summary": "wrote marker",
        "changed_files": [U5_MARKER_REL],
        "evidence": [],
        "scope_expansion": None,
    }
    if model_receipt is not None:
        payload["model_receipt_status"] = model_receipt
    final_text = "Wrote the marker.\n" + json.dumps(payload, separators=(",", ":"))
    events = _cursor_stream_events(final_text)
    writer = (
        "from pathlib import Path; "
        f"p=Path({json.dumps(U5_MARKER_REL)}); "
        "p.parent.mkdir(parents=True); "
        f"p.write_text({json.dumps(body)}, encoding='utf-8')"
    )
    printed = "".join(f"printf '%s\\n' {shlex.quote(line)}\n" for line in events)
    return f"python3 -c {shlex.quote(writer)}\n" + printed + "exit 0\n"


def _cursor_stream_events(final_text: str, *, is_error: bool = False) -> list[str]:
    """Representative Cursor ``--output-format stream-json`` lines."""
    session = "00000000-0000-4000-8000-000000000000"
    events: list[dict[str, object]] = [
        {"type": "system", "subtype": "init", "session_id": session},
        {
            "type": "user",
            "message": {"role": "user", "content": [{"type": "text", "text": "unit"}]},
            "session_id": session,
        },
        {
            "type": "assistant",
            "message": {
                "role": "assistant",
                "content": [{"type": "text", "text": final_text}],
            },
            "session_id": session,
        },
        {
            "type": "result",
            "subtype": "error" if is_error else "success",
            "duration_ms": 10,
            "is_error": is_error,
            "result": final_text,
            "session_id": session,
        },
    ]
    return [json.dumps(event, separators=(",", ":")) for event in events]


def _write_probe_unit(path: Path, timeout_seconds: int) -> None:
    packet = {
        "unit_id": "u5-real-probe",
        "goal": (
            "Create src/u5-marker.txt containing the ASCII bytes "
            "U5-REAL-PROBE-MARKER followed by one newline (0x0A). "
            "Do not create or modify any other path."
        ),
        "allowed_paths": [U5_MARKER_REL],
        "verification": ["test -f src/u5-marker.txt"],
        "constraints": [
            "Write only src/u5-marker.txt",
            "Exact bytes are U5-REAL-PROBE-MARKER plus a single trailing newline",
            "Leave the change uncommitted",
            "Do not commit, push, open a PR, merge, or deploy",
        ],
        "network": True,
        "external_cost": True,
        "timeout_seconds": timeout_seconds,
    }
    path.write_text(json.dumps(packet, indent=2) + "\n", encoding="utf-8")


def _cleanup_probe_runtime(repo: Path, receipt: dict[str, object] | None) -> None:
    if not receipt:
        return
    run_dir = receipt.get("run_dir")
    attempts = receipt.get("attempts")
    if isinstance(attempts, list):
        for attempt in attempts:
            if not isinstance(attempt, dict):
                continue
            worktree = attempt.get("worktree")
            if isinstance(worktree, str) and worktree:
                _git(repo, "worktree", "remove", "--force", worktree, check=False)
                shutil.rmtree(worktree, ignore_errors=True)
    if isinstance(run_dir, str) and run_dir:
        shutil.rmtree(run_dir, ignore_errors=True)


def _marker_probe_config(
    profile: str,
) -> tuple[dict[str, dict[str, object]], list[str], list[str] | None, str]:
    """Run exactly the selected profile from review; normal roles get holders."""
    selected = {
        "adapter": profile,
        "executable": U5_EXECUTABLE[profile],
        "model": "auto",
    }
    return {profile: selected}, [], [profile], "review"


def execute_marker_probe(
    profile: str,
    *,
    offline_bindir: Path | None,
    timeout_seconds: int = 20,
    retain_fixture: bool = False,
) -> dict[str, object]:
    """Run one marker unit against a single profile.

    Offline fixtures pass ``offline_bindir`` and never use the ambient PATH.
    A real profile requires RUN_REAL_AGENT_EXEC=1 and keeps its fixture.
    """
    if profile not in U5_EXECUTABLE:
        raise ValueError(f"unknown real probe profile: {profile}")
    if offline_bindir is None and not real_probe_opt_in():
        raise RuntimeError(
            "real Agent CLI probe refused without RUN_REAL_AGENT_EXEC=1"
        )

    tool_dir: tempfile.TemporaryDirectory[str] | None = None
    repo_dir: tempfile.TemporaryDirectory[str] | None = None
    repo: Path | None = None
    receipt: dict[str, object] | None = None
    try:
        if offline_bindir is None:
            probe_path = os.environ.get("PATH", "")
        else:
            tool_dir = _probe_tool_bindir(offline_bindir)
            probe_path = tool_dir.name
        if retain_fixture:
            repo_root = Path(tempfile.mkdtemp(prefix="u5-real-probe-"))
        else:
            repo_dir = tempfile.TemporaryDirectory(prefix="u5-real-probe-")
            repo_root = Path(repo_dir.name)
        repo = repo_root / "repo"
        repo.mkdir()
        receipts = repo_root / "receipts"
        receipts.mkdir()
        init = _git(repo, "init", "-b", U5_BRANCH, check=False)
        if init.returncode != 0:
            _git(repo, "init")
            _git(repo, "checkout", "-b", U5_BRANCH)
        _git(repo, "config", "user.email", "agent-exec@example.com")
        _git(repo, "config", "user.name", "Agent Exec")
        (repo / "README.md").write_text("fixture\n", encoding="utf-8")
        _git(repo, "add", "README.md")
        _git(repo, "commit", "-m", "init")
        branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
        if branch != U5_BRANCH:
            raise RuntimeError(
                f"probe fixture must use non-default branch {U5_BRANCH}, got {branch}"
            )
        profiles, role_order, review_order, probe_role = _marker_probe_config(profile)
        _write_config(
            repo,
            profiles=profiles,
            role_order=role_order,
            review_order=review_order,
            timeout_seconds=timeout_seconds,
            summary_bytes=4096,
            raw_output_bytes=65536,
        )
        before = _canonical_snapshot(repo)
        unit = repo_root / "unit.json"
        _write_probe_unit(unit, timeout_seconds)
        receipt_path = receipts / "probe.json"
        env = os.environ.copy()
        env["PYTHONDONTWRITEBYTECODE"] = "1"
        env["PATH"] = probe_path
        workspace = Path(__file__).resolve().parents[1]
        existing = env.get("PYTHONPATH", "")
        env["PYTHONPATH"] = (
            f"{workspace}{os.pathsep}{existing}" if existing else str(workspace)
        )
        proc = subprocess.run(
            [
                "python3",
                "-m",
                "scripts.agent_exec",
                "run",
                "--repo",
                str(repo),
                "--role",
                probe_role,
                "--unit",
                str(unit),
                "--receipt",
                str(receipt_path),
            ],
            cwd=str(repo_root),
            capture_output=True,
            text=True,
            env=env,
            timeout=timeout_seconds + 30,
        )
        receipt = (
            json.loads(receipt_path.read_text(encoding="utf-8"))
            if receipt_path.is_file()
            else None
        )
        outcome = _classify_marker_probe(
            profile=profile,
            repo=repo,
            before=before,
            receipt=receipt,
            exit_code=proc.returncode,
            stderr=proc.stderr,
        )
        outcome["fixture_repo"] = str(repo)
        if retain_fixture and receipt_path.is_file():
            durable = Path(f"/tmp/agent-exec-{os.getuid()}") / "u5-probes"
            durable.mkdir(mode=0o700, parents=True, exist_ok=True)
            os.chmod(durable, 0o700)
            stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
            copied = durable / f"{profile}-{stamp}.json"
            shutil.copy2(receipt_path, copied)
            outcome["source_external_receipt"] = str(copied)
        return outcome
    finally:
        if not retain_fixture and repo is not None:
            _cleanup_probe_runtime(repo, receipt if isinstance(receipt, dict) else None)
        if tool_dir is not None:
            tool_dir.cleanup()
        if repo_dir is not None:
            repo_dir.cleanup()


def _probe_config_mismatch(
    profile: str,
    profiles: object,
    roles: object,
) -> str | None:
    """Reject a fixture that does not isolate the selected adapter.

    The selected profile is the whole review role. Implementation, repair,
    and test hold only the uninvoked cursor/antigravity/PI holders, so strict
    role validation passes and the selected CLI can reach PASS.
    """
    if not isinstance(profiles, dict) or not isinstance(roles, dict):
        return "probe configured a profile other than the selection"
    if set(profiles) != {profile, *_PROBE_NORMAL_HOLDERS}:
        return "probe configured a profile other than the selection"
    for role in ("implementation", "repair", "test"):
        if roles.get(role) != _PROBE_NORMAL_HOLDERS:
            return f"{role} role must hold only the uninvoked normal holders"
    if roles.get("review") != [profile]:
        return "selected profile must run alone from the review role"
    selected = profiles.get(profile)
    if not isinstance(selected, dict) or selected.get("adapter") != profile:
        return "probe configured a profile other than the selection"
    return None


def _classify_marker_probe(
    *,
    profile: str,
    repo: Path,
    before: dict[str, str],
    receipt: dict[str, object] | None,
    exit_code: int,
    stderr: str,
) -> dict[str, object]:
    base: dict[str, object] = {
        "profile": profile,
        "exit_code": exit_code,
        "canonical_preserved": _canonical_snapshot(repo) == before,
        "marker_in_canonical": (repo / U5_MARKER_REL).exists(),
        "configured_model": None,
        "model_receipt_status": "not-observable",
        "branch": None,
        "changed_paths": [],
        "transport_parent": None,
        "transport_ref": None,
        "attempt_status": None,
        "run_terminal": None,
        "cleanup": {},
        "host_fallback_authorizations": 0,
        "base_head": before["head"],
    }
    if not base["canonical_preserved"]:
        base["capability"] = "failed"
        base["reason"] = "canonical checkout mutated"
        return base
    if base["marker_in_canonical"]:
        base["capability"] = "failed"
        base["reason"] = "canonical checkout contains the probe marker"
        return base
    if not isinstance(receipt, dict):
        base["capability"] = "failed"
        base["reason"] = f"missing receipt: {stderr[-500:]}"
        return base

    base["run_terminal"] = receipt.get("terminal_status")
    base["host_fallback_authorizations"] = receipt.get(
        "host_fallback_authorizations", 0
    )
    summary = receipt.get("config_summary")
    profiles = summary.get("profiles") if isinstance(summary, dict) else None
    roles = summary.get("roles") if isinstance(summary, dict) else None
    mismatch = _probe_config_mismatch(profile, profiles, roles)
    if mismatch is not None:
        base["capability"] = "failed"
        base["reason"] = mismatch
        return base
    assert isinstance(profiles, dict)
    selected = profiles[profile]
    if isinstance(selected, dict):
        base["configured_model"] = selected.get("model")
    recorded_base = receipt.get("base")
    if isinstance(recorded_base, dict):
        base["branch"] = recorded_base.get("branch")
    if base["branch"] in {"main", "master", "HEAD", None}:
        base["capability"] = "failed"
        base["reason"] = f"probe branch is not a non-default branch: {base['branch']}"
        return base

    attempts = receipt.get("attempts")
    if not isinstance(attempts, list) or len(attempts) != 1:
        base["capability"] = "failed"
        base["reason"] = "probe must record exactly one attempt"
        return base
    attempt = attempts[0]
    if not isinstance(attempt, dict) or attempt.get("profile") != profile:
        base["capability"] = "failed"
        base["reason"] = "attempt profile does not match the selection"
        return base
    base["attempt_status"] = attempt.get("terminal_status")
    base["changed_paths"] = list(attempt.get("changed_paths") or [])
    base["transport_ref"] = attempt.get("transport_ref")
    cleanup = attempt.get("cleanup") if isinstance(attempt.get("cleanup"), dict) else {}
    base["cleanup"] = {
        "exact_restore": bool(cleanup.get("exact_restore")),
        "transport_created": bool(cleanup.get("transport_created")),
        "workspace_preserved": bool(attempt.get("workspace_preserved")),
        "skipped": bool(cleanup.get("skipped")),
    }
    base["model_receipt_status"] = _observe_model_receipt(attempt)
    reason = attempt.get("reason") or receipt.get("reason")

    if attempt.get("terminal_status") == "unavailable":
        base["capability"] = "unavailable"
        base["reason"] = reason or "adapter unavailable"
        if base["transport_ref"] or base["cleanup"]["transport_created"]:
            base["capability"] = "failed"
            base["reason"] = "unavailable probe still created transport"
        return base

    if (
        receipt.get("terminal_status") == "success"
        and attempt.get("terminal_status") == "success"
    ):
        return _classify_successful_marker(
            repo=repo,
            receipt=receipt,
            attempt=attempt,
            before=before,
            base=base,
        )

    changed = list(attempt.get("changed_paths") or [])
    if (
        base["canonical_preserved"]
        and not base["marker_in_canonical"]
        and not attempt.get("transport_ref")
        and not changed
        and attempt.get("terminal_status")
        in {"failed-clean", "failed-unknown", "unavailable"}
    ):
        base["capability"] = "unavailable"
        base["reason"] = reason or "probe produced no marker"
        return base

    base["capability"] = "failed"
    base["reason"] = reason or f"unclassified probe status {attempt.get('terminal_status')}"
    return base


def _classify_successful_marker(
    *,
    repo: Path,
    receipt: dict[str, object],
    attempt: dict[str, object],
    before: dict[str, str],
    base: dict[str, object],
) -> dict[str, object]:
    if base["model_receipt_status"] == "mismatch":
        base["capability"] = "failed"
        base["reason"] = "model receipt mismatch"
        return base
    if base["configured_model"] is not None:
        base["capability"] = "failed"
        base["reason"] = "auto model must omit the configured model value"
        return base
    changed = list(attempt.get("changed_paths") or [])
    if changed != [U5_MARKER_REL]:
        base["capability"] = "failed"
        base["reason"] = f"changed paths are not exactly the marker: {changed}"
        return base
    cleanup = base["cleanup"]
    if not isinstance(cleanup, dict):
        base["capability"] = "failed"
        base["reason"] = "missing cleanup record"
        return base
    if cleanup.get("exact_restore") or not cleanup.get("transport_created"):
        base["capability"] = "failed"
        base["reason"] = "success must retain transport and skip exact restore"
        return base
    if not cleanup.get("workspace_preserved"):
        base["capability"] = "failed"
        base["reason"] = "success must retain the isolated workspace until host review"
        return base
    transport = receipt.get("transport")
    ref = attempt.get("transport_ref")
    if not isinstance(transport, dict) or not isinstance(ref, str) or not ref:
        base["capability"] = "failed"
        base["reason"] = "success is missing a transport ref"
        return base
    if transport.get("ref") != ref:
        base["capability"] = "failed"
        base["reason"] = "receipt transport ref does not match the attempt"
        return base
    commit = _git(repo, "rev-parse", ref).stdout.strip()
    parents = _git(repo, "rev-list", "--parents", "-n", "1", commit).stdout.split()
    if len(parents) != 2:
        base["capability"] = "failed"
        base["reason"] = f"transport commit is not base-parented: {parents}"
        return base
    parent = parents[1]
    base_head = before["head"]
    recorded = receipt.get("base")
    recorded_head = recorded.get("head") if isinstance(recorded, dict) else None
    if parent != base_head or transport.get("parent") != base_head or recorded_head != base_head:
        base["capability"] = "failed"
        base["reason"] = "transport parent is not the fixture base HEAD"
        return base
    blob = _git(repo, "cat-file", "-p", f"{commit}:{U5_MARKER_REL}").stdout
    worktree = attempt.get("worktree")
    live = ""
    if isinstance(worktree, str):
        marker_path = Path(worktree) / U5_MARKER_REL
        if marker_path.is_file():
            live = marker_path.read_text(encoding="utf-8")
    if blob != U5_MARKER_BODY or live != U5_MARKER_BODY:
        base["capability"] = "failed"
        base["reason"] = "marker bytes differ from the required probe content"
        base["marker_bytes"] = blob
        return base
    base["transport_parent"] = parent
    base["marker_bytes"] = blob
    base["capability"] = "PASS"
    base["reason"] = None
    return base


class RealAgentCliProbeTests(unittest.TestCase):
    """U5 marker probe. Real providers run only behind RUN_REAL_AGENT_EXEC=1."""

    def test_profile_selector_accepts_exactly_one_profile(self) -> None:
        self.assertEqual(U5_PROFILES, tuple(U5_EXECUTABLE))
        self.assertFalse(real_probe_opt_in({"RUN_REAL_AGENT_EXEC": "true"}))
        self.assertFalse(real_probe_opt_in({}))
        self.assertTrue(real_probe_opt_in({"RUN_REAL_AGENT_EXEC": "1"}))
        self.assertEqual(
            require_single_real_profile({"AGENT_EXEC_REAL_PROFILE": "pi"}),
            "pi",
        )
        for raw in ("", " cursor", "cursor ", "cursor,codex", "cursor codex", "nope"):
            with self.assertRaises(ValueError):
                require_single_real_profile({"AGENT_EXEC_REAL_PROFILE": raw})

    def test_real_probe_refuses_without_opt_in(self) -> None:
        self.assertNotEqual(os.environ.get("RUN_REAL_AGENT_EXEC"), "1")
        with self.assertRaises(RuntimeError) as caught:
            execute_marker_probe("cursor", offline_bindir=None)
        self.assertIn("RUN_REAL_AGENT_EXEC=1", str(caught.exception))

    def test_offline_success_checks_marker_transport_and_canonical(self) -> None:
        outcome = self._run_cursor(
            _marker_run_body(body=U5_MARKER_BODY, model_receipt=None)
        )
        self.assertEqual(outcome["capability"], "PASS", outcome)
        self.assertEqual(outcome["marker_bytes"], U5_MARKER_BODY)
        self.assertEqual(outcome["changed_paths"], [U5_MARKER_REL])
        self.assertEqual(outcome["transport_parent"], outcome["base_head"])
        self.assertTrue(outcome["transport_parent"])
        self.assertEqual(outcome["branch"], U5_BRANCH)
        self.assertIsNone(outcome["configured_model"])
        self.assertEqual(outcome["model_receipt_status"], "not-observable")
        self.assertEqual(outcome["run_terminal"], "success")
        cleanup = outcome["cleanup"]
        self.assertIsInstance(cleanup, dict)
        assert isinstance(cleanup, dict)
        self.assertTrue(cleanup["transport_created"])
        self.assertTrue(cleanup["workspace_preserved"])
        self.assertFalse(cleanup["exact_restore"])
        self.assertTrue(outcome["canonical_preserved"])
        self.assertFalse(outcome["marker_in_canonical"])
        self.assertEqual(outcome["host_fallback_authorizations"], 0)

    def test_observable_verified_model_receipt_stays_attached_to_success(self) -> None:
        outcome = self._run_cursor(
            _marker_run_body(body=U5_MARKER_BODY, model_receipt="verified")
        )
        self.assertEqual(outcome["capability"], "PASS", outcome)
        self.assertEqual(outcome["model_receipt_status"], "verified")

    def test_model_receipt_mismatch_is_not_pass(self) -> None:
        outcome = self._run_cursor(
            _marker_run_body(body=U5_MARKER_BODY, model_receipt="mismatch")
        )
        self.assertEqual(outcome["capability"], "failed", outcome)
        self.assertNotEqual(outcome["capability"], "PASS")
        self.assertEqual(outcome["model_receipt_status"], "mismatch")

    def test_wrong_marker_is_not_pass(self) -> None:
        outcome = self._run_cursor(
            _marker_run_body(body="WRONG\n", model_receipt=None)
        )
        self.assertEqual(outcome["capability"], "failed", outcome)
        self.assertNotEqual(outcome["capability"], "PASS")
        self.assertTrue(outcome["canonical_preserved"])

    def test_missing_opencode_is_unavailable_not_pass(self) -> None:
        outcome = self._run_offline("opencode", help_text="", run_body="exit 0\n", install=False)
        self.assertEqual(outcome["capability"], "unavailable", outcome)
        self.assertNotEqual(outcome["capability"], "PASS")
        self.assertEqual(outcome["attempt_status"], "unavailable")
        self.assertEqual(outcome["run_terminal"], "unavailable")
        self.assertEqual(outcome["host_fallback_authorizations"], 0)
        self.assertFalse(outcome["marker_in_canonical"])
        self.assertTrue(outcome["canonical_preserved"])
        cleanup = outcome["cleanup"]
        assert isinstance(cleanup, dict)
        self.assertTrue(cleanup["skipped"])
        self.assertFalse(cleanup["transport_created"])

    def test_incompatible_opencode_is_unavailable_not_pass(self) -> None:
        outcome = self._run_offline(
            "opencode",
            help_text="opencode\nusage: opencode\n",
            run_body=(
                "mkdir -p src\n"
                f"printf '%s\\n' 'SHOULD-NOT-RUN' > {U5_MARKER_REL}\n"
                "exit 0\n"
            ),
            install=True,
        )
        self.assertEqual(outcome["capability"], "unavailable", outcome)
        self.assertNotEqual(outcome["capability"], "PASS")
        self.assertEqual(outcome["attempt_status"], "unavailable")
        self.assertIn("missing required capability", str(outcome["reason"]))
        self.assertFalse(outcome["marker_in_canonical"])
        self.assertNotEqual(outcome["run_terminal"], "success")

    def test_launched_failure_without_marker_is_unavailable_not_pass(self) -> None:
        outcome = self._run_cursor("echo 'auth failed' >&2\nexit 3\n")
        self.assertEqual(outcome["capability"], "unavailable", outcome)
        self.assertNotEqual(outcome["capability"], "PASS")
        self.assertNotEqual(outcome["run_terminal"], "success")
        self.assertEqual(outcome["host_fallback_authorizations"], 0)
        self.assertTrue(outcome["canonical_preserved"])
        self.assertFalse(outcome["marker_in_canonical"])

    @unittest.skipUnless(
        real_probe_opt_in(),
        "real CLI probe NOT_RUN: set RUN_REAL_AGENT_EXEC=1 and "
        "AGENT_EXEC_REAL_PROFILE to exactly one profile",
    )
    def test_opt_in_real_profile_probe(self) -> None:
        profile = require_single_real_profile(dict(os.environ))
        outcome = execute_marker_probe(
            profile,
            offline_bindir=None,
            timeout_seconds=1800,
            retain_fixture=True,
        )
        print("U5-PROBE " + json.dumps(outcome, sort_keys=True, default=str))
        if profile == "opencode" and outcome["capability"] == "unavailable":
            self.assertNotEqual(outcome["capability"], "PASS")
            self.assertEqual(outcome["attempt_status"], "unavailable")
            self.assertFalse(outcome["marker_in_canonical"])
            self.assertTrue(outcome["canonical_preserved"])
            return
        self.assertEqual(outcome["capability"], "PASS", outcome)
        self.assertEqual(outcome["marker_bytes"], U5_MARKER_BODY)
        self.assertEqual(outcome["changed_paths"], [U5_MARKER_REL])
        self.assertTrue(outcome["transport_parent"])
        self.assertEqual(outcome["branch"], U5_BRANCH)
        self.assertNotEqual(outcome["model_receipt_status"], "mismatch")
        self.assertTrue(outcome["canonical_preserved"])

    def _run_cursor(self, run_body: str) -> dict[str, object]:
        return self._run_offline(
            "cursor",
            help_text=_cursor_help(),
            run_body=run_body,
            install=True,
        )

    def _run_offline(
        self,
        profile: str,
        *,
        help_text: str,
        run_body: str,
        install: bool,
    ) -> dict[str, object]:
        fake = tempfile.TemporaryDirectory(prefix="u5-fake-cli-")
        self.addCleanup(fake.cleanup)
        bindir = Path(fake.name)
        if install:
            _write_probe_double(
                bindir,
                U5_EXECUTABLE[profile],
                help_text=help_text,
                run_body=run_body,
            )
        return execute_marker_probe(
            profile,
            offline_bindir=bindir,
            timeout_seconds=20,
            retain_fixture=False,
        )


if __name__ == "__main__":
    unittest.main()