feat(agentctl): add macOS process platform layer

Replace Linux /proc hard dependency with Darwin ps+sysctl while keeping
zombie/group/boot three-value semantics fail-closed for unreadable PIDs.
This commit is contained in:
WeiCHEN
2026-09-22 21:44:28 +08:00
parent 364ed7f1a3
commit 47fdeb1655
6 changed files with 465 additions and 52 deletions
+52 -3
View File
@@ -121,7 +121,7 @@ def create(repo, main, branch):
print(json.dumps({"branch": branch, "worktree": str(path), "base_dev_sha": base}))
def assert_no_process_cwd(path):
def _linux_same_user_cwds():
unreadable = []
for proc in Path("/proc").iterdir():
if not proc.name.isdigit():
@@ -130,8 +130,7 @@ def assert_no_process_cwd(path):
if proc.stat().st_uid != os.getuid():
continue
cwd = Path(os.readlink(proc / "cwd"))
if cwd == path or path in cwd.parents:
raise Refused(f"process {proc.name} still uses this worktree as cwd")
yield proc.name, cwd
except (FileNotFoundError, ProcessLookupError):
continue
except PermissionError:
@@ -150,6 +149,56 @@ def assert_no_process_cwd(path):
raise Refused(f"cannot verify cwd of same-user process {','.join(unreadable)}; cleanup_pending")
def _darwin_same_user_cwds():
try:
result = subprocess.run(
["lsof", "-a", "-d", "cwd", "-u", str(os.getuid()), "-F", "pn"],
capture_output=True, text=True)
except OSError as exc:
raise Refused(f"cannot enumerate process cwd via lsof ({exc}); cleanup_pending") from exc
if result.returncode not in (0, 1):
# lsof exits 1 when the selection is empty; other codes are opaque failure.
raise Refused(f"cannot enumerate process cwd via lsof (exit {result.returncode}); cleanup_pending")
if result.stderr.strip() and result.returncode != 0 and not result.stdout.strip():
raise Refused("cannot verify process cwd via lsof; cleanup_pending")
pid, saw_cwd = None, False
for line in result.stdout.splitlines():
if not line:
continue
code, value = line[0], line[1:]
if code == "p":
if pid is not None and not saw_cwd:
raise Refused(f"cannot verify cwd of same-user process {pid}; cleanup_pending")
pid, saw_cwd = value, False
elif code == "n" and pid is not None:
yield pid, Path(value)
saw_cwd = True
# macOS lsof may still emit f/type lines; ignore them.
if pid is not None and not saw_cwd:
raise Refused(f"cannot verify cwd of same-user process {pid}; cleanup_pending")
def iter_same_user_cwds():
"""Yield (pid_str, cwd Path) for same-uid processes. Fail closed on gaps."""
if sys.platform.startswith("linux"):
yield from _linux_same_user_cwds()
elif sys.platform == "darwin":
yield from _darwin_same_user_cwds()
else:
raise Refused(f"process cwd checks unsupported on {sys.platform}; cleanup_pending")
def assert_no_process_cwd(path):
path = path.resolve()
for pid, cwd in iter_same_user_cwds():
try:
resolved = cwd.resolve()
except OSError as exc:
raise Refused(f"cannot verify cwd of same-user process {pid}; cleanup_pending") from exc
if resolved == path or path in resolved.parents:
raise Refused(f"process {pid} still uses this worktree as cwd")
@contextmanager
def removal_lock(main, path):
# Every agentctl run/registration holds the shared side of this same lock.
+218 -37
View File
@@ -2,10 +2,12 @@
"""Local, cooperative agent coordination. OS isolation is intentionally out of scope."""
import argparse
import contextlib
import errno
import fcntl
import hashlib
import json
import os
from dataclasses import dataclass
from pathlib import Path
import re
import signal
@@ -14,6 +16,7 @@ import subprocess
import sys
import tempfile
import time
from typing import Literal, Optional
import uuid
VERSION = 1
@@ -22,6 +25,10 @@ ACTIVE = ('running', 'recovery_required')
TERMINAL = ('completed', 'failed', 'recovered')
GIT_CONTEXT = ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE', 'GIT_PREFIX')
Lookup = Literal['ok', 'gone', 'unreadable']
AliveStatus = Literal['alive', 'dead', 'unknown']
DescentStatus = Literal['yes', 'no', 'unknown']
class Denied(Exception):
def __init__(self, code, message, **details):
@@ -53,46 +60,209 @@ def git(path, *args, index=False, check=True, input=None):
return result.stdout.rstrip('\n') if check else result
def proc(pid):
@dataclass(frozen=True)
class ProcInfo:
"""Process identity; start is an opaque token comparable within the same boot."""
pid: int
ppid: int
pgid: int
state: str
start: str
# ps columns: pid ppid pgid lstart state — lstart has spaces, state is last token.
_PS_LINE = re.compile(r'\s*(\d+)\s+(\d+)\s+(\d+)\s+(.+?)\s+(\S+)\s*$')
def _normalize_state(state):
return (state or '?')[0]
def _linux_read_proc(pid):
path = Path('/proc', str(pid), 'stat')
try:
raw = Path('/proc', str(pid), 'stat').read_text().rsplit(')', 1)[1].split()
return {'state': raw[0], 'ppid': int(raw[1]), 'group': int(raw[2]), 'start': raw[19]}
except (OSError, ValueError, IndexError):
raw = path.read_text().rsplit(')', 1)[1].split()
return 'ok', ProcInfo(pid=pid, ppid=int(raw[1]), pgid=int(raw[2]),
state=_normalize_state(raw[0]), start=raw[19])
except FileNotFoundError:
return 'gone', None
except (PermissionError, IsADirectoryError):
return 'unreadable', None
except OSError as exc:
if getattr(exc, 'errno', None) in (errno.ENOENT, errno.ESRCH):
return 'gone', None
return 'unreadable', None
except (ValueError, IndexError):
return 'unreadable', None
def _parse_ps_line(line):
match = _PS_LINE.match(line)
if not match:
return None
pid, ppid, pgid, start, state = match.groups()
return ProcInfo(pid=int(pid), ppid=int(ppid), pgid=int(pgid),
state=_normalize_state(state), start=start.strip())
def boot():
return Path('/proc/sys/kernel/random/boot_id').read_text().strip()
def _darwin_read_proc(pid):
try:
result = subprocess.run(
['ps', '-o', 'pid=,ppid=,pgid=,lstart=,state=', '-p', str(pid)],
capture_output=True, text=True)
except OSError:
return 'unreadable', None
# BSD ps: nonexistent pid → nonzero + empty stdout. Any other opacity → unreadable.
if result.returncode != 0 and not result.stdout.strip():
# Errno-style chatter (e.g. "process id too large") still means no such process.
return 'gone', None
if not result.stdout.strip():
return 'unreadable', None
info = _parse_ps_line(result.stdout.splitlines()[0])
if not info or info.pid != int(pid):
return 'unreadable', None
return 'ok', info
def alive(pid, start, boot_id):
info = proc(pid)
return bool(info and info['state'] != 'Z' and info['start'] == start and boot_id == boot())
def boot_epoch():
"""Opaque boot generation token; DB field `boot` stays compatible."""
if sys.platform == 'darwin':
try:
raw = subprocess.check_output(['sysctl', '-n', 'kern.boottime'], text=True)
except (OSError, subprocess.CalledProcessError) as exc:
deny('PROCESS_UNREADABLE', f'Unable to read kern.boottime: {exc}')
match = re.search(r'sec\s*=\s*(\d+)', raw)
if not match:
deny('PROCESS_UNREADABLE', 'Unable to parse kern.boottime')
return match.group(1)
if sys.platform.startswith('linux'):
try:
return Path('/proc/sys/kernel/random/boot_id').read_text().strip()
except OSError as exc:
deny('PROCESS_UNREADABLE', f'Unable to read boot_id: {exc}')
deny('UNSUPPORTED_PLATFORM', f'Boot epoch unsupported on {sys.platform}')
def descendants_of(pid):
def read_proc(pid) -> tuple[Lookup, Optional[ProcInfo]]:
"""Lookup a single process. unreadable must never be treated as exited."""
if sys.platform == 'darwin':
return _darwin_read_proc(pid)
if sys.platform.startswith('linux'):
return _linux_read_proc(pid)
deny('UNSUPPORTED_PLATFORM', f'Process inspection unsupported on {sys.platform}')
def _enumerate_group(pgid):
"""Yield ProcInfo in pgid. Returns (saw_member_alive, enum_status).
enum_status is 'ok' when every observable entry was readable, else 'unknown'.
Fail-closed: enum failure or any unreadable member never reports 'all dead'.
"""
if sys.platform == 'darwin':
try:
result = subprocess.run(
['ps', '-axo', 'pid=,ppid=,pgid=,lstart=,state='],
capture_output=True, text=True)
except OSError:
return False, 'unknown'
if result.returncode != 0:
return False, 'unknown'
alive_member = False
for line in result.stdout.splitlines():
if not line.strip():
continue
info = _parse_ps_line(line)
if info is None:
return alive_member, 'unknown'
if info.pgid == pgid and info.state != 'Z':
alive_member = True
return alive_member, 'ok'
if sys.platform.startswith('linux'):
try:
items = list(Path('/proc').iterdir())
except OSError:
return False, 'unknown'
alive_member, saw_unreadable = False, False
for item in items:
if not item.name.isdigit():
continue
lookup, info = _linux_read_proc(int(item.name))
if lookup == 'unreadable':
saw_unreadable = True
continue
if lookup == 'ok' and info and info.pgid == pgid and info.state != 'Z':
alive_member = True
return alive_member, ('unknown' if saw_unreadable else 'ok')
deny('UNSUPPORTED_PLATFORM', f'Process enumeration unsupported on {sys.platform}')
def alive_status(pid, start, boot_id) -> AliveStatus:
if boot_id != boot_epoch():
return 'dead'
lookup, info = read_proc(pid)
if lookup == 'gone':
return 'dead'
if lookup == 'unreadable':
return 'unknown'
if not info or info.state == 'Z' or info.start != start:
return 'dead'
return 'alive'
def group_status(pgid) -> AliveStatus:
if not pgid:
return 'dead'
alive_member, enum_status = _enumerate_group(pgid)
if alive_member:
return 'alive'
if enum_status != 'ok':
return 'unknown'
return 'dead'
def descent_status(ancestor) -> DescentStatus:
current = os.getpid()
seen = set()
while current > 1 and current not in seen:
if current == pid:
return True
if current == ancestor:
return 'yes'
seen.add(current)
info = proc(current)
if not info:
break
current = info['ppid']
return False
lookup, info = read_proc(current)
if lookup == 'unreadable':
return 'unknown'
if lookup == 'gone' or not info:
return 'no'
current = info.ppid
return 'no'
def boot():
return boot_epoch()
def proc(pid):
"""Compat: ok→dict(start/ppid/group/state); gone→None; unreadable→deny (fail-closed)."""
lookup, info = read_proc(pid)
if lookup == 'gone':
return None
if lookup != 'ok' or not info:
deny('PROCESS_UNREADABLE', f'Unable to inspect process {pid}')
return {'pid': info.pid, 'ppid': info.ppid, 'group': info.pgid, 'state': info.state, 'start': info.start}
def alive(pid, start, boot_id):
"""True when the process may still hold resources (alive or unknown)."""
return alive_status(pid, start, boot_id) != 'dead'
def group_alive(group):
if not group:
return False
for item in Path('/proc').iterdir():
if item.name.isdigit():
info = proc(int(item.name))
if info and info['group'] == group and info['state'] != 'Z':
return True
return False
"""True when the group may still hold resources (alive or unknown)."""
return group_status(group) != 'dead'
def descendants_of(pid):
return descent_status(pid) == 'yes'
def relative(value):
@@ -403,7 +573,9 @@ class Lease:
self.handles = []
def __enter__(self):
info = proc(os.getpid())
lookup, info = read_proc(os.getpid())
if lookup != 'ok' or not info:
deny('PROCESS_UNREADABLE', 'Cannot observe supervisor process', lookup=lookup)
with self.repo.db(True) as db:
old = db.execute('SELECT * FROM runs WHERE id=?', (self.request_id,)).fetchone()
if old:
@@ -415,12 +587,12 @@ class Lease:
if old['state'] not in ('busy',):
deny('RECOVERY_REQUIRED', 'Operation already exists; query or recover it')
db.execute("UPDATE runs SET state='queued',pid=?,start=?,boot=?,heartbeat=? WHERE id=?",
(os.getpid(), info['start'], boot(), time.time(), self.request_id))
(os.getpid(), info.start, boot_epoch(), time.time(), self.request_id))
else:
db.execute('''INSERT INTO runs(id,task,fingerprint,resources,mode,state,pid,start,boot,created,heartbeat,kind)
VALUES(?,?,?,?,?,'queued',?,?,?,?,?,?)''',
(self.request_id, self.task['id'], self.fingerprint, json.dumps(self.resources),
self.mode, os.getpid(), info['start'], boot(), time.time(), time.time(), self.kind))
self.mode, os.getpid(), info.start, boot_epoch(), time.time(), time.time(), self.kind))
self.row = dict(db.execute('SELECT * FROM runs WHERE id=?', (self.request_id,)).fetchone())
deadline = time.monotonic() + self.wait
holder = None
@@ -442,7 +614,8 @@ class Lease:
candidates = db.execute("SELECT * FROM runs WHERE id!=? AND state IN ('running','recovery_required','queued') ORDER BY generation", (self.request_id,)).fetchall()
for other in candidates:
if other['state'] == 'queued':
if not alive(other['pid'], other['start'], other['boot']):
# Only reclaim when proven dead; unknown keeps the slot.
if alive_status(other['pid'], other['start'], other['boot']) == 'dead':
db.execute("UPDATE runs SET state='busy' WHERE id=?", (other['id'],))
continue
if other['generation'] > self.row['generation']:
@@ -485,7 +658,9 @@ class Lease:
doc = db.execute('SELECT state FROM documents WHERE run=?', (self.request_id,)).fetchone()
state = db.execute('SELECT * FROM runs WHERE id=?', (self.request_id,)).fetchone()
if state and state['state'] == 'running':
self.finish(1, bool(doc and doc['state'] == 'prepared') or group_alive(state['pgid']))
# unknown/alive both require recovery; only proven-dead releases cleanly.
remaining = group_status(state['pgid']) != 'dead'
self.finish(1, bool(doc and doc['state'] == 'prepared') or remaining)
finally:
self.unlock()
else:
@@ -498,11 +673,12 @@ def inherited(repo, required=None):
row = db.execute('SELECT * FROM runs WHERE id=?', (run_id,)).fetchone()
if not row or row['state'] != 'running':
deny('LEASE_REQUIRED', 'Run the operation through agentctl with')
# Fail-closed: unknown supervisor or uncertain descent denies the lease identity.
if (os.getenv('AGENTCTL_TASK_ID') != row['task']
or os.getenv('AGENTCTL_GENERATION') != str(row['generation'])
or os.getenv('AGENTCTL_EPOCH') != repo.identity()['epoch']
or not alive(row['pid'], row['start'], row['boot'])
or not descendants_of(row['pid'])):
or alive_status(row['pid'], row['start'], row['boot']) != 'alive'
or descent_status(row['pid']) != 'yes'):
deny('STALE_GENERATION', 'Lease identity is stale or caller is not a supervised descendant')
if required:
resources = json.loads(row['resources'])
@@ -530,10 +706,11 @@ def supervise(repo, lease, command, cwd, env=None):
if stopping:
return 128 + stopping[0], False
child = subprocess.Popen(command, cwd=cwd, env=env, start_new_session=True)
info = proc(child.pid)
lookup, info = read_proc(child.pid)
child_start = info.start if lookup == 'ok' and info else ''
with repo.db(True) as db:
db.execute('UPDATE runs SET child_pid=?,child_start=?,pgid=? WHERE id=?',
(child.pid, info['start'] if info else '', child.pid, lease.request_id))
(child.pid, child_start, child.pid, lease.request_id))
last_beat, stop_at = time.monotonic(), None
while child.poll() is None:
if stopping:
@@ -546,7 +723,8 @@ def supervise(repo, lease, command, cwd, env=None):
last_beat = time.monotonic()
time.sleep(0.1)
code = child.wait()
return (code if code >= 0 else 128 - code), group_alive(child.pid)
# remaining=True when alive or unknown — never treat opaque enum as all-dead.
return (code if code >= 0 else 128 - code), group_status(child.pid) != 'dead'
except BaseException:
if child is not None:
with contextlib.suppress(ProcessLookupError):
@@ -791,7 +969,10 @@ def recover(repo, args):
if row['state'] in TERMINAL:
emit({'status': row['state'], 'run_id': args.run})
return
if alive(row['pid'], row['start'], row['boot']) or (row['boot'] == boot() and group_alive(row['pgid'])):
# Deny unless supervisor and (same-boot) child group are proven dead.
if alive_status(row['pid'], row['start'], row['boot']) != 'dead':
deny('LOCK_BUSY', 'Supervisor or child process group is still alive; no takeover')
if row['boot'] == boot_epoch() and group_status(row['pgid']) != 'dead':
deny('LOCK_BUSY', 'Supervisor or child process group is still alive; no takeover')
handles = []
try:
@@ -993,7 +1174,7 @@ def main():
with repo.db() as db:
tasks=[dict(r) for r in db.execute('SELECT * FROM tasks')]
runs=[dict(r) for r in db.execute("SELECT * FROM runs WHERE state IN ('running','queued','recovery_required','busy')")]
for row in runs:row['supervisor_alive']=alive(row['pid'],row['start'],row['boot'])
for row in runs:row['supervisor_alive']=alive_status(row['pid'],row['start'],row['boot'])
emit({'status':'ok','lock_root':str(repo.root),'main':str(repo.main),'version':VERSION,'epoch':identity['epoch'],'tasks':tasks,'runs':runs})
elif args.action=='task':
if args.task_action=='register':
+24 -11
View File
@@ -64,9 +64,9 @@ class WorktreeTest(unittest.TestCase):
return Path(json.loads(result.stdout)["worktree"])
def remove_with_visibility_fixture(self, *args):
"""Fixture only /proc enumeration; Git, registry and lock checks stay real.
"""Fixture only process cwd enumeration; Git, registry and lock checks stay real.
This host hides unrelated same-user daemon cwds. Positive removal tests
This host may hide unrelated same-user daemon cwds. Positive removal tests
model a fully visible process namespace containing the helper itself.
Live cwd rejection below still runs the actual host process enumeration.
"""
@@ -77,13 +77,10 @@ from unittest import mock
spec = importlib.util.spec_from_file_location("helper", sys.argv[1])
helper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(helper)
original = Path.iterdir
def visible_processes(path):
if path == Path("/proc"):
return iter([path / str(os.getpid())])
return original(path)
def visible_cwds():
yield str(os.getpid()), Path.cwd()
sys.argv = sys.argv[1:]
with mock.patch.object(Path, "iterdir", visible_processes):
with mock.patch.object(helper, "iter_same_user_cwds", visible_cwds):
sys.exit(helper.main())
'''
return self.run_cmd([sys.executable, "-c", code, str(self.root / "scripts/agent_worktree.py"), "remove", *args])
@@ -110,7 +107,7 @@ with mock.patch.object(Path, "iterdir", visible_processes):
self.assertEqual(result.returncode, 0, result.stderr)
data = json.loads(result.stdout)
self.assertEqual(data["base_dev_sha"], self.base)
self.assertEqual(Path(data["worktree"]), self.root / ".worktrees/second")
self.assertEqual(Path(data["worktree"]).resolve(), (self.root / ".worktrees/second").resolve())
self.assertEqual(self.git("rev-parse", "fix/second").stdout.strip(), self.base)
self.assertFalse((first / ".worktrees").exists())
self.assertFalse((first / ".locks").exists())
@@ -308,12 +305,28 @@ class ProcessVisibilityTest(unittest.TestCase):
spec = importlib.util.spec_from_file_location("agent_worktree_under_test", SOURCE / "agent_worktree.py")
helper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(helper)
def unreadable():
raise helper.Refused("cannot verify cwd of same-user process 987654321; cleanup_pending")
with tempfile.TemporaryDirectory() as temp:
with mock.patch.object(helper, "iter_same_user_cwds", side_effect=unreadable):
with self.assertRaisesRegex(helper.Refused, "cannot verify cwd"):
helper.assert_no_process_cwd(Path(temp) / "task")
def test_linux_permission_error_path_remains_fail_closed(self):
spec = importlib.util.spec_from_file_location("agent_worktree_under_test", SOURCE / "agent_worktree.py")
helper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(helper)
with tempfile.TemporaryDirectory() as temp:
proc = Path(temp) / "987654321"
proc.mkdir()
(proc / "comm").write_text("coding-agent\n")
with mock.patch.object(helper.Path, "iterdir", return_value=iter([proc])), \
mock.patch.object(helper.os, "readlink", side_effect=PermissionError):
with mock.patch.object(helper.sys, "platform", "linux"), \
mock.patch.object(helper.Path, "iterdir", return_value=iter([proc])), \
mock.patch.object(helper.os, "getuid", return_value=0), \
mock.patch.object(helper.os, "readlink", side_effect=PermissionError), \
mock.patch.object(Path, "stat", lambda self: type("S", (), {"st_uid": 0})()):
with self.assertRaisesRegex(helper.Refused, "cannot verify cwd"):
helper.assert_no_process_cwd(Path(temp) / "task")
+100
View File
@@ -258,6 +258,8 @@ class AgentctlTests(unittest.TestCase):
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"])
@@ -533,6 +535,8 @@ class AgentctlTests(unittest.TestCase):
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")
@@ -650,5 +654,101 @@ class AgentctlTests(unittest.TestCase):
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()
+3 -1
View File
@@ -22,8 +22,10 @@ class HerdrTests(unittest.TestCase):
self.task = self.initialized_task()
self.bin = Path(self.temp.name) / 'bin'; self.bin.mkdir()
self.log = Path(self.temp.name) / 'calls.jsonl'
# Resolve so macOS /var → /private/var matches agentctl Repo.current strings.
task_cwd = str(self.task.resolve())
self.env = dict(os.environ, HERDR_ENV='1', PATH=str(self.bin)+os.pathsep+os.environ['PATH'],
MOCK_LOG=str(self.log), MOCK_CWD=str(self.task), MOCK_MODE='ok')
MOCK_LOG=str(self.log), MOCK_CWD=task_cwd, MOCK_MODE='ok')
self.env['MOCK_AGENTS'] = json.dumps(['qwen_max__alpha','qwen_flash__alpha','context_kimi__alpha','implementer_cursor__alpha','reviewer_glm__alpha'])
fake = self.bin / 'herdr'
fake.write_text("#!/usr/bin/env python3\n" + r'''import json,os,re,sys