WeiCHEN
47fdeb1655
Replace Linux /proc hard dependency with Darwin ps+sysctl while keeping zombie/group/boot three-value semantics fail-closed for unreadable PIDs.
755 lines
42 KiB
Python
755 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""Black-box agentctl regressions using isolated real Git repositories."""
|
|
|
|
import json
|
|
import fcntl
|
|
import hashlib
|
|
import os
|
|
import pathlib
|
|
import select
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl.py"
|
|
|
|
|
|
class AgentctlTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="agentctl-test-")
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = pathlib.Path(self.temp.name) / "repo"
|
|
self.root.mkdir()
|
|
self.git("init", "-b", "dev")
|
|
self.git("config", "user.name", "Agentctl Test")
|
|
self.git("config", "user.email", "agentctl-test@example.invalid")
|
|
(self.root / "scripts").mkdir()
|
|
shutil.copy2(SOURCE, self.root / "scripts" / "agentctl.py")
|
|
(self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
|
|
for directory in ("src/a", "src/b", ".doc/ops", ".doc/status"):
|
|
(self.root / directory).mkdir(parents=True, exist_ok=True)
|
|
self.current_rows = {
|
|
"alpha": "| alpha | Alice | feature/alpha | active |",
|
|
"beta": "| beta | Bob | feature/beta | active |",
|
|
}
|
|
(self.root / ".doc/ops/CURRENT.md").write_text(
|
|
"# Coordination\n\n| Task | Owner | Branch | State |\n| --- | --- | --- | --- |\n"
|
|
+ "\n".join(self.current_rows.values()) + "\n", encoding="utf-8")
|
|
(self.root / ".doc/status/FEATURE-MATRIX.md").write_text(
|
|
"# Features\n\n| Feature | State |\n| --- | --- |\n| core | planned |\n", encoding="utf-8")
|
|
self.git("add", ".")
|
|
self.git("commit", "-m", "test fixture")
|
|
|
|
def git(self, *args, cwd=None):
|
|
return subprocess.run(["git", *args], cwd=cwd or self.root,
|
|
text=True, capture_output=True, check=True).stdout.strip()
|
|
|
|
def cli(self, *args, cwd=None, check=True, env=None):
|
|
result = subprocess.run([sys.executable, str(self.root / "scripts/agentctl.py"), *args],
|
|
cwd=cwd or self.root, text=True, capture_output=True,
|
|
env=env, timeout=20)
|
|
if check:
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
return result
|
|
|
|
def initialized_task(self, task="alpha", scopes=("src/a", ".doc"), capabilities=()):
|
|
if not (self.root / ".locks").exists():
|
|
self.cli("init")
|
|
path = self.root / ".worktrees" / task
|
|
self.git("worktree", "add", "-b", "feature/" + task, str(path), "dev")
|
|
args = ["task", "register", "--id", task, "--owner", "Test Agent"]
|
|
for scope in scopes:
|
|
args.extend(("--scope", scope))
|
|
for capability in capabilities:
|
|
args.extend(("--capability", capability))
|
|
self.cli(*args, cwd=path)
|
|
return path
|
|
|
|
def run_lease(self, task, resources, code="pass", cwd=None, env=None):
|
|
args = ["with", "--task", task, "--wait", "0"]
|
|
if cwd is not None:
|
|
args.extend(("--cwd", str(cwd)))
|
|
for resource in resources:
|
|
args.extend(("--resource", resource))
|
|
return self.cli(*args, "--", sys.executable, "-c", code,
|
|
cwd=cwd, check=False, env=env)
|
|
|
|
def live_lease(self, task, resources, cwd):
|
|
"""Wait for the actual supervised child via a socket, never timing sleeps."""
|
|
listener = socket.socket(socket.AF_UNIX)
|
|
self.addCleanup(listener.close)
|
|
socket_path = str(pathlib.Path(self.temp.name) / ("control-" + str(len(self._cleanups))))
|
|
listener.bind(socket_path)
|
|
listener.listen(1)
|
|
listener.settimeout(15)
|
|
code = (
|
|
"import json,os,socket; s=socket.socket(socket.AF_UNIX); "
|
|
f"s.connect({socket_path!r}); "
|
|
"s.sendall((json.dumps({'run':os.environ.get('AGENTCTL_RUN_ID'),"
|
|
"'task':os.environ.get('AGENTCTL_TASK_ID'),'pid':os.getpid(),"
|
|
"'credentials':{k:v for k,v in os.environ.items() if k.startswith('AGENTCTL_')}})+'\\n').encode()); "
|
|
"s.recv(1)"
|
|
)
|
|
args = [sys.executable, str(self.root / "scripts/agentctl.py"), "with", "--task", task]
|
|
for resource in resources:
|
|
args.extend(("--resource", resource))
|
|
process = subprocess.Popen(args + ["--", sys.executable, "-c", code], cwd=cwd,
|
|
text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
def cleanup():
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
try:
|
|
process.communicate(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.communicate(timeout=5)
|
|
self.addCleanup(cleanup)
|
|
try:
|
|
connection, _ = listener.accept()
|
|
except TimeoutError:
|
|
if process.poll() is not None:
|
|
out, err = process.communicate()
|
|
self.fail("supervised child did not start: " + out + err)
|
|
raise
|
|
connection.settimeout(10)
|
|
self.addCleanup(connection.close)
|
|
payload = b""
|
|
while not payload.endswith(b"\n"):
|
|
payload += connection.recv(4096)
|
|
return process, connection, json.loads(payload)
|
|
|
|
def finish_lease(self, process, connection):
|
|
connection.sendall(b"x")
|
|
out, err = process.communicate(timeout=10)
|
|
self.assertEqual(process.returncode, 0, out + err)
|
|
|
|
def docs_request(self, changes, request_id="request-alpha"):
|
|
path = pathlib.Path(self.temp.name) / (request_id + ".json")
|
|
path.write_text(json.dumps({"request_id": request_id, "changes": changes}), encoding="utf-8")
|
|
return str(path)
|
|
|
|
def doc_change(self, file, record, old, replacement):
|
|
return {"file": file, "record_id": record,
|
|
"expected_sha256": hashlib.sha256(old.encode()).hexdigest() if old else None,
|
|
"replacement": replacement}
|
|
|
|
def test_uninitialized_status_and_doctor_do_not_create_state(self):
|
|
for command in ("status", "doctor"):
|
|
result = self.cli(command, check=False)
|
|
json.loads(result.stdout or result.stderr)
|
|
self.assertFalse((self.root / ".locks").exists())
|
|
self.assertFalse((self.root / ".git/agent-control").exists())
|
|
|
|
def test_init_preserves_existing_legacy_lock_inode(self):
|
|
legacy = self.root / ".git/agent-control/dev-integration.lock"
|
|
legacy.parent.mkdir()
|
|
legacy.write_text("")
|
|
original_inode = legacy.stat().st_ino
|
|
self.cli("init")
|
|
stable = self.root / ".locks/os/dev-integration.lock"
|
|
self.assertEqual(legacy.stat().st_ino, original_inode)
|
|
self.assertEqual(stable.stat().st_ino, original_inode)
|
|
self.cli("init")
|
|
self.assertEqual(stable.stat().st_ino, original_inode)
|
|
|
|
def test_init_cannot_steal_live_legacy_lock(self):
|
|
legacy = self.root / ".git/agent-control/dev-integration.lock"
|
|
legacy.parent.mkdir()
|
|
with legacy.open("w") as handle:
|
|
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
inode = legacy.stat().st_ino
|
|
result = self.cli("init", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(legacy.stat().st_ino, inode)
|
|
self.cli("init")
|
|
|
|
def test_linked_worktree_uses_only_main_coordination_root(self):
|
|
task = self.initialized_task()
|
|
json.loads(self.cli("status", cwd=task).stdout)
|
|
self.assertTrue((self.root / ".locks").is_dir())
|
|
self.assertFalse((task / ".locks").exists())
|
|
self.cli("init", cwd=task, check=False)
|
|
self.assertFalse((task / ".locks").exists())
|
|
|
|
def test_task_registration_rejects_protected_branch_and_invalid_scopes(self):
|
|
self.cli("init")
|
|
result = self.cli("task", "register", "--id", "main-task", "--owner", "Test",
|
|
"--scope", "src", check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
path = self.root / ".worktrees/invalid"
|
|
self.git("worktree", "add", "-b", "feature/invalid", str(path), "dev")
|
|
for index, scope in enumerate(("../outside", "/tmp/absolute", ".git", ".locks", "src/../../outside")):
|
|
with self.subTest(scope=scope):
|
|
result = self.cli("task", "register", "--id", "invalid-" + str(index),
|
|
"--owner", "Test", "--scope", scope, cwd=path, check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_path_scope_and_capability_are_enforced(self):
|
|
task = self.initialized_task()
|
|
self.assertEqual(self.run_lease("alpha", ["path:src/a/item"], cwd=task).returncode, 0)
|
|
for resource in ("path:src/b", "path:src/ab", "path:../outside", "env:dev", "git:dev"):
|
|
with self.subTest(resource=resource):
|
|
result = self.run_lease("alpha", [resource], cwd=task)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_parent_child_path_conflict_and_disjoint_coexistence(self):
|
|
task = self.initialized_task(scopes=("src",))
|
|
process, connection, _ = self.live_lease("alpha", ["path:src/a"], task)
|
|
for resource in ("path:src", "path:src/a/file", "path:src/a"):
|
|
with self.subTest(resource=resource):
|
|
result = self.run_lease("alpha", [resource], cwd=task)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.run_lease("alpha", ["path:src/b"], cwd=task).returncode, 0)
|
|
self.finish_lease(process, connection)
|
|
self.assertEqual(self.run_lease("alpha", ["path:src/a"], cwd=task).returncode, 0)
|
|
|
|
def test_multi_resource_acquisition_is_atomic(self):
|
|
task = self.initialized_task(scopes=("src",))
|
|
process, connection, _ = self.live_lease("alpha", ["path:src/a"], task)
|
|
result = self.run_lease("alpha", ["path:src/b", "path:src/a"], cwd=task)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
result = self.run_lease("alpha", ["path:src/b"], cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.finish_lease(process, connection)
|
|
|
|
def test_task_index_lease_serializes_index_but_not_disjoint_work(self):
|
|
task = self.initialized_task(scopes=("src",))
|
|
other_task = self.initialized_task(task="beta", scopes=("src/b",))
|
|
process, connection, _ = self.live_lease("alpha", ["git:index", "path:src/a"], task)
|
|
result = self.run_lease("alpha", ["git:index", "path:src/b"], cwd=task)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
result = self.run_lease("alpha", ["path:src/b"], cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
result = self.run_lease("beta", ["git:index", "path:src/b"], cwd=other_task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.finish_lease(process, connection)
|
|
result = self.run_lease("alpha", ["git:index", "path:src/a"], cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_failed_child_releases_lease_and_preserves_failure(self):
|
|
task = self.initialized_task()
|
|
result = self.run_lease("alpha", ["path:src/a"], "raise SystemExit(17)", cwd=task)
|
|
self.assertEqual(result.returncode, 17, result.stdout + result.stderr)
|
|
self.assertEqual(self.run_lease("alpha", ["path:src/a"], cwd=task).returncode, 0)
|
|
|
|
def test_live_and_stale_credentials_cannot_be_spoofed(self):
|
|
task = self.initialized_task()
|
|
process, connection, payload = self.live_lease("alpha", ["path:src/a"], task)
|
|
self.assertTrue(payload["run"])
|
|
environment = dict(os.environ, **payload["credentials"])
|
|
result = self.cli("guard", "resource", "path:src/a", cwd=task, env=environment, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.finish_lease(process, connection)
|
|
result = self.cli("guard", "resource", "path:src/a", cwd=task, env=environment, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
|
|
def test_live_run_cannot_be_recovered_or_task_closed(self):
|
|
task = self.initialized_task()
|
|
process, connection, payload = self.live_lease("alpha", ["path:src/a"], task)
|
|
result = self.cli("recover", "--run", payload["run"], "--reason", "test", check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
result = self.cli("task", "close", "--id", "alpha", "--status", "done", cwd=task, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.finish_lease(process, connection)
|
|
|
|
def test_recovery_refuses_dead_supervisor_with_live_child(self):
|
|
if not hasattr(os, "pidfd_open"):
|
|
self.skipTest("pidfd_open is Linux-only")
|
|
task = self.initialized_task()
|
|
process, connection, payload = self.live_lease("alpha", ["path:src/a"], task)
|
|
child_fd = os.pidfd_open(payload["pid"])
|
|
self.addCleanup(os.close, child_fd)
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
self.assertFalse(select.select([child_fd], [], [], 0)[0], "fixture child must still be alive")
|
|
result = self.cli("recover", "--run", payload["run"], "--reason", "supervisor died", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertNotEqual(self.run_lease("alpha", ["path:src/a"], cwd=task).returncode, 0)
|
|
connection.sendall(b"x")
|
|
self.assertTrue(select.select([child_fd], [], [], 10)[0], "fixture child must finish")
|
|
process.communicate(timeout=5)
|
|
self.cli("recover", "--run", payload["run"], "--reason", "all processes stopped")
|
|
self.assertEqual(self.run_lease("alpha", ["path:src/a"], cwd=task).returncode, 0)
|
|
|
|
def test_descendant_resource_guard_accepts_only_held_resource(self):
|
|
task = self.initialized_task()
|
|
code = (
|
|
"import subprocess,sys; "
|
|
f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r},"
|
|
"'guard','resource','path:src/a/item']);sys.exit(p.returncode)"
|
|
)
|
|
result = self.run_lease("alpha", ["path:src/a"], code, cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
result = self.run_lease("alpha", ["path:src/a"], code.replace("path:src/a/item", "path:src/b"), cwd=task)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
|
|
def test_shared_lease_does_not_satisfy_write_guard(self):
|
|
task = self.initialized_task()
|
|
result = self.cli("with", "--task", "alpha", "--resource", "path:src/a", "--shared",
|
|
"--", sys.executable, str(self.root / "scripts/agentctl.py"),
|
|
"guard", "resource", "path:src/a", cwd=task, check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.run_lease("alpha", ["path:src/a"], cwd=task).returncode, 0)
|
|
|
|
def test_env_guard_refuses_linked_worktree_even_with_capability(self):
|
|
task = self.initialized_task(capabilities=("env:dev",))
|
|
code = (
|
|
"import subprocess,sys; "
|
|
f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r},"
|
|
"'guard','env','--target','dev']);sys.exit(p.returncode)"
|
|
)
|
|
result = self.run_lease("alpha", ["env:dev"], code, cwd=task)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_env_guard_accepts_clean_main_and_requires_all_target_leases(self):
|
|
self.initialized_task(capabilities=("env:dev", "env:test", "env:main", "env:shared"))
|
|
code = (
|
|
"import subprocess,sys; "
|
|
f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r},"
|
|
"'guard','env','--target','dev']);sys.exit(p.returncode)"
|
|
)
|
|
result = self.run_lease("alpha", ["env:dev"], code, cwd=self.root)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
all_code = code.replace("'dev'", "'all'")
|
|
result = self.run_lease("alpha", ["env:dev", "env:test", "env:main", "env:shared"], all_code, cwd=self.root)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
result = self.run_lease("alpha", ["env:dev"], all_code, cwd=self.root)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
|
|
def test_docs_row_cas_preserves_unrelated_rows_and_is_idempotent(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
filename = ".doc/ops/CURRENT.md"
|
|
snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "alpha", cwd=task).stdout)
|
|
self.assertEqual(snapshot["sha256"], hashlib.sha256(self.current_rows["alpha"].encode()).hexdigest())
|
|
document = task / filename
|
|
other = self.current_rows["beta"].replace("active", "reviewing")
|
|
document.write_text(document.read_text().replace(self.current_rows["beta"], other))
|
|
# A separate accepted update advanced the source baseline; row alpha is unchanged.
|
|
self.git("add", filename, cwd=task)
|
|
self.git("commit", "-m", "independent beta update", cwd=task)
|
|
replacement = self.current_rows["alpha"].replace("active", "done")
|
|
request = self.docs_request([self.doc_change(filename, "alpha", self.current_rows["alpha"], replacement)])
|
|
self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task)
|
|
first = document.read_bytes()
|
|
self.assertIn(other, first.decode())
|
|
self.assertIn(replacement, first.decode())
|
|
self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task)
|
|
self.assertEqual(document.read_bytes(), first)
|
|
self.assertIn(self.current_rows["alpha"], (self.root / filename).read_text())
|
|
|
|
def test_docs_foreign_current_row_is_rejected(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
filename = ".doc/ops/CURRENT.md"
|
|
before = (task / filename).read_bytes()
|
|
request = self.docs_request([self.doc_change(filename, "beta", self.current_rows["beta"],
|
|
self.current_rows["beta"].replace("active", "done"))])
|
|
result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual((task / filename).read_bytes(), before)
|
|
|
|
def test_docs_multi_file_preflight_failure_has_no_partial_writes(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
current = ".doc/ops/CURRENT.md"
|
|
features = ".doc/status/FEATURE-MATRIX.md"
|
|
before = {name: (task / name).read_bytes() for name in (current, features)}
|
|
request = self.docs_request([
|
|
self.doc_change(current, "alpha", self.current_rows["alpha"], self.current_rows["alpha"].replace("active", "done")),
|
|
self.doc_change(features, "core", "| core | stale |", "| core | done |"),
|
|
])
|
|
result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
for filename, contents in before.items():
|
|
self.assertEqual((task / filename).read_bytes(), contents)
|
|
|
|
def test_docs_cannot_launder_unmanaged_other_task_changes(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
filename = ".doc/ops/CURRENT.md"
|
|
document = task / filename
|
|
document.write_text(document.read_text().replace(self.current_rows["beta"], self.current_rows["beta"].replace("active", "done")))
|
|
before = document.read_bytes()
|
|
request = self.docs_request([self.doc_change(filename, "alpha", self.current_rows["alpha"], self.current_rows["alpha"].replace("active", "done"))])
|
|
result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("UNMANAGED_DOCUMENT_CHANGE", result.stderr)
|
|
self.assertEqual(document.read_bytes(), before)
|
|
|
|
def test_commit_lease_must_cover_staged_files_not_only_task_scope(self):
|
|
task = self.initialized_task(scopes=("src",))
|
|
self.cli("hooks", "install")
|
|
file = task / "src/b/other.py"
|
|
file.parent.mkdir(parents=True, exist_ok=True)
|
|
file.write_text("value = 2\n")
|
|
self.git("add", "src/b/other.py", cwd=task)
|
|
code = "import subprocess,sys;sys.exit(subprocess.run(['git','commit','-m','missing file lease']).returncode)"
|
|
result = self.run_lease("alpha", ["git:index", "path:src/a"], code, cwd=task)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("LEASE_REQUIRED", result.stderr)
|
|
|
|
def test_guard_build_requires_exact_clean_commit(self):
|
|
task = self.initialized_task()
|
|
head = self.git("rev-parse", "HEAD", cwd=task)
|
|
self.cli("guard", "build", "--commit", head, cwd=task)
|
|
self.assertNotEqual(self.cli("guard", "build", "--commit", "0" * 40, cwd=task, check=False).returncode, 0)
|
|
(task / "new-source.py").write_text("print('dirty')\n")
|
|
self.assertNotEqual(self.cli("guard", "build", "--commit", head, cwd=task, check=False).returncode, 0)
|
|
|
|
def test_installed_hooks_reject_direct_dev_commit(self):
|
|
self.cli("init")
|
|
self.cli("hooks", "install")
|
|
original = self.git("rev-parse", "HEAD")
|
|
(self.root / "source.txt").write_text("change\n")
|
|
self.git("add", "source.txt")
|
|
result = subprocess.run(["git", "commit", "-m", "unauthorized direct dev commit"],
|
|
cwd=self.root, text=True, capture_output=True, timeout=20)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.git("rev-parse", "HEAD"), original)
|
|
|
|
def test_hooks_refuse_overwriting_foreign_hook(self):
|
|
self.cli("init")
|
|
hook = self.root / ".git/hooks/pre-commit"
|
|
original = "#!/bin/sh\n# existing local protection\nexit 0\n"
|
|
hook.write_text(original)
|
|
hook.chmod(0o755)
|
|
result = self.cli("hooks", "install", check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(hook.read_text(), original)
|
|
|
|
def test_hook_checks_reject_out_of_scope_and_secret_paths(self):
|
|
task = self.initialized_task()
|
|
self.cli("hooks", "install")
|
|
for name in ("outside.txt", "src/a/.env"):
|
|
with self.subTest(name=name):
|
|
target = task / name
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text("test fixture, no actual secret\n")
|
|
self.git("add", "--", name, cwd=task)
|
|
code = (
|
|
"import subprocess,sys; "
|
|
f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r},"
|
|
"'hooks','check']);sys.exit(p.returncode)"
|
|
)
|
|
result = self.run_lease("alpha", ["git:index", "path:src/a"], code, cwd=task)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.git("reset", "--", name, cwd=task)
|
|
target.unlink()
|
|
|
|
def test_hook_rejects_shared_document_without_publish_receipt(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
self.cli("hooks", "install")
|
|
current = task / ".doc/ops/CURRENT.md"
|
|
current.write_text(current.read_text().replace(self.current_rows["alpha"],
|
|
self.current_rows["alpha"].replace("active", "done")))
|
|
self.git("add", ".doc/ops/CURRENT.md", cwd=task)
|
|
code = (
|
|
"import subprocess,sys; "
|
|
f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r},"
|
|
"'hooks','check']);sys.exit(p.returncode)"
|
|
)
|
|
result = self.run_lease("alpha", ["git:index", "path:.doc", "docs:publish"], code, cwd=task)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_hooks_allow_scoped_task_commit_under_live_lease(self):
|
|
task = self.initialized_task()
|
|
self.cli("hooks", "install")
|
|
source = task / "src/a/example.py"
|
|
source.parent.mkdir(parents=True, exist_ok=True)
|
|
source.write_text("value = 1\n")
|
|
self.git("add", "src/a/example.py", cwd=task)
|
|
code = "import subprocess,sys; sys.exit(subprocess.run(['git','commit','-m','scoped task change']).returncode)"
|
|
result = self.run_lease("alpha", ["git:index", "path:src/a"], code, cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.git("status", "--porcelain", cwd=task), "")
|
|
|
|
def test_admin_full_document_requires_explicit_admin_capability(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
filename = ".doc/ops/CURRENT.md"
|
|
before = (task/filename).read_text()
|
|
request = self.docs_request([self.doc_change(filename,"@document",before,before+"\n")], "no-admin")
|
|
result = self.cli("docs","apply","--task","alpha","--request",request,cwd=task,check=False)
|
|
self.assertNotEqual(result.returncode,0)
|
|
self.assertIn("CAPABILITY_REQUIRED",result.stderr)
|
|
self.assertEqual((task/filename).read_text(),before)
|
|
|
|
def test_registered_active_task_must_close_before_cleanup(self):
|
|
task = self.initialized_task()
|
|
result = self.cli("worktree-check","--path",str(task),check=False)
|
|
self.assertNotEqual(result.returncode,0)
|
|
self.assertIn("TASK_ACTIVE",result.stderr)
|
|
self.cli("task","close","--id","alpha")
|
|
self.cli("worktree-check","--path",str(task))
|
|
|
|
def test_failed_revalidation_revokes_old_success_receipt(self):
|
|
task = self.initialized_task()
|
|
self.cli("validate","--task","alpha","--",sys.executable,"-c","pass",cwd=task)
|
|
result = self.cli("validate","--task","alpha","--",sys.executable,"-c","raise SystemExit(3)",cwd=task,check=False)
|
|
self.assertEqual(result.returncode,3)
|
|
import sqlite3
|
|
with sqlite3.connect(self.root/'.locks/state.sqlite') as db:
|
|
self.assertEqual(db.execute('SELECT count(*) FROM validations').fetchone()[0],0)
|
|
result=self.cli('validate','--task','alpha','--',sys.executable,'-c',"from pathlib import Path;Path('dirty-generated').write_text('x')",cwd=task,check=False)
|
|
self.assertNotEqual(result.returncode,0)
|
|
with sqlite3.connect(self.root/'.locks/state.sqlite') as db:
|
|
self.assertEqual(db.execute('SELECT count(*) FROM validations').fetchone()[0],0)
|
|
|
|
def test_invalid_document_request_shape_has_no_write_side_effects(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
original = (task/'.doc/ops/CURRENT.md').read_bytes()
|
|
for changes in [[None],[{'file':[]}]]:
|
|
request = self.docs_request(changes,'bad-shape-'+str(len(str(changes))))
|
|
result = self.cli('docs','apply','--task','alpha','--request',request,cwd=task,check=False)
|
|
self.assertNotEqual(result.returncode,0)
|
|
json.loads(result.stderr)
|
|
self.assertEqual((task/'.doc/ops/CURRENT.md').read_bytes(),original)
|
|
|
|
def test_admin_document_cannot_bypass_done_with_compact_cells(self):
|
|
task = self.initialized_task(capabilities=("docs:publish", "docs:admin"))
|
|
filename = ".doc/status/FEATURE-MATRIX.md"
|
|
snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "@document", cwd=task).stdout)
|
|
original = (task / filename).read_text()
|
|
for indent in range(4):
|
|
with self.subTest(indent=indent):
|
|
request = self.docs_request([{"file":filename,"record_id":"@document","expected_sha256":snapshot["sha256"],
|
|
"replacement":original.replace("| core | planned |", " "*indent+"| core |done|")}], "compact-done-"+str(indent))
|
|
result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual((task / filename).read_text(), original)
|
|
|
|
def test_validated_linear_commit_cannot_bypass_candidate_entrypoint(self):
|
|
task = self.initialized_task(scopes=("src",), capabilities=("git:dev",))
|
|
self.cli("hooks", "install")
|
|
(task / "src/a/new.py").parent.mkdir(parents=True, exist_ok=True)
|
|
(task / "src/a/new.py").write_text("ok = 1\n")
|
|
self.git("add", "src/a/new.py", cwd=task)
|
|
result = self.run_lease("alpha", ["git:index", "path:src"], "import subprocess,sys;sys.exit(subprocess.run(['git','commit','-m','linear']).returncode)", cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.cli("validate", "--task", "alpha", "--", "git", "diff", "--check", cwd=task)
|
|
original = self.git("rev-parse", "dev")
|
|
result = self.cli("with", "--task", "alpha", "--resource", "git:dev", "--cwd", str(self.root),
|
|
"--", "git", "merge", "--ff-only", "feature/alpha", check=False)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.git("rev-parse", "dev"), original)
|
|
|
|
def test_interrupted_validation_stops_child_before_releasing_scope(self):
|
|
if not hasattr(os, "pidfd_open"):
|
|
self.skipTest("pidfd_open is Linux-only")
|
|
task = self.initialized_task()
|
|
listener = socket.socket(socket.AF_UNIX)
|
|
endpoint = str(pathlib.Path(self.temp.name) / "validate.sock")
|
|
listener.bind(endpoint)
|
|
listener.listen(1)
|
|
listener.settimeout(10)
|
|
code = "import os,socket,time;s=socket.socket(socket.AF_UNIX);s.connect(%r);s.sendall(str(os.getpid()).encode());time.sleep(60)" % endpoint
|
|
process = subprocess.Popen([sys.executable,str(self.root/'scripts/agentctl.py'),'validate','--task','alpha','--',sys.executable,'-c',code],cwd=task,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
|
|
child_pid = None
|
|
try:
|
|
connection,_ = listener.accept()
|
|
with connection:
|
|
child_pid = int(connection.recv(100))
|
|
pidfd = os.pidfd_open(child_pid)
|
|
try:
|
|
process.send_signal(__import__('signal').SIGINT)
|
|
process.wait(timeout=10)
|
|
self.assertNotEqual(process.returncode, 0)
|
|
self.assertTrue(select.select([pidfd],[],[],2)[0], 'validator released with live child')
|
|
finally:
|
|
os.close(pidfd)
|
|
finally:
|
|
listener.close()
|
|
if child_pid:
|
|
try: os.killpg(child_pid,__import__('signal').SIGKILL)
|
|
except ProcessLookupError: pass
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.communicate(timeout=5)
|
|
|
|
def test_validated_merge_candidate_is_exactly_promoted_to_dev(self):
|
|
task = self.initialized_task()
|
|
self.cli("hooks", "install")
|
|
(task / "src/a/feature.py").parent.mkdir(parents=True, exist_ok=True)
|
|
(task / "src/a/feature.py").write_text("feature = True\n")
|
|
self.git("add", "src/a/feature.py", cwd=task)
|
|
change = self.run_lease("alpha", ["git:index", "path:src/a"],
|
|
"import subprocess,sys;sys.exit(subprocess.run(['git','commit','-m','feature']).returncode)", cwd=task)
|
|
self.assertEqual(change.returncode, 0, change.stderr)
|
|
baseline = self.git("rev-parse", "dev")
|
|
candidate = self.initialized_task("candidate", scopes=(".",), capabilities=("git:dev",))
|
|
merged = self.run_lease("candidate", ["git:index", "git:dev", "path:."],
|
|
"import subprocess,sys;sys.exit(subprocess.run(['git','merge','--no-ff','-m','candidate','feature/alpha']).returncode)", cwd=candidate)
|
|
self.assertEqual(merged.returncode, 0, merged.stderr)
|
|
head = self.git("rev-parse", "HEAD", cwd=candidate)
|
|
denied = self.cli("integrate", "--task", "candidate", "--expected-dev", baseline, cwd=candidate, check=False)
|
|
self.assertNotEqual(denied.returncode, 0)
|
|
self.assertIn("VALIDATION_REQUIRED", denied.stderr)
|
|
self.cli("validate", "--task", "candidate", "--", "git", "diff", "--check", baseline, cwd=candidate)
|
|
# Interruption at the Git boundary must reap its child and retain recovery.
|
|
listener = socket.socket(socket.AF_UNIX)
|
|
endpoint = str(pathlib.Path(self.temp.name)/"integrate.sock")
|
|
listener.bind(endpoint); listener.listen(1); listener.settimeout(10)
|
|
proxy_dir = pathlib.Path(self.temp.name)/"proxy-bin"; proxy_dir.mkdir()
|
|
real_git = shutil.which("git")
|
|
proxy = proxy_dir/"git"
|
|
proxy.write_text("#!/usr/bin/env python3\nimport os,sys,socket\n"
|
|
+ "if 'merge' in sys.argv and '--ff-only' in sys.argv:\n"
|
|
+ f" s=socket.socket(socket.AF_UNIX);s.connect({endpoint!r});s.sendall(os.environ['AGENTCTL_RUN_ID'].encode());s.recv(1)\n"
|
|
+ f"os.execv({real_git!r},[{real_git!r},*sys.argv[1:]])\n")
|
|
proxy.chmod(0o755)
|
|
process = subprocess.Popen([sys.executable,str(self.root/'scripts/agentctl.py'),'integrate','--task','candidate','--expected-dev',baseline],cwd=candidate,
|
|
env={**os.environ,'PATH':str(proxy_dir)+os.pathsep+os.environ['PATH']},stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
|
|
try:
|
|
connection,_=listener.accept()
|
|
with connection:
|
|
run_id=connection.recv(100).decode()
|
|
process.send_signal(__import__('signal').SIGINT)
|
|
process.communicate(timeout=10)
|
|
self.assertNotEqual(process.returncode,0)
|
|
self.assertEqual(self.git('rev-parse','dev'),baseline)
|
|
blocked=self.run_lease('candidate',['git:index'],cwd=candidate)
|
|
self.assertNotEqual(blocked.returncode,0)
|
|
self.cli('recover','--run',run_id,'--reason','test confirmed Git proxy stopped before ref update')
|
|
finally:
|
|
listener.close()
|
|
if process.poll() is None:
|
|
process.terminate();process.communicate(timeout=10)
|
|
self.cli("integrate", "--task", "candidate", "--expected-dev", baseline, cwd=candidate)
|
|
self.assertEqual(self.git("rev-parse", "dev"), head)
|
|
self.assertEqual(self.git("status", "--porcelain"), "")
|
|
# Reusing the old expected base cannot replay a second ref update.
|
|
stale = self.cli("integrate", "--task", "candidate", "--expected-dev", baseline, cwd=candidate, check=False)
|
|
self.assertNotEqual(stale.returncode, 0)
|
|
self.assertEqual(self.git("rev-parse", "dev"), head)
|
|
|
|
def test_admin_document_cas_preserves_slot_and_needs_capability(self):
|
|
task = self.initialized_task(capabilities=("docs:publish", "docs:admin"))
|
|
filename = ".doc/ops/CURRENT.md"
|
|
snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "@document", cwd=task).stdout)
|
|
original = (task / filename).read_text()
|
|
request = self.docs_request([{ "file":filename, "record_id":"@document", "expected_sha256":snapshot['sha256'],
|
|
"replacement":original.replace("# Coordination", "# Active coordination") }], "admin-layout")
|
|
self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task)
|
|
self.assertIn("# Active coordination", (task / filename).read_text())
|
|
|
|
def test_replaced_compatibility_inode_is_not_a_new_free_lock(self):
|
|
self.cli("init")
|
|
bridge = self.root / ".locks/os/dev-integration.lock"
|
|
bridge.unlink()
|
|
bridge.write_text("")
|
|
self.assertNotEqual(self.cli("status", check=False).returncode, 0)
|
|
|
|
def test_hooks_allow_published_shared_document_receipt(self):
|
|
task = self.initialized_task(capabilities=("docs:publish",))
|
|
self.cli("hooks", "install")
|
|
filename = ".doc/ops/CURRENT.md"
|
|
request = self.docs_request([self.doc_change(filename, "alpha", self.current_rows["alpha"],
|
|
self.current_rows["alpha"].replace("active", "done"))])
|
|
self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task)
|
|
self.git("add", filename, cwd=task)
|
|
code = "import subprocess,sys; sys.exit(subprocess.run(['git','commit','-m','published task state']).returncode)"
|
|
result = self.run_lease("alpha", ["git:index", "path:.doc", "docs:publish"], code, cwd=task)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.git("status", "--porcelain", cwd=task), "")
|
|
|
|
|
|
class ProcessPlatformTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("agentctl_under_test", SOURCE)
|
|
cls.mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(cls.mod)
|
|
|
|
def test_boot_and_self_proc_are_observable(self):
|
|
boot = self.mod.boot_epoch()
|
|
self.assertTrue(boot)
|
|
lookup, info = self.mod.read_proc(os.getpid())
|
|
self.assertEqual(lookup, "ok")
|
|
self.assertIsNotNone(info)
|
|
self.assertNotEqual(info.state, "Z")
|
|
self.assertTrue(info.start)
|
|
self.assertEqual(self.mod.alive_status(os.getpid(), info.start, boot), "alive")
|
|
self.assertEqual(self.mod.alive_status(os.getpid(), info.start, "other-boot"), "dead")
|
|
self.assertEqual(self.mod.descent_status(os.getpid()), "yes")
|
|
|
|
def test_gone_pid_is_dead(self):
|
|
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
|
|
pid = child.pid
|
|
lookup, info = self.mod.read_proc(pid)
|
|
self.assertEqual(lookup, "ok")
|
|
self.assertIsNotNone(info)
|
|
child.kill()
|
|
child.wait(timeout=5)
|
|
self.assertEqual(self.mod.read_proc(pid)[0], "gone")
|
|
self.assertEqual(self.mod.alive_status(pid, info.start, self.mod.boot_epoch()), "dead")
|
|
|
|
def test_zombie_is_not_alive(self):
|
|
info = self.mod.ProcInfo(pid=9, ppid=1, pgid=9, state="Z", start="token")
|
|
from unittest import mock
|
|
with mock.patch.object(self.mod, "read_proc", return_value=("ok", info)), \
|
|
mock.patch.object(self.mod, "boot_epoch", return_value="boot"):
|
|
self.assertEqual(self.mod.alive_status(9, "token", "boot"), "dead")
|
|
|
|
def test_unreadable_proc_is_unknown_not_dead(self):
|
|
from unittest import mock
|
|
with mock.patch.object(self.mod, "read_proc", return_value=("unreadable", None)), \
|
|
mock.patch.object(self.mod, "boot_epoch", return_value="boot"):
|
|
self.assertEqual(self.mod.alive_status(9, "start", "boot"), "unknown")
|
|
self.assertTrue(self.mod.alive(9, "start", "boot"))
|
|
with mock.patch.object(self.mod, "_enumerate_group", return_value=(False, "unknown")):
|
|
self.assertEqual(self.mod.group_status(42), "unknown")
|
|
self.assertTrue(self.mod.group_alive(42))
|
|
with mock.patch.object(self.mod, "_enumerate_group", return_value=(False, "ok")):
|
|
self.assertEqual(self.mod.group_status(42), "dead")
|
|
with mock.patch.object(self.mod, "_enumerate_group", return_value=(True, "unknown")):
|
|
self.assertEqual(self.mod.group_status(42), "alive")
|
|
|
|
def test_descent_unreadable_is_unknown(self):
|
|
from unittest import mock
|
|
with mock.patch.object(self.mod, "read_proc", return_value=("unreadable", None)):
|
|
self.assertEqual(self.mod.descent_status(1), "unknown")
|
|
self.assertFalse(self.mod.descendants_of(1))
|
|
|
|
def test_lease_denies_when_supervisor_proc_unreadable(self):
|
|
from unittest import mock
|
|
temp = tempfile.TemporaryDirectory(prefix="agentctl-proc-")
|
|
self.addCleanup(temp.cleanup)
|
|
root = pathlib.Path(temp.name) / "repo"
|
|
root.mkdir()
|
|
subprocess.run(["git", "init", "-b", "dev"], cwd=root, check=True, capture_output=True)
|
|
subprocess.run(["git", "config", "user.name", "t"], cwd=root, check=True, capture_output=True)
|
|
subprocess.run(["git", "config", "user.email", "t@example.invalid"], cwd=root, check=True, capture_output=True)
|
|
(root / "scripts").mkdir()
|
|
shutil.copy2(SOURCE, root / "scripts" / "agentctl.py")
|
|
(root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
|
|
(root / ".doc/ops").mkdir(parents=True)
|
|
(root / ".doc/ops/CURRENT.md").write_text("# c\n", encoding="utf-8")
|
|
(root / ".doc/status").mkdir(parents=True)
|
|
(root / ".doc/status/FEATURE-MATRIX.md").write_text("# f\n", encoding="utf-8")
|
|
(root / "src/a").mkdir(parents=True)
|
|
subprocess.run(["git", "add", "."], cwd=root, check=True, capture_output=True)
|
|
subprocess.run(["git", "commit", "-m", "fixture"], cwd=root, check=True, capture_output=True)
|
|
cli = [sys.executable, str(root / "scripts/agentctl.py")]
|
|
subprocess.run(cli + ["init"], cwd=root, check=True, capture_output=True)
|
|
task = root / ".worktrees" / "alpha"
|
|
subprocess.run(["git", "worktree", "add", "-b", "feature/alpha", str(task), "dev"],
|
|
cwd=root, check=True, capture_output=True)
|
|
subprocess.run(cli + ["task", "register", "--id", "alpha", "--owner", "t", "--scope", "src/a"],
|
|
cwd=task, check=True, capture_output=True)
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("agentctl_lease", root / "scripts/agentctl.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
repo = mod.Repo(task)
|
|
task_row = mod.task_row(repo, "alpha")
|
|
with mock.patch.object(mod, "read_proc", return_value=("unreadable", None)):
|
|
with self.assertRaises(mod.Denied) as raised:
|
|
mod.Lease(repo, task_row, ["path:src/a"], None, "fp", 0).__enter__()
|
|
self.assertEqual(raised.exception.code, "PROCESS_UNREADABLE")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|