#!/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" POLICY_SOURCE = SOURCE.parents[1] / "configs" / "agent-governance.json" def install_governance_policy(root: pathlib.Path) -> None: dest = root / "configs" dest.mkdir(parents=True, exist_ok=True) shutil.copy2(POLICY_SOURCE, dest / "agent-governance.json") 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") shutil.copy2(SOURCE.parent / "agentctl_w4.py", self.root / "scripts" / "agentctl_w4.py") shutil.copy2(SOURCE.parent / "agentctl_catalog.py", self.root / "scripts" / "agentctl_catalog.py") shutil.copy2(SOURCE.parent / "agentctl_visibility.py", self.root / "scripts" / "agentctl_visibility.py") shutil.copy2(SOURCE.parent / "agentctl_console.py", self.root / "scripts" / "agentctl_console.py") install_governance_policy(self.root) (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| record_id | 模块 | 状态 |\n| --- | --- | --- |\n| core | 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 _last_json(self, text): lines = [line for line in text.splitlines() if line.strip()] self.assertTrue(lines, text) return json.loads(lines[-1]) 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 | core | stale |", "| core | 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 | core | planned |", " "*indent+"| core | 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_matrix_done_without_evidence_refused(self): task = self.initialized_task(capabilities=("docs:publish", "docs:approve", "docs:admin")) filename = ".doc/status/FEATURE-MATRIX.md" old = "| core | core | planned |" # Row update to done without evidence list → EVIDENCE_REQUIRED request = self.docs_request([ self.doc_change(filename, "core", old, "| core | core | done |"), ], "matrix-done-no-evidence") denied = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("EVIDENCE_REQUIRED", denied.stderr) self.assertIn("| core | core | planned |", (task / filename).read_text()) # @document rewrite that bumps partial/planned→done without per-row evidence original = (task / filename).read_text() snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "@document", cwd=task).stdout) bumped = original.replace("| core | core | planned |", "| core | core | done |") request = self.docs_request([{ "file": filename, "record_id": "@document", "expected_sha256": snapshot["sha256"], "replacement": bumped, }], "matrix-doc-done-bypass") denied = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("EVIDENCE_REQUIRED", denied.stderr) self.assertEqual((task / filename).read_text(), original) def test_matrix_record_id_layout_migration_without_new_dones(self): """Additive record_id column must not trip EVIDENCE_REQUIRED when statuses unchanged.""" task = self.initialized_task(capabilities=("docs:publish", "docs:admin")) filename = ".doc/status/FEATURE-MATRIX.md" legacy = ( "# Features\n\n" "| 模块 | 状态 |\n" "| --- | --- |\n" "| core | planned |\n" "| intent | done |\n" ) (task / filename).write_text(legacy, encoding="utf-8") self.git("add", filename, cwd=task) self.git("commit", "-m", "legacy matrix", cwd=task) snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "@document", cwd=task).stdout) migrated = ( "# Features\n\n" "| record_id | 模块 | 状态 |\n" "| --- | --- | --- |\n" "| core | core | planned |\n" "| intent-l1 | intent | done |\n" ) request = self.docs_request([{ "file": filename, "record_id": "@document", "expected_sha256": snapshot["sha256"], "replacement": migrated, }], "matrix-layout-migrate") result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task) self.assertEqual(result.returncode, 0, result.stderr) text = (task / filename).read_text() self.assertIn("| record_id | 模块 | 状态 |", text) self.assertIn("| intent-l1 | intent | done |", text) self.assertIn("| core | core | planned |", text) def test_matrix_unknown_columns_and_lineage_preserved(self): task = self.initialized_task(capabilities=("docs:publish", "docs:admin")) filename = ".doc/status/FEATURE-MATRIX.md" body = ( "# Features\n\n" "| record_id | 模块 | 状态 | extra |\n" "| --- | --- | --- | --- |\n" "| core | core | planned | keep-me |\n\n" "## Wave 0 lineage\n\n" "trailing history must survive layout edits.\n" ) (task / filename).write_text(body, encoding="utf-8") self.git("add", filename, cwd=task) self.git("commit", "-m", "seed matrix lineage", cwd=task) snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "@document", cwd=task).stdout) updated = body.replace("| core | core | planned | keep-me |", "| core | core | planned | keep-me-too |") request = self.docs_request([{ "file": filename, "record_id": "@document", "expected_sha256": snapshot["sha256"], "replacement": updated, }], "matrix-preserve-wave0") result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task) self.assertEqual(result.returncode, 0, result.stderr) text = (task / filename).read_text() self.assertIn("## Wave 0 lineage", text) self.assertIn("trailing history must survive layout edits.", text) self.assertIn("keep-me-too", text) self.assertIn("| record_id | 模块 | 状态 | extra |", text) def test_task_close_does_not_auto_promote_matrix_done(self): task = self.initialized_task(capabilities=("docs:publish",)) filename = ".doc/status/FEATURE-MATRIX.md" before = (task / filename).read_text() self.assertIn("| core | core | planned |", before) closed = json.loads(self.cli("task", "close", "--id", "alpha", "--status", "done").stdout) self.assertEqual(closed["status"], "done") after = (task / filename).read_text() self.assertEqual(after, before) self.assertIn("| core | core | planned |", after) self.assertFalse((task / ".doc/remediation/gap-matrix.md").exists()) def test_matrix_duplicate_or_rename_record_id_rejected(self): task = self.initialized_task(capabilities=("docs:publish", "docs:admin")) filename = ".doc/status/FEATURE-MATRIX.md" # Duplicate record_id in document → INVALID_DOCUMENT / AMBIGUOUS dup = ( "# Features\n\n" "| record_id | 模块 | 状态 |\n" "| --- | --- | --- |\n" "| core | core | planned |\n" "| core | other | partial |\n" ) (task / filename).write_text(dup, encoding="utf-8") self.git("add", filename, cwd=task) self.git("commit", "-m", "dup matrix", cwd=task) snapshot = json.loads(self.cli("docs", "snapshot", "--file", filename, "--record", "@document", cwd=task).stdout) request = self.docs_request([{ "file": filename, "record_id": "@document", "expected_sha256": snapshot["sha256"], "replacement": dup, }], "matrix-dup-apply") denied = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(denied.returncode, 0) self.assertTrue( "INVALID_DOCUMENT" in denied.stderr or "AMBIGUOUS_RECORD" in denied.stderr, denied.stderr) # Replacement missing / renaming record_id → INVALID_REQUEST good = ( "# Features\n\n" "| record_id | 模块 | 状态 |\n" "| --- | --- | --- |\n" "| core | core | planned |\n" ) (task / filename).write_text(good, encoding="utf-8") self.git("add", filename, cwd=task) self.git("commit", "-m", "reset matrix", cwd=task) old = "| core | core | planned |" # Renaming record_id while editing under the old id request = self.docs_request([ self.doc_change(filename, "core", old, "| core-renamed | core | planned |"), ], "matrix-rename-id") denied = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("INVALID_REQUEST", denied.stderr) request = self.docs_request([ self.doc_change(filename, "core", old, "| enigmamodule | planned |"), ], "matrix-missing-id") denied = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("INVALID_REQUEST", denied.stderr) 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','verified: 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), "") def test_queue_subparser_registered_once(self): """Duplicate queue registration must not shadow handler-compatible flags.""" import importlib.util spec = importlib.util.spec_from_file_location("agentctl_queue_parser", SOURCE) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) parser = mod.parser() enqueue = parser.parse_args( ["queue", "enqueue", "--task", "alpha", "--expected-dev", "dev", "--head", "deadbeef"]) self.assertEqual(enqueue.head, "deadbeef") self.assertFalse(hasattr(enqueue, "head_sha")) cancel = parser.parse_args( ["queue", "cancel", "--task", "alpha", "--id", "item-1", "--reason", "test"]) self.assertEqual(cancel.task, "alpha") schedule = parser.parse_args(["queue", "schedule", "--drain"]) self.assertTrue(schedule.drain) # A second registration that only exposes --head-sha must not win. with self.assertRaises(SystemExit): parser.parse_args( ["queue", "enqueue", "--task", "alpha", "--expected-dev", "dev", "--head-sha", "deadbeef"]) def test_queue_enqueue_list_cancel_roundtrip(self): task = self.initialized_task(scopes=(".",), capabilities=("git:dev",)) baseline = self.git("rev-parse", "dev") (task / "src/a/feature.py").parent.mkdir(parents=True, exist_ok=True) (task / "src/a/feature.py").write_text("feature = True\n", encoding="utf-8") self.git("add", "src/a/feature.py", cwd=task) change = self.run_lease( "alpha", ["git:index", "path:."], "import subprocess,sys;sys.exit(subprocess.run(['git','commit','-m','feature']).returncode)", cwd=task) self.assertEqual(change.returncode, 0, change.stderr) 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) self.cli("validate", "--task", "candidate", "--", "git", "diff", "--check", baseline, cwd=candidate) enqueued = json.loads(self.cli( "queue", "enqueue", "--task", "candidate", "--expected-dev", baseline, "--head", head, "--request-id", "q-alpha-1", cwd=candidate).stdout) self.assertEqual(enqueued["status"], "enqueued") self.assertEqual(enqueued["item"]["state"], "pending") listed = json.loads(self.cli("queue", "list").stdout) self.assertEqual(len(listed["items"]), 1) self.assertEqual(listed["items"][0]["id"], "q-alpha-1") foreign = self.cli( "queue", "cancel", "--task", "alpha", "--id", "q-alpha-1", "--reason", "wrong owner", cwd=task, check=False) self.assertNotEqual(foreign.returncode, 0) self.assertIn("PATH_OUT_OF_SCOPE", foreign.stderr) cancelled = json.loads(self.cli( "queue", "cancel", "--task", "candidate", "--id", "q-alpha-1", "--reason", "abort", cwd=candidate).stdout) self.assertEqual(cancelled["status"], "cancelled") self.assertEqual(cancelled["item"]["state"], "cancelled") def test_queue_schedule_promotes_validated_candidate(self): task = self.initialized_task(scopes=(".",), capabilities=("git:dev",)) (task / "src/a/feature.py").parent.mkdir(parents=True, exist_ok=True) (task / "src/a/feature.py").write_text("feature = True\n", encoding="utf-8") self.git("add", "src/a/feature.py", cwd=task) change = self.run_lease( "alpha", ["git:index", "path:."], "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) self.cli("validate", "--task", "candidate", "--", "git", "diff", "--check", baseline, cwd=candidate) self.cli("queue", "enqueue", "--task", "candidate", "--expected-dev", baseline, "--head", head, "--request-id", "q-sched-1", cwd=candidate) denied = self.cli("queue", "schedule", cwd=candidate, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("MAIN_WORKTREE_REQUIRED", denied.stderr) scheduled_raw = self.cli("queue", "schedule", cwd=self.root) scheduled = json.loads([line for line in scheduled_raw.stdout.splitlines() if line.startswith("{")][-1]) self.assertEqual(scheduled["status"], "ok") self.assertEqual(scheduled["processed"], 1) self.assertEqual(scheduled["results"][0]["status"], "completed") self.assertEqual(self.git("rev-parse", "dev"), head) listed = json.loads(self.cli("queue", "list", "--all").stdout) item = next(i for i in listed["items"] if i["id"] == "q-sched-1") self.assertEqual(item["state"], "completed") def test_policy_missing_rejects_write_but_status_stays_readonly(self): self.cli("init") policy = self.root / "configs" / "agent-governance.json" policy.unlink() self.assertFalse(policy.exists()) denied = self.cli("hooks", "install", check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("POLICY_MISSING", denied.stderr) self.assertFalse(policy.exists()) status = json.loads(self.cli("status").stdout) self.assertEqual(status["status"], "ok") self.assertEqual(status["policy"]["code"], "POLICY_MISSING") self.assertFalse(policy.exists()) doctor = json.loads(self.cli("doctor").stdout) self.assertEqual(doctor["status"], "ok") self.assertEqual(doctor["policy"]["code"], "POLICY_MISSING") self.assertFalse(policy.exists()) def test_policy_invalid_json_rejects_write(self): policy = self.root / "configs" / "agent-governance.json" policy.write_text("{not-json", encoding="utf-8") denied = self.cli("init", check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("POLICY_INVALID", denied.stderr) def test_policy_protocol_too_low_rejects_write(self): policy = self.root / "configs" / "agent-governance.json" data = json.loads(policy.read_text(encoding="utf-8")) data["protocol_version"] = 0 policy.write_text(json.dumps(data), encoding="utf-8") denied = self.cli("init", check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("POLICY_MISMATCH", denied.stderr) def _register_claim_task(self, name, scopes=("src",)): path = self.root / ".worktrees" / name if not path.exists(): self.git("worktree", "add", "-b", "feature/" + name, str(path), "dev") args = ["task", "register", "--id", name, "--owner", "ClaimAgent"] for scope in scopes: args.extend(("--scope", scope)) self.cli(*args, cwd=path) return path def test_claim_concurrent_twenty_exactly_one_held(self): import concurrent.futures self.cli("init") (self.root / "src").mkdir(exist_ok=True) tasks = [] for idx in range(20): name = f"c{idx:02d}" self._register_claim_task(name, scopes=("src",)) tasks.append(name) def attempt(name): result = subprocess.run( [sys.executable, str(self.root / "scripts/agentctl.py"), "claim", "acquire", "--task", name, "--resource", "claim:src/shared", "--request-id", f"req-{name}"], cwd=self.root / ".worktrees" / name, text=True, capture_output=True, timeout=30) return name, result.returncode, result.stdout, result.stderr with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool: outcomes = list(pool.map(attempt, tasks)) held = [item for item in outcomes if item[1] == 0] busy = [item for item in outcomes if item[1] != 0] self.assertEqual(len(held), 1, outcomes) self.assertEqual(len(busy), 19) for _, code, _out, err in busy: self.assertEqual(code, 75) payload = json.loads(err) self.assertEqual(payload["code"], "LOCK_BUSY") self.assertIn("holder", payload) self.assertIn("ticket", payload) self.assertIn("retry_after", payload) listed = json.loads(self.cli("claim", "list").stdout) held_rows = [row for row in listed["claims"] if row["state"] == "held"] self.assertEqual(len(held_rows), 1) self.assertEqual(held_rows[0]["resource"], "claim:src/shared") def test_claim_unrelated_resources_parallel_ok(self): alpha = self.initialized_task("alpha", scopes=("src/a", "src/b")) beta = self.initialized_task("beta", scopes=("src/a", "src/b")) a = json.loads(self.cli( "claim", "acquire", "--task", "alpha", "--resource", "claim:src/a", "--request-id", "par-a", cwd=alpha).stdout) b = json.loads(self.cli( "claim", "acquire", "--task", "beta", "--resource", "claim:src/b", "--request-id", "par-b", cwd=beta).stdout) self.assertEqual(a["status"], "held") self.assertEqual(b["status"], "held") def test_claim_parent_child_overlap_and_sibling_ok(self): alpha = self.initialized_task("alpha", scopes=("src", "src2")) beta = self.initialized_task("beta", scopes=("src", "src2")) self.cli("claim", "acquire", "--task", "alpha", "--resource", "claim:src", "--request-id", "parent-1", cwd=alpha) child = self.cli( "claim", "acquire", "--task", "beta", "--resource", "claim:src/a.py", "--request-id", "child-1", cwd=beta, check=False) self.assertEqual(child.returncode, 75) self.assertIn("LOCK_BUSY", child.stderr) sibling = json.loads(self.cli( "claim", "acquire", "--task", "beta", "--resource", "claim:src2", "--request-id", "sib-1", cwd=beta).stdout) self.assertEqual(sibling["status"], "held") def test_claim_frozen_rejects_even_holder(self): alpha = self.initialized_task("alpha", scopes=("src",)) self.cli("claim", "acquire", "--task", "alpha", "--resource", "claim:src/locked", "--request-id", "fz-1", cwd=alpha) self.cli("freeze", "--task", "alpha", "--resource", "claim:src/locked", "--reason", "maintenance", cwd=alpha) denied = self.cli( "claim", "acquire", "--task", "alpha", "--resource", "claim:src/locked", "--request-id", "fz-2", cwd=alpha, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("FROZEN", denied.stderr) def test_claim_request_id_idempotent_and_reuse_guard(self): alpha = self.initialized_task("alpha", scopes=("src",)) first = json.loads(self.cli( "claim", "acquire", "--task", "alpha", "--resource", "claim:src/idem", "--request-id", "same-req", cwd=alpha).stdout) self.assertEqual(first["status"], "held") replay = json.loads(self.cli( "claim", "acquire", "--task", "alpha", "--resource", "claim:src/idem", "--request-id", "same-req", cwd=alpha).stdout) self.assertTrue(replay.get("replayed")) reused = self.cli( "claim", "acquire", "--task", "alpha", "--resource", "claim:src/other", "--request-id", "same-req", cwd=alpha, check=False) self.assertNotEqual(reused.returncode, 0) self.assertIn("REQUEST_ID_REUSED", reused.stderr) def test_claim_blocks_task_close_until_released(self): alpha = self.initialized_task("alpha", scopes=("src",)) self.cli("claim", "acquire", "--task", "alpha", "--resource", "claim:src/close", "--request-id", "close-1", cwd=alpha) denied = self.cli("task", "close", "--id", "alpha", check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("LOCK_BUSY", denied.stderr) self.cli("claim", "release", "--task", "alpha", "--resource", "claim:src/close", cwd=alpha) closed = json.loads(self.cli("task", "close", "--id", "alpha").stdout) self.assertEqual(closed["status"], "done") def _seed_run(self, run_id, task, resources, state, *, heartbeat_age=0, suspect_age=None, pid=None, start="start-token", boot=None, pgid=None, fence=0): import sqlite3 import time as time_mod now = time_mod.time() boot = boot or "boot-token" pid = 1 if pid is None else pid conn = sqlite3.connect(self.root / ".locks" / "state.sqlite") try: cols = {row[1] for row in conn.execute("PRAGMA table_info(runs)")} for name, decl in ( ("fence", "INTEGER NOT NULL DEFAULT 0"), ("suspect_since", "REAL"), ("quarantined_at", "REAL"), ("quarantine_reason", "TEXT"), ("verified_by", "TEXT"), ): if name not in cols: conn.execute(f"ALTER TABLE runs ADD COLUMN {name} {decl}") suspect_since = None if suspect_age is None else now - suspect_age conn.execute( """INSERT INTO runs(id,task,fingerprint,resources,mode,state,pid,start,boot, created,heartbeat,kind,pgid,fence,suspect_since) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (run_id, task, "fp", json.dumps(resources), "X", state, pid, start, boot, now, now - heartbeat_age, "command", pgid or pid, fence, suspect_since)) conn.commit() finally: conn.close() def test_quarantine_sweep_keeps_live_child_and_release_refuses(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") # Force heartbeat stale so sweep considers demotion; live child still blocks release. import sqlite3 import time as time_mod conn = sqlite3.connect(self.root / ".locks" / "state.sqlite") conn.execute("UPDATE runs SET heartbeat=? WHERE id=?", (time_mod.time() - 10_000, payload["run"])) conn.commit() conn.close() swept = json.loads(self.cli("quarantine", "sweep").stdout) states = {item["to"] for item in swept.get("demoted", []) if item["run_id"] == payload["run"]} self.assertTrue(not states or states <= {"suspect"}, swept) release = self.cli( "quarantine", "release", "--run", payload["run"], "--reason", "verified: operator checked", "--verified-by", "tester", check=False) self.assertNotEqual(release.returncode, 0, release.stdout + release.stderr) self.assertIn("LOCK_BUSY", release.stderr) connection.sendall(b"x") self.assertTrue(select.select([child_fd], [], [], 10)[0], "fixture child must finish") process.communicate(timeout=5) def test_quarantine_unreadable_proc_stops_at_suspect(self): from unittest import mock import importlib.util self.initialized_task() self._seed_run("run-unknown", "alpha", ["path:src/a"], "running", heartbeat_age=10_000) self._seed_run("run-suspect", "alpha", ["path:src/a"], "suspect", heartbeat_age=10_000, suspect_age=10_000) spec = importlib.util.spec_from_file_location("agentctl_q", self.root / "scripts/agentctl.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) repo = mod.Repo(self.root) mod.require_policy(repo) with mock.patch.object(mod, "alive_status", return_value="unknown"), \ mock.patch.object(mod, "group_status", return_value="unknown"): mod.quarantine_sweep(repo) import sqlite3 conn = sqlite3.connect(self.root / ".locks" / "state.sqlite") rows = {r[0]: r[1] for r in conn.execute( "SELECT id,state FROM runs WHERE id IN ('run-unknown','run-suspect')")} conn.close() self.assertEqual(rows["run-unknown"], "suspect") self.assertEqual(rows["run-suspect"], "suspect") def test_quarantine_ttl_demotes_never_auto_completes(self): from unittest import mock import importlib.util self.initialized_task() self._seed_run("hb-stale", "alpha", ["path:src/a"], "running", heartbeat_age=10_000) self._seed_run("sus-stale", "alpha", ["path:src/b"], "suspect", heartbeat_age=10_000, suspect_age=10_000) spec = importlib.util.spec_from_file_location("agentctl_ttl", self.root / "scripts/agentctl.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) repo = mod.Repo(self.root) mod.require_policy(repo) with mock.patch.object(mod, "alive_status", return_value="dead"), \ mock.patch.object(mod, "group_status", return_value="dead"): mod.quarantine_sweep(repo) import sqlite3 conn = sqlite3.connect(self.root / ".locks" / "state.sqlite") rows = {r[0]: r[1] for r in conn.execute( "SELECT id,state FROM runs WHERE id IN ('hb-stale','sus-stale')")} conn.close() self.assertEqual(rows["hb-stale"], "suspect") self.assertEqual(rows["sus-stale"], "quarantine") listed = json.loads(self.cli("quarantine", "list").stdout) ids = {r["id"] for r in listed["runs"]} self.assertIn("hb-stale", ids) self.assertIn("sus-stale", ids) self.assertNotIn("completed", rows.values()) self.assertNotIn("recovered", rows.values()) def test_stale_fence_env_is_stale_generation(self): task = self.initialized_task() code = ( "import os,subprocess,sys;" "os.environ['AGENTCTL_FENCE']='999';" f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r}," "'guard','resource','path:src/a']);sys.exit(p.returncode)" ) result = self.run_lease("alpha", ["path:src/a"], code, cwd=task) self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) self.assertIn("STALE_GENERATION", result.stderr) def test_unset_fence_env_matches_default_zero(self): """Self-bootstrap: hooks re-exec updated agentctl without AGENTCTL_FENCE set.""" task = self.initialized_task() code = ( "import os,subprocess,sys;" "os.environ.pop('AGENTCTL_FENCE', None);" f"p=subprocess.run([sys.executable,{str(self.root / 'scripts/agentctl.py')!r}," "'guard','resource','path:src/a']);sys.exit(p.returncode)" ) result = self.run_lease("alpha", ["path:src/a"], code, cwd=task) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) def test_recover_git_dev_requires_verification_marker(self): task = self.initialized_task(capabilities=("git:dev",)) self._seed_run("git-dev-run", "alpha", ["git:dev"], "recovery_required", heartbeat_age=10_000, pid=2**30 - 3, boot="other-boot") denied = self.cli("recover", "--run", "git-dev-run", "--reason", "looks dead", check=False) self.assertNotEqual(denied.returncode, 0, denied.stdout + denied.stderr) self.assertIn("EVIDENCE_REQUIRED", denied.stderr) import sqlite3 conn = sqlite3.connect(self.root / ".locks" / "state.sqlite") state = conn.execute("SELECT state FROM runs WHERE id='git-dev-run'").fetchone()[0] conn.close() self.assertEqual(state, "recovery_required") def test_daemon_single_instance_lock_busy_with_holder_pid(self): self.cli("init") import concurrent.futures cmd = [sys.executable, str(self.root / "scripts/agentctl.py"), "daemon", "run", "--interval", "0", "--max-cycles", "3"] def attempt(): return subprocess.run(cmd, cwd=self.root, text=True, capture_output=True, timeout=20) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: first = pool.submit(attempt) second = pool.submit(attempt) results = [first.result(), second.result()] codes = sorted(r.returncode for r in results) self.assertEqual(codes, [0, 75], [(r.returncode, r.stdout, r.stderr) for r in results]) busy = next(r for r in results if r.returncode == 75) payload = json.loads(busy.stderr) self.assertEqual(payload["code"], "LOCK_BUSY") self.assertIn("holder_pid", payload) self.assertIsInstance(payload["holder_pid"], int) def test_daemon_stop_leaves_pending_queue_untouched(self): task = self.initialized_task(scopes=(".",), capabilities=("git:dev",)) tip_before = self.git("rev-parse", "dev") # Seed a pending item that is NOT validated — daemon must not promote it. import sqlite3 import time as time_mod import uuid item_id = str(uuid.uuid4()) now = time_mod.time() with sqlite3.connect(self.root / ".locks" / "state.sqlite") as db: db.execute( "INSERT INTO queue_items(id,task,head_sha,expected_dev,priority,state,created,updated)" " VALUES(?,?,?,?,?,?,?,?)", (item_id, "alpha", tip_before, tip_before, 0, "pending", now, now)) db.commit() self.cli("daemon", "run", "--interval", "0", "--max-cycles", "2", cwd=self.root) self.assertEqual(self.git("rev-parse", "dev"), tip_before) with sqlite3.connect(self.root / ".locks" / "state.sqlite") as db: state = db.execute("SELECT state FROM queue_items WHERE id=?", (item_id,)).fetchone()[0] # Failed admit is ok; must not become completed/merged into tip. self.assertNotEqual(state, "completed") self.assertEqual(self.git("rev-parse", "dev"), tip_before) def test_daemon_does_not_grant_register_or_docs_apply(self): self.cli("init") self.cli("daemon", "run", "--interval", "0", "--max-cycles", "1", cwd=self.root) denied = self.cli("task", "register", "--id", "rogue", "--owner", "x", "--scope", "src", cwd=self.root, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("WORKTREE_REQUIRED", denied.stderr) status = json.loads(self.cli("daemon", "status", cwd=self.root).stdout) self.assertEqual(status["status"], "ok") self.assertIn("manager_lock", status) self.assertIn("queue", status) def _write_current_with_sections(self, task_path, *, claim_rows=None, slot_body=None): """CURRENT fixture with human sections + integration slot for protection tests.""" rows = claim_rows or [ "| owner | task | branch | worktree | mutex | status |", "| --- | --- | --- | --- | --- | --- |", "| Test Agent | alpha | feature/alpha | .worktrees/alpha | - | planned |", "| Test Agent | beta | feature/beta | .worktrees/beta | - | planned |", ] slot = slot_body or ( "| holder | phase | since | scope | queue |\n" "|--------|-------|-------|-------|-------|\n" "| - | idle | 2026-09-23 | test | - |\n" ) body = ( "# Coordination\n\n" "## 焦点\n\nfocus line\n\n" "## 并行认领\n\n" + "\n".join(rows) + "\n\n" "## 清理接管与遗留现场\n\n" "| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n" "| --- | --- | --- | --- |\n" "| none | none | - | - |\n\n" "## 集成槽位\n\n" + slot + "\n" "## 受控约束\n\n- constraint\n\n" "## 活跃入口\n\n- entry\n" ) path = task_path / ".doc/ops/CURRENT.md" path.write_text(body, encoding="utf-8") return path def _project_apply_with_retry(self, task_id, cwd, *extra, attempts=40): import time last = None for _ in range(attempts): last = self.cli("docs", "project", "--task", task_id, "--apply", *extra, cwd=cwd, check=False) if last.returncode == 0: return last err = last.stderr + last.stdout if "LOCK_BUSY" in err: time.sleep(0.05) continue return last return last def test_docs_project_parallel_apply_keeps_both_rows(self): import concurrent.futures alpha = self.initialized_task(capabilities=("docs:publish",)) beta = self.initialized_task(task="beta", scopes=("src/b", ".doc"), capabilities=("docs:publish",)) self._write_current_with_sections(alpha) self.git("add", ".doc/ops/CURRENT.md", cwd=alpha) self.git("commit", "-m", "sectioned current", cwd=alpha) # Mirror committed baseline into beta worktree. self.git("checkout", "feature/alpha", "--", ".doc/ops/CURRENT.md", cwd=beta) self.git("add", ".doc/ops/CURRENT.md", cwd=beta) self.git("commit", "-m", "sync current", cwd=beta) def run(task_id, cwd, status): return self._project_apply_with_retry( task_id, cwd, "--mutex", task_id, "--status", status) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: fut_a = pool.submit(run, "alpha", alpha, "in_progress") fut_b = pool.submit(run, "beta", beta, "blocked") results = [fut_a.result(), fut_b.result()] for result in results: self.assertEqual(result.returncode, 0, result.stdout + result.stderr) payload = self._last_json(result.stdout) self.assertEqual(payload["status"], "projected_and_applied") text_a = (alpha / ".doc/ops/CURRENT.md").read_text(encoding="utf-8") text_b = (beta / ".doc/ops/CURRENT.md").read_text(encoding="utf-8") self.assertRegex(text_a, r"\| Test Agent \| alpha \|.*\| in_progress \|") self.assertRegex(text_a, r"\| Test Agent \| beta \|") self.assertRegex(text_b, r"\| Test Agent \| beta \|.*\| blocked \|") self.assertRegex(text_b, r"\| Test Agent \| alpha \|") def test_docs_project_same_row_race_base_changed_no_partial(self): import concurrent.futures task = self.initialized_task(capabilities=("docs:publish",)) self._write_current_with_sections(task) self.git("add", ".doc/ops/CURRENT.md", cwd=task) self.git("commit", "-m", "sectioned current", cwd=task) snap = json.loads(self.cli( "docs", "snapshot", "--file", ".doc/ops/CURRENT.md", "--record", "alpha", cwd=task).stdout) row = snap["row"] self.assertIsNotNone(row) req_a = self.docs_request( [self.doc_change( ".doc/ops/CURRENT.md", "alpha", row, "| Test Agent | alpha | feature/alpha | .worktrees/alpha | - | in_progress |")], "same-row-a") req_b = self.docs_request( [self.doc_change( ".doc/ops/CURRENT.md", "alpha", row, "| Test Agent | alpha | feature/alpha | .worktrees/alpha | - | done |")], "same-row-b") baseline = (task / ".doc/ops/CURRENT.md").read_bytes() with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: r1 = pool.submit(lambda: self.cli( "docs", "apply", "--task", "alpha", "--request", req_a, cwd=task, check=False)) r2 = pool.submit(lambda: self.cli( "docs", "apply", "--task", "alpha", "--request", req_b, cwd=task, check=False)) results = [r1.result(), r2.result()] codes = sorted(r.returncode for r in results) self.assertEqual(codes[0], 0, [(r.returncode, r.stderr) for r in results]) self.assertNotEqual(codes[1], 0) loser = next(r for r in results if r.returncode != 0) if "LOCK_BUSY" in loser.stderr: detail = json.loads(loser.stderr) rid = detail.get("request_id") retry_req = req_a if rid == "same-row-a" else req_b loser = self.cli("docs", "apply", "--task", "alpha", "--request", retry_req, cwd=task, check=False) self.assertNotEqual(loser.returncode, 0, loser.stderr) self.assertIn("BASE_CHANGED", loser.stderr) after = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8") self.assertEqual(after.count("| alpha |"), 1) self.assertIn("## 焦点", after) self.assertIn("focus line", after) self.assertNotEqual(after.encode(), baseline) # Exactly one of the two statuses landed. self.assertTrue(("| in_progress |" in after) ^ ("| done |" in after)) def test_admin_document_cannot_overwrite_integration_slot(self): task = self.initialized_task(capabilities=("docs:publish", "docs:admin")) path = self._write_current_with_sections(task) self.git("add", ".doc/ops/CURRENT.md", cwd=task) self.git("commit", "-m", "sectioned current", cwd=task) original = path.read_text(encoding="utf-8") snap = json.loads(self.cli( "docs", "snapshot", "--file", ".doc/ops/CURRENT.md", "--record", "@document", cwd=task).stdout) mutated = original.replace("| - | idle | 2026-09-23 | test | - |", "| alice | merging | 2026-09-23 | stolen | - |") self.assertNotEqual(mutated, original) request = self.docs_request([{ "file": ".doc/ops/CURRENT.md", "record_id": "@document", "expected_sha256": snap["sha256"], "replacement": mutated, }], "slot-steal") result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("PATH_OUT_OF_SCOPE", result.stderr) self.assertEqual(path.read_text(encoding="utf-8"), original) def test_admin_document_cannot_overwrite_human_sections(self): task = self.initialized_task(capabilities=("docs:publish", "docs:admin")) path = self._write_current_with_sections(task) self.git("add", ".doc/ops/CURRENT.md", cwd=task) self.git("commit", "-m", "sectioned current", cwd=task) original = path.read_text(encoding="utf-8") snap = json.loads(self.cli( "docs", "snapshot", "--file", ".doc/ops/CURRENT.md", "--record", "@document", cwd=task).stdout) for old, new, label in ( ("focus line", "machine rewrote focus", "focus"), ("| none | none | - | - |", "| hacked | x | y | z |", "cleanup"), ("- constraint", "- machine constraint", "constraints"), ("- entry", "- machine entry", "entries"), ): with self.subTest(section=label): mutated = original.replace(old, new) self.assertNotEqual(mutated, original) request = self.docs_request([{ "file": ".doc/ops/CURRENT.md", "record_id": "@document", "expected_sha256": snap["sha256"], "replacement": mutated, }], "human-" + label) result = self.cli("docs", "apply", "--task", "alpha", "--request", request, cwd=task, check=False) self.assertNotEqual(result.returncode, 0, result.stderr) self.assertIn("PATH_OUT_OF_SCOPE", result.stderr) self.assertEqual(path.read_text(encoding="utf-8"), original) def test_docs_project_preview_apply_receipts_and_projections(self): task = self.initialized_task(capabilities=("docs:publish",)) self._write_current_with_sections(task) self.git("add", ".doc/ops/CURRENT.md", cwd=task) self.git("commit", "-m", "sectioned current", cwd=task) preview = self._last_json(self.cli( "docs", "project", "--task", "alpha", "--mutex", "m1", "--status", "in_progress", cwd=task).stdout) self.assertEqual(preview["status"], "projected") self.assertIn("request_id", preview["request"]) self.assertTrue(preview["request"]["request_id"].startswith("alpha-project-")) replay = self._last_json(self.cli( "docs", "project", "--task", "alpha", "--mutex", "m1", "--status", "in_progress", cwd=task).stdout) self.assertEqual(replay["request"]["request_id"], preview["request"]["request_id"]) bad = self.cli("docs", "project", "--task", "alpha", "--status", "queued", cwd=task, check=False) self.assertNotEqual(bad.returncode, 0) self.assertIn("INVALID_REQUEST", bad.stderr) applied = self._last_json(self.cli( "docs", "project", "--task", "alpha", "--apply", "--mutex", "m1", "--status", "in_progress", cwd=task).stdout) self.assertEqual(applied["status"], "projected_and_applied") text = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8") self.assertRegex(text, r"\| Test Agent \| alpha \|.*\| in_progress \|") import sqlite3 with sqlite3.connect(self.root / ".locks" / "state.sqlite") as db: receipt = db.execute( "SELECT hash FROM receipts WHERE task=? AND file=?", ("alpha", ".doc/ops/CURRENT.md")).fetchone() self.assertIsNotNone(receipt) self.assertEqual(receipt[0], hashlib.sha256(text.encode()).hexdigest()) ledger = self._last_json(self.cli( "docs", "projections", "--file", ".doc/ops/CURRENT.md", cwd=task).stdout) self.assertEqual(ledger["status"], "ok") self.assertTrue(any(p["record_id"] == "alpha" and p["task"] == "alpha" for p in ledger["projections"])) 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") shutil.copy2(SOURCE.parent / "agentctl_w4.py", root / "scripts" / "agentctl_w4.py") shutil.copy2(SOURCE.parent / "agentctl_catalog.py", root / "scripts" / "agentctl_catalog.py") shutil.copy2(SOURCE.parent / "agentctl_visibility.py", root / "scripts" / "agentctl_visibility.py") shutil.copy2(SOURCE.parent / "agentctl_console.py", root / "scripts" / "agentctl_console.py") install_governance_policy(root) (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()