weichen/Agent-System-Dev
1
0
Fork 0
Code Issues 1 Pull Requests 3 Actions 2 Packages Projects Releases Wiki Activity
Files
818e17f4216fa3253d16c886e37c10a04408334c
Agent-System-Dev/tests/test_agent_exec.py
T
weichen 818e17f421 Remove host fallback and enforce strict normal routes 
Why: The approved contract requires exactly three safe attempts for Cursor, agy, and PI, then one Codex emergency attempt, never host implementation authority.

Evidence: Routing/config tests were RED for never, max attempts, forbidden adapters, and legacy host fallback; 133 related tests now pass with one opt-in probe skipped.

Impact: Normal emergency-mode roles reject all non-Cursor/agy/PI adapters, project fallback is never, and Codex remains available through emergency/review paths.

Refs: independent Codex review of feature/agent-cli-enforcement
2026-10-15 18:45:08 +00:00

3259 lines
125 KiB
Python
Raw Blame History

"""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
)
def _write_config(
repo: Path,
*,
profiles: dict[str, dict[str, object]],
role_order: list[str],
review_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:
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")
order = ", ".join(role_order)
first = role_order[0]
review = ", ".join(review_order) if review_order is not None else first
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: [{order}]
repair: [{first}]
test: [{first}]
review: [{review}]
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
def _write_fake_worker(
bindir: Path,
name: str,
*,
body: str,
help_text: str = "custom worker\n",
) -> Path:
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"
+ body,
encoding="utf-8",
)
path.chmod(path.stat().st_mode | stat.S_IEXEC)
return path
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"
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-a",
body=textwrap.dedent(
f"""\
workspace="$1"
echo "$$" > "{meta}/a.pid"
echo "$workspace" > "{meta}/a.wt"
# fail clean after no writes
exit 7
"""
),
)
_write_fake_worker(
self.bindir,
"worker-b",
body=textwrap.dedent(
f"""\
workspace="$1"
result="$3"
echo "$$" > "{meta}/b.pid"
echo "$workspace" > "{meta}/b.wt"
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={
"a": {"adapter": "custom", "executable": "worker-a"},
"b": {"adapter": "custom", "executable": "worker-b"},
},
role_order=["a", "b"],
)
self.base_snap = _canonical_snapshot(self.repo)
unit = _write_unit(self.root / "unit.json")
receipt = self.receipts / "fresh.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")
a_pid = (meta / "a.pid").read_text(encoding="utf-8").strip()
b_pid = (meta / "b.pid").read_text(encoding="utf-8").strip()
a_wt = (meta / "a.wt").read_text(encoding="utf-8").strip()
b_wt = (meta / "b.wt").read_text(encoding="utf-8").strip()
self.assertNotEqual(a_pid, b_pid)
self.assertNotEqual(a_wt, b_wt)
self.assertTrue(a_wt.startswith(f"/tmp/agent-exec-{os.getuid()}/"))
self.assertTrue(b_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": "custom", "executable": "does-not-exist-xyz"},
"second": {"adapter": "custom", "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": "custom", "executable": "gone-one"},
"m2": {"adapter": "custom", "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(len(data["attempts"]), 2)
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": "custom", "executable": "gone-one"},
"m2": {"adapter": "custom", "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_failed_clean_can_continue_after_exact_restore(self) -> None:
_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(
"""\
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={
"dirty": {"adapter": "custom", "executable": "dirty"},
"ok": {"adapter": "custom", "executable": "ok"},
},
role_order=["dirty", "ok"],
)
unit = _write_unit(self.root / "unit.json")
receipt = self.receipts / "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_SUCCESS, proc.stderr + proc.stdout)
data = json.loads(receipt.read_text(encoding="utf-8"))
self.assertEqual(data["attempts"][0]["terminal_status"], "failed-clean")
self.assertTrue(data["attempts"][0]["cleanup"]["exact_restore"])
self.assertEqual(data["terminal_status"], "success")
self._assert_canonical_untouched()
def test_failed_clean_retries_profile_before_next_success(self) -> None:
launches = self.root / "profile-launches.log"
launches.write_text("", encoding="utf-8")
_write_fake_worker(
self.bindir,
"retryme",
body=textwrap.dedent(
f"""\
printf 'retryme\\n' >> "{launches}"
workspace="$1"
mkdir -p "$workspace/src"
printf 'tmp\\n' > "$workspace/src/marker.txt"
exit 9
"""
),
)
_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": "custom", "executable": "retryme"},
"thenok": {"adapter": "custom", "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, ["retryme", "retryme", "retryme", "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, "failed-clean"),
("retryme", 2, "failed-clean"),
("retryme", 3, "failed-clean"),
("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_failed_clean(self) -> None:
launches = self.root / "emergency-launches.log"
launches.write_text("", encoding="utf-8")
_write_fake_worker(
self.bindir,
"regular",
help_text=_pi_help(),
body=textwrap.dedent(
f"""\
printf 'regular\\n' >> "{launches}"
mkdir -p src
printf 'tmp\\n' > src/marker.txt
printf '%s\\n' '{{"terminal_status":"failed","summary":"regular failed clean","changed_files":["src/marker.txt"]}}'
exit 9
"""
),
)
_write_fake_worker(
self.bindir,
"codex",
help_text=_codex_help(),
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
"""
),
)
self._configure(
profiles={
"regular": {"adapter": "pi", "executable": "regular", "model": "auto"},
"codex": {"adapter": "codex", "executable": "codex", "model": "auto"},
},
role_order=["regular"],
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)
self.assertEqual(loaded.profiles["regular"].adapter, "pi")
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,
[
("regular", 1, "failed-clean", False),
("regular", 2, "failed-clean", False),
("regular", 3, "failed-clean", False),
("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, ["regular", "regular", "regular", "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": "custom", "executable": "unknown"},
"next": {"adapter": "custom", "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": "custom", "executable": "scope"},
"next": {"adapter": "custom", "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": "custom", "executable": worker},
"next": {
"adapter": "custom",
"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": "custom", "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": "custom", "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": "custom", "executable": "silent"},
"idlenext": {"adapter": "custom", "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": "custom", "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": "custom", "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": "custom", "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": "custom", "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": "custom", "executable": "canonmut"},
"canonnext": {"adapter": "custom", "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": "custom", "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": "custom", "executable": "netfail"},
"netnext": {"adapter": "custom", "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": "custom", "executable": "adminmut"},
"adminnext": {"adapter": "custom", "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": "custom", "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": "custom", "executable": "symworker"},
"symnext": {"adapter": "custom", "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": "custom", "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.
pad = "P" * (128 * 1024)
_write_fake_worker(
self.bindir,
"pipefill",
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=["pipefill"],
timeout_seconds=15,
raw_output_bytes=1024,
)
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",
"implementation",
"--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": "custom", "executable": "txok"},
"txnext": {"adapter": "custom", "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": "custom", "executable": "renamer"},
"renamenext": {"adapter": "custom", "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_unproven_restore_is_failed_unknown_and_stops(self) -> None:
"""failed-clean is allowed only after _restore_worktree succeeds."""
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": "custom",
"executable": f"restorer-{label}",
},
"restorenext": {
"adapter": "custom",
"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"
with mock.patch.object(
agent_exec, "_restore_worktree", return_value=False
):
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": "custom", "executable": "netclean"},
"netcleannext": {"adapter": "custom", "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": "custom",
"executable": f"failstop-{label}",
},
"failnext": {
"adapter": "custom",
"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()
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": "custom",
"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": "custom",
"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": "custom",
"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": "custom", "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["attestation"], record["attestation"])
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_attestation_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-role validation rejects every other adapter on implementation/repair/test.
# The probe still has to launch those adapters, so they sit only on review.
_PROBE_NORMAL_HOLDER = "piholder"
U5_BRANCH = "feature/u5-real-probe"
U5_MARKER_REL = "src/u5-marker.txt"
U5_MARKER_BODY = "U5-REAL-PROBE-MARKER\n"
_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
encoded = json.dumps(payload, separators=(",", ":"))
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')"
)
return (
f"python3 -c {shlex.quote(writer)}\n"
f"printf '%s\\n' {shlex.quote(encoded)}\n"
"exit 0\n"
)
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]:
"""Place one profile where strict normal-role validation allows it to run."""
selected = {
"adapter": profile,
"executable": U5_EXECUTABLE[profile],
"model": "auto",
}
if profile in NORMAL_ROLE_ADAPTERS:
return {profile: selected}, [profile], None, "implementation"
return (
{
_PROBE_NORMAL_HOLDER: {
"adapter": "pi",
"executable": "pi-role-holder",
"model": "auto",
},
profile: selected,
},
[_PROBE_NORMAL_HOLDER],
[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.
Normal adapters stay on implementation. Codex and other non-normal adapters
stay on review, beside an uninvoked pi holder, so strict role validation
still 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"
implementation = roles.get("implementation")
if profile in NORMAL_ROLE_ADAPTERS:
if set(profiles) != {profile}:
return "probe configured a profile other than the selection"
if implementation != [profile]:
return "implementation role is not exactly the selected profile"
return None
if set(profiles) != {profile, _PROBE_NORMAL_HOLDER}:
return "probe configured a profile other than the selection"
if implementation != [_PROBE_NORMAL_HOLDER] or roles.get("review") != [profile]:
return "non-normal adapter must run only 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()
Reference in New Issue View Git Blame Copy Permalink