merge(dev): sync governance-automation onto current local dev

scripts/agent_worktree.py

@@ -121,26 +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):
-    status = inspect_process_cwd(path)
-    if status['state'] == 'clear':
-        return
-    if status['state'] == 'occupied':
-        raise Refused(f"process {status['pid']} still uses this worktree as cwd")
-    if status['state'] == 'unreadable':
-        raise Refused(f"cannot verify cwd of same-user process {','.join(status['unreadable'])}; cleanup_pending")
-    raise Refused(status.get('message') or f"process cwd check blocked ({status['state']}); cleanup_pending")
-
-
-def inspect_process_cwd(path):
-    """Structured process-cwd visibility for cleanup reports (never force-clears)."""
-    path = Path(path).resolve()
-    if not sys.platform.startswith('linux'):
-        return {
-            'state': 'unsupported_platform',
-            'platform': sys.platform,
-            'message': 'process cwd enumeration requires Linux /proc; macOS agentctl W1-B not merged yet',
-        }
+def _linux_same_user_cwds():
     unreadable = []
     for proc in Path("/proc").iterdir():
         if not proc.name.isdigit():
@@ -149,11 +130,13 @@ def inspect_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:
-                return {'state': 'occupied', 'pid': proc.name}
+            yield proc.name, cwd
         except (FileNotFoundError, ProcessLookupError):
             continue
         except PermissionError:
+            # Linux may hide cwd for the per-user OS service manager despite a
+            # matching UID. This narrowly named OS process is not a task agent;
+            # all other unreadable same-user processes remain fail-closed.
             try:
                 if (proc / "comm").read_text().strip() == "systemd":
                     print(f"[notice] OS user service manager {proc.name} hides cwd; "
@@ -163,8 +146,57 @@ def inspect_process_cwd(path):
                 pass
             unreadable.append(proc.name)
     if unreadable:
-        return {'state': 'unreadable', 'unreadable': unreadable}
-    return {'state': 'clear'}
+        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
@@ -227,135 +259,10 @@ def remove_checked(repo, main, branch, delete_branch, path):
     print(json.dumps({"removed": str(path), "branch": branch, "branch_deleted": delete_branch}))
 
 
-def inspect_worktree(repo, main, entry):
-    """Read-only cleanup readiness for one porcelain worktree entry."""
-    path = Path(entry.get("worktree", ""))
-    branch_ref = entry.get("branch", "")
-    report = {
-        "path": str(path),
-        "branch": branch_ref[len("refs/heads/"):] if branch_ref.startswith("refs/heads/") else branch_ref or None,
-        "verdict": "blocked",
-        "blockers": [],
-        "notes": [],
-    }
-    if "bare" in entry:
-        report["blockers"].append("bare_worktree")
-        return report
-    try:
-        safe_directory(main, path.resolve())
-    except Refused as exc:
-        report["blockers"].append(f"path:{exc}")
-        return report
-    if "locked" in entry:
-        report["blockers"].append("git_locked")
-    if "prunable" in entry:
-        report["blockers"].append("git_prunable")
-    if not path.is_dir():
-        report["blockers"].append("missing_directory")
-        return report
-    if not report["branch"]:
-        report["blockers"].append("detached_or_unmapped_branch")
-        return report
-    try:
-        actual = git(path, "symbolic-ref", "--quiet", "HEAD").stdout.strip()
-        if actual != f"refs/heads/{report['branch']}":
-            report["blockers"].append("branch_identity_mismatch")
-        head = git(path, "rev-parse", "HEAD").stdout.strip()
-        report["head"] = head
-        if git(main, "merge-base", "--is-ancestor", head, "refs/heads/dev", check=False).returncode:
-            report["blockers"].append("not_ancestor_of_dev")
-        else:
-            report["notes"].append("ancestor_of_dev")
-        for marker in ("MERGE_HEAD", "CHERRY_PICK_HEAD", "REVERT_HEAD", "rebase-merge", "rebase-apply", "BISECT_START"):
-            marker_path = Path(git(path, "rev-parse", "--path-format=absolute", "--git-path", marker).stdout.strip())
-            if marker_path.exists():
-                report["blockers"].append(f"unfinished_git:{marker}")
-        dirty = git(path, "status", "--porcelain=v1", "--untracked-files=all", "--ignored=matching").stdout
-        if dirty:
-            report["blockers"].append("dirty_or_ignored_files")
-            report["dirty_sample"] = dirty.splitlines()[:8]
-    except Refused as exc:
-        report["blockers"].append(f"git:{exc}")
-    check = control(repo, "worktree-check", "--path", str(path), check=False)
-    if check.returncode:
-        try:
-            payload = json.loads(check.stderr.strip().splitlines()[-1])
-            report["blockers"].append(f"registry:{payload.get('code', 'DENIED')}")
-            if payload.get("message"):
-                report["notes"].append(payload["message"])
-        except (ValueError, IndexError, json.JSONDecodeError):
-            report["blockers"].append("registry:worktree_check_failed")
-    else:
-        report["notes"].append("registry_clear")
-    process = inspect_process_cwd(path)
-    report["process"] = process
-    if process["state"] == "occupied":
-        report["blockers"].append(f"process_cwd:{process['pid']}")
-    elif process["state"] == "unreadable":
-        report["blockers"].append("process_cwd_unreadable")
-    elif process["state"] == "unsupported_platform":
-        report["blockers"].append("process_cwd_platform_pending")
-        report["notes"].append(process.get("message") or "platform process check pending")
-    key = hashlib.sha256(str(path.resolve()).encode()).hexdigest()
-    lock_path = main / ".locks" / "os" / f"worktree-{key}.lock"
-    if lock_path.exists():
-        fd = None
-        try:
-            fd = os.open(lock_path, os.O_RDWR | os.O_NOFOLLOW, 0o600)
-            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-            fcntl.flock(fd, fcntl.LOCK_UN)
-            report["notes"].append("lifecycle_lock_idle")
-        except BlockingIOError:
-            report["blockers"].append("lifecycle_lock_busy")
-        except OSError as exc:
-            report["blockers"].append(f"lifecycle_lock:{exc}")
-        finally:
-            if fd is not None:
-                os.close(fd)
-    if not report["blockers"]:
-        report["verdict"] = "ready"
-    return report
-
-
-def cleanup_report(repo, main):
-    """Surface cleanup readiness without deleting anything."""
-    try:
-        control(repo, "status")
-        registry = "initialized"
-    except Exception:
-        registry = "unavailable"
-    items = []
-    for entry in worktrees(main):
-        path = entry.get("worktree")
-        if not path:
-            continue
-        resolved = Path(path).resolve()
-        if resolved == main.resolve():
-            continue
-        if resolved.parent != (main / ".worktrees").resolve():
-            items.append({
-                "path": str(resolved),
-                "branch": entry.get("branch"),
-                "verdict": "blocked",
-                "blockers": ["outside_dot_worktrees"],
-                "notes": ["not managed by agent-worktree cleanup"],
-            })
-            continue
-        items.append(inspect_worktree(repo, main, entry))
-    ready = sum(1 for item in items if item.get("verdict") == "ready")
-    print(json.dumps({
-        "status": "ok",
-        "registry": registry,
-        "worktrees": items,
-        "summary": {"total": len(items), "ready": ready, "blocked": len(items) - ready},
-    }, ensure_ascii=False, sort_keys=True))
-
-
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     commands = parser.add_subparsers(dest="command", required=True)
     commands.add_parser("list", help="read-only Git worktree listing")
-    commands.add_parser("cleanup-report", help="read-only cleanup readiness for .worktrees entries")
     for name in ("create", "remove"):
         child = commands.add_parser(name)
         child.add_argument("branch")
@@ -368,9 +275,6 @@ def main():
         if args.command == "list":
             print(git(root, "worktree", "list").stdout, end="")
             return 0
-        if args.command == "cleanup-report":
-            cleanup_report(repo, root)
-            return 0
         safe_branch(args.branch)
         with write_lock(repo, root):
             if args.command == "create":

scripts/agentctl.py

@@ -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
 
 # Pure W4 helpers (deploy/projection/PI checks) live beside this file for tests.
@@ -43,6 +46,10 @@ CREATE TABLE IF NOT EXISTS queue_items(
 CREATE INDEX IF NOT EXISTS queue_pending ON queue_items(state, priority DESC, created);
 '''
 
+Lookup = Literal['ok', 'gone', 'unreadable']
+AliveStatus = Literal['alive', 'dead', 'unknown']
+DescentStatus = Literal['yes', 'no', 'unknown']
+
 
 class Denied(Exception):
     def __init__(self, code, message, **details):
@@ -74,46 +81,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):
@@ -433,7 +603,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:
@@ -445,12 +617,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
@@ -472,7 +644,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']:
@@ -515,7 +688,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:
@@ -528,11 +703,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'])
@@ -560,10 +736,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:
@@ -576,7 +753,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):
@@ -821,7 +999,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:
@@ -1378,7 +1559,8 @@ def main():
     elif args.action in ('status','doctor'):
         identity=repo.identity()
         try:
-            ensure_schema(repo)
+            with repo.db(True) as db:
+                ensure_schema(db)
         except Denied:
             pass
         with repo.db() as db:
@@ -1389,7 +1571,7 @@ def main():
                     "SELECT * FROM queue_items WHERE state IN ('pending','claimed') ORDER BY priority DESC, created")]
             except sqlite3.Error:
                 queue=[]
-        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,'queue':queue})
     elif args.action=='task':

scripts/run-nightly-real-pi.sh

@@ -26,9 +26,9 @@ RUNNER_URL="${PI_AGENT_RUNNER_URL:-http://127.0.0.1:18865}"
 TOKEN="${PI_AGENT_RUNNER_AUTH_TOKEN:-${RUNNER_AUTH_TOKEN:-}}"
 MODEL_ID="${PI_AGENT_MODEL:-MiniMax-M2.7-highspeed}"
 PROTOCOL="${PI_AGENT_PROTOCOL:-typed-result-v1}"
-TIMEOUT="${NIGHTLY_SUITE_TIMEOUT:-90m}"
+TIMEOUT="${NIGHTLY_SUITE_TIMEOUT:-180m}"
 COMMIT="$(git rev-parse HEAD)"
-SUITE="nightly-real-pi-u8-fullnone"
+SUITE="nightly-real-pi-u8-four-workflows"
 STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
 EVIDENCE_DIR="${NIGHTLY_EVIDENCE_DIR:-${ROOT}/.doc/test-reports/assets/promotion-receipts/evidence-${COMMIT:0:12}-${STAMP}}"
 mkdir -p "${EVIDENCE_DIR}"
@@ -172,7 +172,9 @@ preflight_shared_kb
 # Clear env pollution that flakes short/config tests elsewhere.
 unset AGENT_TIMEOUT KB_DATABASE_URL KB_EMBEDDING_DIM 2>/dev/null || true
 
-RUN_FILTER='TestMultiIntentU8_RealPI_ConfirmGenerate_PG|TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain'
+# Four workflow governance chains (main / product_manager / test_manager / FullNone)
+# plus U8 confirm-generate. Rdev P3 remains the Python harness (not yet a Go PG top-level).
+RUN_FILTER='TestMultiIntentU8_RealPI_ConfirmGenerate_PG|TestWorkflowGovernance_PiAgent_PG_(FullChain|ProductManager_FullChain|TestManager_FullChain|FullNoneMixedChain)$'
 LOG_FILE="${EVIDENCE_DIR}/go-test.txt"
 set +e
 go test -count=1 -timeout "${TIMEOUT}" -v ./internal/server/ -run "${RUN_FILTER}" 2>&1 | tee "${LOG_FILE}"
@@ -190,9 +192,12 @@ fi
 set -e
 
 if [[ "${TEST_RC}" -eq 0 ]]; then
-  # Exit zero alone also covers no tests and partial skips; require both named tests.
+  # Exit zero alone also covers no tests and partial skips; require all named top-level tests.
   if ! python3 "${ROOT}/scripts/verify_go_test_evidence.py" --log "${RESULT_LOG}" \
       --expected TestMultiIntentU8_RealPI_ConfirmGenerate_PG \
+      --expected TestWorkflowGovernance_PiAgent_PG_FullChain \
+      --expected TestWorkflowGovernance_PiAgent_PG_ProductManager_FullChain \
+      --expected TestWorkflowGovernance_PiAgent_PG_TestManager_FullChain \
       --expected TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain \
       > "${EVIDENCE_DIR}/expected-tests.json"; then
     write_receipt infra_fail infra "${LOG_FILE}" "${EVIDENCE_DIR}/expected-tests.json"

scripts/tests/test_agent_worktree.py

@@ -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")
 

scripts/tests/test_agentctl.py

@@ -259,6 +259,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"])
@@ -534,6 +536,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")
@@ -651,5 +655,102 @@ 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")
+        shutil.copy2(SOURCE.parent / "agentctl_w4.py", root / "scripts" / "agentctl_w4.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()

scripts/tests/test_herdr_agents.py

@@ -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