weichen
8f11d002ae
Why: provide isolated, recoverable per-unit CLI execution and safe host fallback Evidence: 43 focused tests, configuration validation, governance regression, and diff check Impact: adds agent-exec validate/run controller, receipts, transports, and Make entrypoint Refs: U3; docs/superpowers/plans/2026-10-15-project-agent-execution.md
1373 lines
51 KiB
Python
1373 lines
51 KiB
Python
"""End-to-end execution controller contract for agent-exec (U3)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import textwrap
|
|
import time
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from scripts import agent_exec
|
|
|
|
|
|
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 _write_config(
|
|
repo: Path,
|
|
*,
|
|
profiles: dict[str, dict[str, object]],
|
|
role_order: list[str],
|
|
timeout_seconds: int = 30,
|
|
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]
|
|
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}
|
|
summary_bytes: {summary_bytes}
|
|
raw_output_bytes: {raw_output_bytes}
|
|
|
|
roles:
|
|
implementation: [{order}]
|
|
repair: [{first}]
|
|
test: [{first}]
|
|
review: [{first}]
|
|
|
|
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
|
|
|
|
|
|
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_authorizes_host_fallback_once(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_HOST_FALLBACK, proc.stderr + proc.stdout
|
|
)
|
|
data = json.loads(receipt.read_text(encoding="utf-8"))
|
|
self.assertEqual(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"), 1)
|
|
# Controller must not implement the unit itself.
|
|
after = list(self.repo.rglob("*"))
|
|
self.assertEqual(before, after)
|
|
self.assertFalse((self.repo / "src" / "marker.txt").exists())
|
|
self.assertIn("allowed_paths", data["host_scope"])
|
|
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_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"))
|
|
if child_pid_file.exists():
|
|
child = int(child_pid_file.read_text(encoding="utf-8").strip())
|
|
# Child must not still be alive.
|
|
still = subprocess.run(
|
|
["kill", "-0", str(child)],
|
|
check=False,
|
|
capture_output=True,
|
|
)
|
|
self.assertNotEqual(still.returncode, 0)
|
|
self._assert_canonical_untouched()
|
|
|
|
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 worktree clean is blocked."""
|
|
launched = self.root / "canon-next.log"
|
|
launched.write_text("", encoding="utf-8")
|
|
# Resolve the main checkout from the linked worktree .git pointer.
|
|
_write_fake_worker(
|
|
self.bindir,
|
|
"canonmut",
|
|
body=textwrap.dedent(
|
|
"""\
|
|
workspace="$1"
|
|
common=$(git -C "$workspace" rev-parse --path-format=absolute --git-common-dir)
|
|
canonical=$(dirname "$common")
|
|
printf 'canonical-mutated\\n' >> "$canonical/README.md"
|
|
git -C "$canonical" add README.md
|
|
# Detached worktree 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 worktree .git pointer must produce a stable blocked receipt."""
|
|
_write_fake_worker(
|
|
self.bindir,
|
|
"corruptgit",
|
|
body=textwrap.dedent(
|
|
"""\
|
|
workspace="$1"
|
|
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_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()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|