Constrain controlled CLI sessions and prove native budget overshoot

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

tests/test_native_cli_boundary.py

@@ -0,0 +1,148 @@
+"""Opt-in installed Claude binary against a namespace-local synthetic API."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from llm_connect.models import RunConfig
+
+from rein_aharness.adapter import AgenticClaudeCodeAdapter
+from rein_aharness.native_limits import NativeLimitError, terminal_accounting
+
+pytestmark = pytest.mark.skipif(
+    os.environ.get("REIN_REAL_CLAUDE") != "1",
+    reason="opt-in installed CLI boundary proof",
+)
+
+
+@pytest.mark.parametrize("case", ["overrun", "tools"])
+def test_installed_cli_boundary(tmp_path, case):
+    binary = Path(
+        os.environ.get("REIN_CLAUDE_PROOF_BINARY") or shutil.which("claude") or ""
+    ).resolve(strict=True)
+    version = subprocess.check_output([str(binary), "--version"], text=True).strip()
+    assert version.startswith("2.1.266 "), (
+        "revalidate the fixture for this binary version before claiming the pinned proof"
+    )
+    adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, model="claude-sonnet-4-6")
+    cap = 0.01 if case == "overrun" else 1
+    argv = adapter._build_command(
+        RunConfig(model_params={"max_budget_usd": cap, "max_turns": 4})
+    )
+    argv[0] = "/opt/claude"
+    (tmp_path / "case.json").write_text(json.dumps({"case": case, "argv": argv}))
+    (tmp_path / "passwd").write_text(
+        f"fixture:x:{os.getuid()}:{os.getgid()}:Fixture:/work/home:/bin/sh\n"
+    )
+    (tmp_path / "group").write_text(f"fixture:x:{os.getgid()}:\n")
+    (tmp_path / "home").mkdir()
+    shutil.copyfile(
+        Path(__file__).with_name("native_cli_fixture.py"), tmp_path / "fixture.py"
+    )
+    command = [
+        "bwrap",
+        "--unshare-all",
+        "--die-with-parent",
+        "--new-session",
+        "--ro-bind",
+        "/usr",
+        "/usr",
+        "--ro-bind",
+        "/lib",
+        "/lib",
+        "--ro-bind",
+        "/lib64",
+        "/lib64",
+        "--symlink",
+        "usr/bin",
+        "/bin",
+        "--proc",
+        "/proc",
+        "--dev",
+        "/dev",
+        "--tmpfs",
+        "/tmp",
+        "--dir",
+        "/etc",
+        "--ro-bind",
+        str(tmp_path / "passwd"),
+        "/etc/passwd",
+        "--ro-bind",
+        str(tmp_path / "group"),
+        "/etc/group",
+        "--dir",
+        "/opt",
+        "--ro-bind",
+        str(binary),
+        "/opt/claude",
+        "--bind",
+        str(tmp_path),
+        "/work",
+        "--chdir",
+        "/work",
+        "--clearenv",
+        "--setenv",
+        "PATH",
+        "/usr/bin:/bin",
+        "/usr/bin/python3",
+        "/work/fixture.py",
+    ]
+    proc = subprocess.run(command, capture_output=True, text=True, timeout=55, check=False)
+    assert proc.returncode == 0, proc.stderr[-1000:]
+    result = json.loads(proc.stdout)
+    assert result["network_namespace"] != os.readlink("/proc/self/ns/net")
+    assert result["markers"] == {
+        "hook-ran": False,
+        "mcp-ran": False,
+        "denied-ran": False,
+    }
+    requests = [row for row in result["requests"] if "count_tokens" not in row["path"]]
+    expected_tools = {"Read", "Write", "Edit", "Glob", "Grep", "Bash"}
+    for request in requests:
+        # The CLI may add its terminal EndConversation tool; it cannot add work tools.
+        assert set(request["tools"]) - {"EndConversation"} <= expected_tools
+        assert {"Bash", "Read", "Edit"} <= set(request["tools"])
+        assert request["model"] == "claude-sonnet-4-6"
+        assert not request["ambient_context_loaded"]
+    terminal = result["terminal"]
+    if case == "overrun":
+        assert len(requests) == 1
+        assert proc.returncode == 0 and result["returncode"] == 1
+        assert terminal["subtype"] == "error_max_budget_usd"
+        assert terminal["total_cost_usd"] == pytest.approx(0.18015)
+        assert terminal["total_cost_usd"] > cap
+        with pytest.raises(NativeLimitError) as caught:
+            terminal_accounting(terminal, max_budget_usd=cap, max_turns=4)
+        assert caught.value.cost_usd == terminal["total_cost_usd"]
+    else:
+        assert result["returncode"] == 0 and terminal["subtype"] == "success"
+        assert len(requests) == 2
+        assert {"id": "tool_allowed", "is_error": False} in requests[1]["tool_results"]
+        assert {"id": "tool_denied", "is_error": True} in requests[1]["tool_results"]
+        assert {"id": "tool_create", "is_error": False} in requests[1]["tool_results"]
+        assert result["created_file"] == "fixture-created\n"
+        usage, cost = terminal_accounting(terminal, max_budget_usd=cap, max_turns=4)
+        assert usage["total_tokens"] > 0 and cost > 0
+    if output_dir := os.environ.get("REIN_CLAUDE_PROOF_OUTPUT"):
+        target = Path(output_dir)
+        target.mkdir(parents=True, exist_ok=True)
+        (target / f"{case}.json").write_text(
+            json.dumps(
+                {
+                    "case": case,
+                    "binary_version": version,
+                    "binary_sha256": hashlib.sha256(binary.read_bytes()).hexdigest(),
+                    "argv": argv,
+                    "scope": "synthetic API in isolated network namespace; no real credential or provider request",
+                    **result,
+                },
+                indent=2,
+            )
+            + "\n"
+        )