feat(ops): advance governance automation queue and projection increments

.doc/ops/short-term/2026-09-22-governance-automation.md

@@ -0,0 +1,69 @@
+---
+title: "W4 治理自动化剩余：持久队列与闭环"
+task_id: governance-automation-2026-09-22
+status: in_progress
+created: 2026-09-22
+updated: 2026-09-22
+owner: Cursor
+task_type: implementation
+task_tier: multi_stage
+policy_revision: "agent-governance-2026-09-21-r1 @ 364ed7f1a3f0393ef77ec4623a9518116ed2cb87"
+branch: feature/governance-automation-2026-09-22
+worktree: .worktrees/governance-automation-2026-09-22
+base_dev_sha: 364ed7f1a3f0393ef77ec4623a9518116ed2cb87
+write_scope: [".doc", "scripts", "docker-compose.dev.yml", "docker-compose.dev.pi-image.yml"]
+allowed_environments: []
+head_sha: ""
+mutex: agent-workflow-governance-automation
+review_state: pending
+integration_state: pending
+deployment: not_required
+cleanup: pending
+cleanup_owner: "Cursor 之外的集成/维护执行者"
+cleanup_trigger: "合入 dev、验收通过且 owner 退出后核对清理门禁"
+remote_sync: not_requested
+---
+
+# W4 治理自动化剩余：持久队列与闭环
+
+## 用户目标与授权
+
+- 用户目标：推进 agent-workflow-governance 的 W1–W4 遗留——后台持久队列与自动调度、通用文档投影、业务任务部署/恢复闭环、产品 PI 源码镜像交付、清理可见性。
+- 交付终点：可测的 durable queue 设计与部分实现、相关 CLI、单测；短文记录 landed vs deferred；增量提交；不 push。
+- 禁止：push；强删 worktree；操作未授权环境；宣称完整 W3/W4 已关闭。
+- 依赖：sibling `agentctl-macos-2026-09-22` 修 `/proc`；本任务 Linux-first，macOS 仅设计与 fail-closed，不阻塞整体。
+
+## 范围与非目标
+
+- 允许：`scripts/agentctl*.py`、`scripts/agent_worktree.py`、`scripts/tests/*`、compose PI image override、本任务/长期 backlog 文档。
+- 非目标：OS 沙箱、跨机队列、真实 dev 重建、FEATURE-MATRIX 升 done、合入/push。
+
+## 步骤与验收
+
+1. [ ] Durable queue schema + enqueue/list/cancel/tick + 单测
+2. [ ] Generic doc projection（registry + project）
+3. [ ] Deploy/recover operation 状态机 + 收据
+4. [ ] PI 源码镜像交付 compose profile + 检查入口
+5. [ ] Cleanup visibility report
+6. [ ] short-term 写明 landed vs deferred；提交
+
+## 验证与证据
+
+| 范围 / 被测 SHA | 命令与环境 | 预期 | 实际 / 判定 | 证据 / 脱敏说明 |
+|---|---|---|---|---|
+| 治理脚本 | `python3 -B -m unittest discover -s scripts/tests -p 'test_*.py' -v` | 本任务新增用例绿；既有 /proc 相关在 Darwin 可 skip | pending | 本 worktree |
+
+## 收口与交接
+
+- 见文末「Landed vs deferred」。
+- mutex 与 sibling `scripts-agentctl-process` 分离；进程平台层不抢写。
+
+## 相关链接
+
+- [长期 W0–W4](../long-term/2026-09-21-agent-workflow-governance-automation.md)
+- [方案](../../remediation/2026-09-21-agent-workflow-governance-plan.md)
+- [sibling macOS](./2026-09-22-agentctl-macos.md)（路径在 sibling worktree）
+
+## Landed vs deferred
+
+（实现后回填）

scripts/agent_worktree.py

@@ -122,6 +122,25 @@ def create(repo, main, branch):
 
 
 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',
+        }
     unreadable = []
     for proc in Path("/proc").iterdir():
         if not proc.name.isdigit():
@@ -131,13 +150,10 @@ def assert_no_process_cwd(path):
                 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")
+                return {'state': 'occupied', 'pid': proc.name}
         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; "
@@ -147,7 +163,8 @@ def assert_no_process_cwd(path):
                 pass
             unreadable.append(proc.name)
     if unreadable:
-        raise Refused(f"cannot verify cwd of same-user process {','.join(unreadable)}; cleanup_pending")
+        return {'state': 'unreadable', 'unreadable': unreadable}
+    return {'state': 'clear'}
 
 
 @contextmanager
@@ -210,10 +227,135 @@ 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")
@@ -226,6 +368,9 @@ 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

@@ -16,11 +16,32 @@ import tempfile
 import time
 import uuid
 
+# Pure W4 helpers (deploy/projection/PI checks) live beside this file for tests.
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import agentctl_w4 as w4  # noqa: E402
+
 VERSION = 1
 SHARED_DOCS = {'.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'}
 ACTIVE = ('running', 'recovery_required')
 TERMINAL = ('completed', 'failed', 'recovered')
+QUEUE_TERMINAL = ('completed', 'failed', 'cancelled')
 GIT_CONTEXT = ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE', 'GIT_PREFIX')
+QUEUE_SCHEMA = '''
+CREATE TABLE IF NOT EXISTS queue_items(
+  id TEXT PRIMARY KEY,
+  task TEXT NOT NULL,
+  head_sha TEXT NOT NULL,
+  expected_dev TEXT NOT NULL,
+  priority INTEGER NOT NULL DEFAULT 0,
+  state TEXT NOT NULL,
+  created REAL NOT NULL,
+  updated REAL NOT NULL,
+  claim_run TEXT,
+  reason TEXT,
+  UNIQUE(task, head_sha)
+);
+CREATE INDEX IF NOT EXISTS queue_pending ON queue_items(state, priority DESC, created);
+'''
 
 
 class Denied(Exception):
@@ -226,6 +247,7 @@ class Repo:
             if write:
                 db.execute('PRAGMA synchronous=FULL')
                 db.execute('BEGIN IMMEDIATE')
+                ensure_schema(db)
             yield db
             if write:
                 db.commit()
@@ -274,8 +296,9 @@ def initialize(repo):
         fcntl.flock(initialization, fcntl.LOCK_EX | fcntl.LOCK_NB)
         if (repo.root / 'identity.json').exists():
             identity = repo.identity()
-            with repo.db() as db:
+            with repo.db(True) as db:
                 db.execute('SELECT count(*) FROM runs').fetchone()
+                db.execute('SELECT count(*) FROM queue_items').fetchone()
             emit({'status': 'initialized', 'lock_root': str(repo.root), **identity})
             return
         old = legacy / 'dev-integration.lock'
@@ -327,7 +350,7 @@ def initialize(repo):
                 CREATE TABLE IF NOT EXISTS freezes(resource TEXT PRIMARY KEY, owner TEXT, reason TEXT);
                 CREATE TABLE IF NOT EXISTS events(id INTEGER PRIMARY KEY AUTOINCREMENT,
                   created REAL NOT NULL, operation TEXT NOT NULL, detail TEXT NOT NULL);
-                ''')
+                ''' + QUEUE_SCHEMA)
                 db.commit()
             finally:
                 db.close()
@@ -338,6 +361,13 @@ def initialize(repo):
             emit({'status': 'initialized', 'lock_root': str(repo.root), **identity})
 
 
+def ensure_schema(db):
+    """Additive tables for durable queue + W4 deploy/projection; write txn only."""
+    for statement in filter(None, (part.strip() for part in QUEUE_SCHEMA.split(';'))):
+        db.execute(statement)
+    w4.ensure_schema(db)
+
+
 def task_row(repo, task_id):
     with repo.db() as db:
         row = db.execute('SELECT * FROM tasks WHERE id=?', (task_id,)).fetchone()
@@ -958,6 +988,348 @@ def integrate(repo, args):
         return code
 
 
+def queue_row(row):
+    return dict(row) if row is not None else None
+
+
+def queue_enqueue(repo, args):
+    task = task_row(repo, args.task)
+    if repo.current != Path(task['worktree']):
+        deny('WORKTREE_MISMATCH', 'Enqueue from the registered task worktree')
+    head = git(task['worktree'], 'rev-parse', 'HEAD')
+    if args.head and git(task['worktree'], 'rev-parse', args.head + '^{commit}') != head:
+        deny('BASE_CHANGED', 'Requested head is not the owning task HEAD')
+    expected = git(repo.main, 'rev-parse', args.expected_dev + '^{commit}')
+    admit_candidate(repo, task, head, expected)
+    item_id = args.request_id or str(uuid.uuid4())
+    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', item_id):
+        deny('INVALID_REQUEST', 'Invalid queue request id')
+    now = time.time()
+    with repo.db(True) as db:
+        existing = db.execute(
+            "SELECT * FROM queue_items WHERE task=? AND head_sha=? AND state NOT IN ('cancelled')",
+            (args.task, head)).fetchone()
+        if existing:
+            if existing['state'] in ('pending', 'claimed') and existing['expected_dev'] == expected:
+                emit({'status': 'duplicate', 'item': queue_row(existing)})
+                return 0
+            deny('QUEUE_CONFLICT', 'A non-cancelled queue item already covers this task head',
+                 item=queue_row(existing))
+        if db.execute('SELECT id FROM queue_items WHERE id=?', (item_id,)).fetchone():
+            deny('REQUEST_ID_REUSED', 'Queue request id already exists')
+        db.execute(
+            '''INSERT INTO queue_items(id,task,head_sha,expected_dev,priority,state,created,updated,claim_run,reason)
+               VALUES(?,?,?,?,?,'pending',?,?,NULL,NULL)''',
+            (item_id, args.task, head, expected, max(0, int(args.priority)), now, now))
+        event(db, item_id, {'event': 'queue_enqueued', 'task': args.task, 'head_sha': head})
+        row = db.execute('SELECT * FROM queue_items WHERE id=?', (item_id,)).fetchone()
+    emit({'status': 'enqueued', 'item': queue_row(row)})
+    return 0
+
+
+def queue_list(repo, args):
+    states = None if args.all else ('pending', 'claimed')
+    with repo.db(True) as db:
+        if states:
+            rows = db.execute(
+                "SELECT * FROM queue_items WHERE state IN (?,?) ORDER BY priority DESC, created ASC",
+                states).fetchall()
+        else:
+            rows = db.execute('SELECT * FROM queue_items ORDER BY created ASC').fetchall()
+    emit({'status': 'ok', 'items': [queue_row(r) for r in rows]})
+    return 0
+
+
+def queue_cancel(repo, args):
+    task_row(repo, args.task)
+    with repo.db(True) as db:
+        row = db.execute('SELECT * FROM queue_items WHERE id=?', (args.id,)).fetchone()
+        if not row:
+            deny('UNKNOWN_QUEUE_ITEM', 'Unknown queue item id')
+        if row['task'] != args.task:
+            deny('PATH_OUT_OF_SCOPE', 'Queue item belongs to another task')
+        if row['state'] in QUEUE_TERMINAL:
+            emit({'status': 'replayed', 'item': queue_row(row)})
+            return 0
+        if row['state'] == 'claimed':
+            deny('LOCK_BUSY', 'Queue item is claimed by an in-flight schedule; recover or wait')
+        db.execute("UPDATE queue_items SET state='cancelled',reason=?,updated=? WHERE id=?",
+                   (args.reason, time.time(), args.id))
+        event(db, args.id, {'event': 'queue_cancelled', 'reason': args.reason})
+        row = db.execute('SELECT * FROM queue_items WHERE id=?', (args.id,)).fetchone()
+    emit({'status': 'cancelled', 'item': queue_row(row)})
+    return 0
+
+
+def queue_schedule(repo, args):
+    """Claim the next durable queue item and run integrate once (or until drain/stop)."""
+    if repo.current != repo.main or git(repo.main, 'branch', '--show-current') != 'dev':
+        deny('MAIN_WORKTREE_REQUIRED', 'Schedule integrations only from the main dev worktree')
+    processed, results = 0, []
+    while True:
+        with repo.db(True) as db:
+            item = db.execute(
+                "SELECT * FROM queue_items WHERE state='pending' ORDER BY priority DESC, created ASC LIMIT 1"
+            ).fetchone()
+            if not item:
+                break
+            db.execute("UPDATE queue_items SET state='claimed',updated=?,claim_run=? WHERE id=? AND state='pending'",
+                       (time.time(), None, item['id']))
+            claimed = db.execute('SELECT * FROM queue_items WHERE id=?', (item['id'],)).fetchone()
+        item = queue_row(claimed)
+        try:
+            admit_candidate(repo, task_row(repo, item['task']), item['head_sha'], item['expected_dev'])
+        except Denied as exc:
+            with repo.db(True) as db:
+                db.execute("UPDATE queue_items SET state='failed',reason=?,updated=? WHERE id=?",
+                           (exc.code + ': ' + exc.message, time.time(), item['id']))
+                event(db, item['id'], {'event': 'queue_failed', 'code': exc.code})
+            results.append({'id': item['id'], 'status': 'failed', 'code': exc.code, 'message': exc.message})
+            processed += 1
+            if not args.drain:
+                break
+            continue
+        request_id = 'queue-' + item['id']
+        integrate_args = argparse.Namespace(task=item['task'], expected_dev=item['expected_dev'],
+                                            request_id=request_id)
+        try:
+            code = integrate(repo, integrate_args)
+        except Denied as exc:
+            with repo.db(True) as db:
+                state = 'pending' if exc.code == 'LOCK_BUSY' else 'failed'
+                db.execute("UPDATE queue_items SET state=?,reason=?,updated=?,claim_run=NULL WHERE id=?",
+                           (state, exc.code + ': ' + exc.message, time.time(), item['id']))
+                event(db, item['id'], {'event': 'queue_rescheduled' if state == 'pending' else 'queue_failed',
+                                       'code': exc.code})
+            results.append({'id': item['id'], 'status': state, 'code': exc.code, 'message': exc.message})
+            processed += 1
+            if exc.code == 'LOCK_BUSY' or not args.drain:
+                break
+            continue
+        with repo.db(True) as db:
+            if code == 0:
+                db.execute("UPDATE queue_items SET state='completed',updated=?,claim_run=?,reason=NULL WHERE id=?",
+                           (time.time(), request_id, item['id']))
+                event(db, item['id'], {'event': 'queue_completed', 'run_id': request_id})
+                results.append({'id': item['id'], 'status': 'completed', 'run_id': request_id})
+            else:
+                db.execute("UPDATE queue_items SET state='failed',updated=?,claim_run=?,reason=? WHERE id=?",
+                           (time.time(), request_id, 'integrate exit ' + str(code), item['id']))
+                event(db, item['id'], {'event': 'queue_failed', 'exit_code': code})
+                results.append({'id': item['id'], 'status': 'failed', 'exit_code': code, 'run_id': request_id})
+        processed += 1
+        if not args.drain:
+            break
+    emit({'status': 'ok', 'processed': processed, 'results': results})
+    return 0 if all(r.get('status') == 'completed' for r in results) or not results else 1
+
+
+def docs_project(repo, args):
+    """Project machine registry / queue into a docs-apply request for CURRENT (generic snapshot JSON)."""
+    task = task_row(repo, args.task)
+    if repo.current != Path(task['worktree']):
+        deny('WORKTREE_MISMATCH', 'Project documents from the registered task worktree')
+    filename = args.file
+    if filename not in SHARED_DOCS:
+        deny('PATH_OUT_OF_SCOPE', 'Only shared governance documents are projected')
+    if not filename.endswith('CURRENT.md'):
+        deny('INVALID_REQUEST', 'Generic projection currently emits CURRENT claim rows only')
+    content = safe_path(repo.current, filename).read_text()
+    existing = doc_rows(content, args.task)
+    expected = digest(existing[0][1]) if existing else None
+    with repo.db(True) as db:
+        pending = [queue_row(r) for r in db.execute(
+            "SELECT * FROM queue_items WHERE task=? AND state IN ('pending','claimed') ORDER BY priority DESC, created",
+            (args.task,)).fetchall()]
+        active_runs = [dict(r) for r in db.execute(
+            "SELECT id,state,resources,kind FROM runs WHERE task=? AND state IN ('running','queued','recovery_required','busy')",
+            (args.task,)).fetchall()]
+    queue_state = pending[0]['state'] if pending else 'idle'
+    status = args.status or ('queued' if pending else 'in_progress')
+    worktree_path = Path(task['worktree'])
+    try:
+        worktree_cell = str(worktree_path.relative_to(repo.main))
+    except ValueError:
+        worktree_cell = str(worktree_path)
+    replacement = (
+        f"| {task['owner']} | {task['id']} | {task['branch']} | {worktree_cell} "
+        f"| {args.mutex or '-'} | {status} |"
+    )
+    change = {
+        'file': filename,
+        'record_id': args.task,
+        'expected_sha256': expected,
+        'replacement': replacement,
+    }
+    machine = {
+        'task': {
+            'id': task['id'], 'owner': task['owner'], 'branch': task['branch'],
+            'worktree': task['worktree'], 'state': task['state'],
+            'scopes': json.loads(task['scopes']), 'capabilities': json.loads(task['capabilities']),
+            'head_sha': git(task['worktree'], 'rev-parse', 'HEAD'),
+        },
+        'queue': pending,
+        'runs': active_runs,
+        'queue_phase': queue_state,
+        'snapshot_time': time.time(),
+        'dev_sha': git(repo.main, 'rev-parse', 'dev'),
+    }
+    emit({
+        'status': 'projected',
+        'request': {'request_id': args.request_id or (args.task + '-project-1'), 'changes': [change]},
+        'machine': machine,
+    })
+    return 0
+
+
+def cleanup_process_gate(path):
+    """Report process-cwd visibility without deleting. Fail-closed unknowns are explicit."""
+    path = Path(path).resolve()
+    if not sys.platform.startswith('linux'):
+        return {
+            'status': 'unknown',
+            'platform': sys.platform,
+            'reason': 'process cwd enumeration requires platform support; cleanup_pending until verified',
+            'holders': [],
+            'unreadable': [],
+        }
+    holders, unreadable = [], []
+    try:
+        entries = list(Path('/proc').iterdir())
+    except OSError as exc:
+        return {
+            'status': 'unknown', 'platform': sys.platform,
+            'reason': f'cannot list /proc: {exc}', 'holders': [], 'unreadable': [],
+        }
+    for proc in entries:
+        if not proc.name.isdigit():
+            continue
+        try:
+            if proc.stat().st_uid != os.getuid():
+                continue
+            cwd = Path(os.readlink(proc / 'cwd'))
+            if cwd == path or path in cwd.parents:
+                holders.append(proc.name)
+        except (FileNotFoundError, ProcessLookupError):
+            continue
+        except PermissionError:
+            try:
+                if (proc / 'comm').read_text().strip() == 'systemd':
+                    continue
+            except (OSError, UnicodeError):
+                pass
+            unreadable.append(proc.name)
+    if holders:
+        return {'status': 'blocked', 'platform': sys.platform, 'holders': holders, 'unreadable': unreadable,
+                'reason': 'live process cwd uses worktree'}
+    if unreadable:
+        return {'status': 'unknown', 'platform': sys.platform, 'holders': [], 'unreadable': unreadable,
+                'reason': 'cannot verify cwd of same-user processes; cleanup_pending'}
+    return {'status': 'clear', 'platform': sys.platform, 'holders': [], 'unreadable': [], 'reason': None}
+
+
+def cleanup_status(repo, args):
+    """Structured cleanup gate report; never removes a worktree."""
+    if args.path:
+        target = Path(args.path).resolve()
+    else:
+        matches = [e for e in repo.worktrees() if e.get('branch') == f'refs/heads/{args.branch}']
+        if len(matches) != 1:
+            deny('UNKNOWN_WORKTREE', 'Expected exactly one worktree for branch', branch=args.branch)
+        target = Path(matches[0]['worktree']).resolve()
+    gates = []
+
+    def gate(name, status, detail=None, **extra):
+        item = {'name': name, 'status': status, 'detail': detail}
+        item.update(extra)
+        gates.append(item)
+
+    if not target.is_dir():
+        gate('path_exists', 'fail', 'worktree directory missing')
+        emit({'status': 'blocked', 'path': str(target), 'gates': gates, 'cleanup_allowed': False})
+        return 0
+
+    branch_result = git(target, 'branch', '--show-current', check=False)
+    head_result = git(target, 'rev-parse', 'HEAD', check=False)
+    branch = branch_result.stdout.strip() if branch_result.returncode == 0 else ''
+    head = head_result.stdout.strip() if head_result.returncode == 0 else ''
+    if not branch or not head:
+        gate('git_identity', 'fail', 'unable to read branch/HEAD')
+    else:
+        gate('git_identity', 'pass', branch=branch, head=head)
+        if git(repo.main, 'merge-base', '--is-ancestor', head, 'refs/heads/dev', check=False).returncode:
+            gate('dev_ancestor', 'fail', 'task HEAD is not an ancestor of dev; cleanup_pending')
+        else:
+            gate('dev_ancestor', 'pass')
+
+    unfinished = []
+    for marker in ('MERGE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD', 'rebase-merge', 'rebase-apply', 'BISECT_START'):
+        marker_abs = Path(git(target, 'rev-parse', '--path-format=absolute', '--git-path', marker))
+        if marker_abs.exists():
+            unfinished.append(marker)
+    if unfinished:
+        gate('git_operation', 'fail', 'unfinished Git operation; cleanup_pending', markers=unfinished)
+    else:
+        gate('git_operation', 'pass')
+
+    porcelain = git(target, 'status', '--porcelain=v1', '--untracked-files=all', '--ignored=matching')
+    if porcelain:
+        gate('worktree_clean', 'fail', 'tracked, untracked or ignored files remain',
+             sample=porcelain.splitlines()[:20])
+    else:
+        gate('worktree_clean', 'pass')
+
+    with repo.db(True) as db:
+        tasks = [dict(r) for r in db.execute('SELECT id,state FROM tasks WHERE worktree=?', (str(target),)).fetchall()]
+        busy = []
+        for task in tasks:
+            row = db.execute(
+                "SELECT id,state FROM runs WHERE task=? AND state IN ('running','recovery_required','queued')",
+                (task['id'],)).fetchone()
+            if row:
+                busy.append(dict(row))
+    if any(t['state'] == 'active' for t in tasks):
+        gate('registry_task', 'fail', 'active registered task owns this worktree', tasks=tasks)
+    else:
+        gate('registry_task', 'pass', tasks=tasks)
+    if busy:
+        gate('registry_runs', 'fail', 'live or unresolved operation', runs=busy)
+    else:
+        gate('registry_runs', 'pass')
+
+    key = digest(str(target))
+    lock_path = repo.root / 'os' / f'worktree-{key}.lock'
+    lifecycle = 'unknown'
+    if lock_path.exists():
+        try:
+            with lock_path.open('a+') as handle:
+                try:
+                    fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                    fcntl.flock(handle, fcntl.LOCK_UN)
+                    lifecycle = 'free'
+                except BlockingIOError:
+                    lifecycle = 'held'
+        except OSError as exc:
+            lifecycle = f'error:{exc}'
+    else:
+        lifecycle = 'absent'
+    gate('lifecycle_lock', 'fail' if lifecycle == 'held' else ('pass' if lifecycle in ('free', 'absent') else 'unknown'),
+         lifecycle)
+
+    process = cleanup_process_gate(target)
+    gate('process_cwd', 'pass' if process['status'] == 'clear' else (
+        'fail' if process['status'] == 'blocked' else 'unknown'), process.get('reason'), **process)
+
+    allowed = all(g['status'] == 'pass' for g in gates)
+    emit({
+        'status': 'allowed' if allowed else 'blocked',
+        'path': str(target),
+        'cleanup_allowed': allowed,
+        'gates': gates,
+    })
+    return 0
+
+
 def parser():
     p=argparse.ArgumentParser(description=__doc__);p.add_argument('--repo',default='.')
     sub=p.add_subparsers(dest='action',required=True)
@@ -974,6 +1346,16 @@ def parser():
     d=sub.add_parser('docs').add_subparsers(dest='doc_action',required=True)
     snap=d.add_parser('snapshot');snap.add_argument('--file',required=True);snap.add_argument('--record',required=True)
     apply=d.add_parser('apply');apply.add_argument('--task',required=True);apply.add_argument('--request',required=True)
+    proj=d.add_parser('project');proj.add_argument('--task',required=True);proj.add_argument('--file',default='.doc/ops/CURRENT.md')
+    proj.add_argument('--mutex',default='');proj.add_argument('--status');proj.add_argument('--request-id')
+    q=sub.add_parser('queue').add_subparsers(dest='queue_action',required=True)
+    qe=q.add_parser('enqueue');qe.add_argument('--task',required=True);qe.add_argument('--expected-dev',required=True)
+    qe.add_argument('--head');qe.add_argument('--priority',type=int,default=0);qe.add_argument('--request-id')
+    ql=q.add_parser('list');ql.add_argument('--all',action='store_true')
+    qc=q.add_parser('cancel');qc.add_argument('--task',required=True);qc.add_argument('--id',required=True);qc.add_argument('--reason',required=True)
+    qs=q.add_parser('schedule');qs.add_argument('--drain',action='store_true')
+    cu=sub.add_parser('cleanup').add_subparsers(dest='cleanup_action',required=True)
+    cs=cu.add_parser('status');cs.add_argument('--path');cs.add_argument('--branch')
     rec=sub.add_parser('recover');rec.add_argument('--run',required=True);rec.add_argument('--reason',required=True)
     op=sub.add_parser('operation');op.add_argument('run')
     for action in ('freeze','unfreeze'):
@@ -982,6 +1364,11 @@ def parser():
     hooks.add_parser('install');hc=hooks.add_parser('check');hc.add_argument('--merge',action='store_true');hr=hooks.add_parser('ref');hr.add_argument('state')
     v=sub.add_parser('validate');v.add_argument('--task',required=True);v.add_argument('command',nargs=argparse.REMAINDER)
     i=sub.add_parser('integrate');i.add_argument('--task',required=True);i.add_argument('--expected-dev',required=True);i.add_argument('--request-id')
+    q=sub.add_parser('queue').add_subparsers(dest='queue_action',required=True)
+    qe=q.add_parser('enqueue');qe.add_argument('--task',required=True);qe.add_argument('--expected-dev',default='dev');qe.add_argument('--head-sha');qe.add_argument('--priority',type=int,default=0);qe.add_argument('--request-id')
+    ql=q.add_parser('list');ql.add_argument('--all',action='store_true')
+    qc=q.add_parser('cancel');qc.add_argument('--id',required=True);qc.add_argument('--reason',required=True)
+    q.add_parser('schedule')
     return p
 
 
@@ -990,11 +1377,21 @@ def main():
     if args.action=='init':initialize(repo)
     elif args.action in ('status','doctor'):
         identity=repo.identity()
+        try:
+            ensure_schema(repo)
+        except Denied:
+            pass
         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')")]
+            try:
+                queue=[dict(r) for r in db.execute(
+                    "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'])
-        emit({'status':'ok','lock_root':str(repo.root),'main':str(repo.main),'version':VERSION,'epoch':identity['epoch'],'tasks':tasks,'runs':runs})
+        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':
         if args.task_action=='register':
             if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}',args.id):deny('INVALID_TASK','Invalid task id')
@@ -1031,7 +1428,19 @@ def main():
     elif args.action=='worktree-check':worktree_check(repo,args.path)
     elif args.action=='docs':
         if args.doc_action=='snapshot':docs_snapshot(repo,args)
+        elif args.doc_action=='project':return docs_project(repo,args)
         else:return docs_apply(repo,args)
+    elif args.action=='queue':
+        if args.queue_action=='enqueue':return queue_enqueue(repo,args)
+        if args.queue_action=='list':return queue_list(repo,args)
+        if args.queue_action=='cancel':return queue_cancel(repo,args)
+        return queue_schedule(repo,args)
+    elif args.action=='cleanup':
+        if not args.path and not args.branch:
+            deny('INVALID_REQUEST', 'cleanup status requires --path or --branch')
+        if args.path and args.branch:
+            deny('INVALID_REQUEST', 'Provide only one of --path or --branch')
+        return cleanup_status(repo,args)
     elif args.action=='recover':recover(repo,args)
     elif args.action=='operation':
         with repo.db() as db:row=db.execute('SELECT * FROM runs WHERE id=?',(args.run,)).fetchone()
@@ -1056,6 +1465,11 @@ def main():
         else:hook_ref(repo,args.state)
     elif args.action=='validate':return validate(repo,args)
     elif args.action=='integrate':return integrate(repo,args)
+    elif args.action=='queue':
+        if args.queue_action=='enqueue':return queue_enqueue(repo,args)
+        if args.queue_action=='list':return queue_list(repo,args)
+        if args.queue_action=='cancel':return queue_cancel(repo,args)
+        if args.queue_action=='schedule':return queue_schedule(repo,args)
     return 0
 
 

scripts/agentctl_automation.py

@@ -0,0 +1,413 @@
+#!/usr/bin/env python3
+"""W4 governance automation: durable queue, doc projection, deploy ops.
+
+Process liveness remains in agentctl.py (Linux /proc today; macOS sibling).
+This module is intentionally importable for unit tests without supervising children.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+import uuid
+from pathlib import Path
+
+# Job / operation terminal and active states.
+JOB_PENDING = 'pending'
+JOB_CLAIMED = 'claimed'
+JOB_RUNNING = 'running'
+JOB_COMPLETED = 'completed'
+JOB_FAILED = 'failed'
+JOB_CANCELLED = 'cancelled'
+JOB_ACTIVE = (JOB_PENDING, JOB_CLAIMED, JOB_RUNNING)
+JOB_TERMINAL = (JOB_COMPLETED, JOB_FAILED, JOB_CANCELLED)
+
+DEPLOY_PHASES = (
+    'planned', 'deploying', 'accepting', 'passed', 'failed',
+    'recovery_required', 'recovered',
+)
+
+# Managed shared docs plus projection-capable documents.
+DEFAULT_DOC_REGISTRY = {
+    '.doc/ops/CURRENT.md': {
+        'kind': 'shared_cas',
+        'header': '| owner | task |',
+        'rules': {'own_row_only': True, 'slot_protected': True},
+    },
+    '.doc/status/FEATURE-MATRIX.md': {
+        'kind': 'shared_cas',
+        'header': '| 模块 |',
+        'rules': {'evidence_for_done': True},
+    },
+    '.doc/ops/projections/cleanup-visibility.md': {
+        'kind': 'projection',
+        'header': '| path | branch | reason | cleanup_owner |',
+        'rules': {'generated': True},
+    },
+}
+
+SCHEMA_SQL = '''
+CREATE TABLE IF NOT EXISTS jobs(
+  generation INTEGER PRIMARY KEY AUTOINCREMENT,
+  id TEXT UNIQUE NOT NULL,
+  task TEXT NOT NULL,
+  kind TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  resources TEXT NOT NULL,
+  state TEXT NOT NULL,
+  priority INTEGER NOT NULL DEFAULT 100,
+  created REAL NOT NULL,
+  claimed_at REAL,
+  claimed_by TEXT,
+  run_id TEXT,
+  finished_at REAL,
+  result TEXT,
+  fingerprint TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, priority, generation);
+CREATE TABLE IF NOT EXISTS operations(
+  id TEXT PRIMARY KEY,
+  task TEXT NOT NULL,
+  kind TEXT NOT NULL,
+  phase TEXT NOT NULL,
+  env TEXT NOT NULL,
+  expected_sha TEXT,
+  runtime_sha TEXT,
+  receipt TEXT NOT NULL,
+  created REAL NOT NULL,
+  updated REAL NOT NULL
+);
+CREATE INDEX IF NOT EXISTS operations_task ON operations(task, updated);
+CREATE TABLE IF NOT EXISTS doc_registry(
+  path TEXT PRIMARY KEY,
+  kind TEXT NOT NULL,
+  header TEXT,
+  rules TEXT NOT NULL DEFAULT '{}'
+);
+'''
+
+
+def ensure_schema(db):
+    """Idempotent schema evolution; keeps protocol VERSION stable."""
+    db.executescript(SCHEMA_SQL)
+    for path, meta in DEFAULT_DOC_REGISTRY.items():
+        db.execute(
+            'INSERT OR IGNORE INTO doc_registry(path,kind,header,rules) VALUES(?,?,?,?)',
+            (path, meta['kind'], meta.get('header'), json.dumps(meta.get('rules') or {})),
+        )
+
+
+def process_platform_ready():
+    """True when the host can supply process start tokens for leases.
+
+    Linux uses /proc. Darwin needs the agentctl-macos process layer; until that
+    lands, lease-backed dispatch stays deferred rather than TypeError.
+    """
+    if os.environ.get('AGENTCTL_FORCE_PROC') == '1':
+        return True
+    if os.environ.get('AGENTCTL_FORCE_PROC') == '0':
+        return False
+    return Path('/proc/self/stat').is_file()
+
+
+def job_fingerprint(kind, payload, resources):
+    blob = json.dumps({'kind': kind, 'payload': payload, 'resources': sorted(resources)},
+                      sort_keys=True, ensure_ascii=False)
+    return blob
+
+
+def enqueue_job(db, task_id, kind, payload, resources, job_id=None, priority=100):
+    if kind not in ('command', 'validate', 'integrate', 'deploy', 'docs_apply', 'cleanup_report'):
+        raise ValueError(f'unsupported job kind: {kind}')
+    if not isinstance(payload, dict):
+        raise ValueError('payload must be an object')
+    if not isinstance(resources, (list, tuple)) or not resources:
+        raise ValueError('resources must be a nonempty list')
+    job_id = job_id or str(uuid.uuid4())
+    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', job_id):
+        raise ValueError('invalid job id')
+    fingerprint = __import__('hashlib').sha256(
+        job_fingerprint(kind, payload, list(resources)).encode()).hexdigest()
+    existing = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
+    if existing:
+        if existing['fingerprint'] != fingerprint or existing['task'] != task_id:
+            raise ValueError('job id reused with different fingerprint')
+        return dict(existing), True
+    now = time.time()
+    db.execute(
+        '''INSERT INTO jobs(id,task,kind,payload,resources,state,priority,created,fingerprint)
+           VALUES(?,?,?,?,?,?,?,?,?)''',
+        (job_id, task_id, kind, json.dumps(payload, ensure_ascii=False, sort_keys=True),
+         json.dumps(list(resources)), JOB_PENDING, int(priority), now, fingerprint))
+    row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
+    return dict(row), False
+
+
+def list_jobs(db, state=None, limit=50):
+    if state:
+        rows = db.execute(
+            'SELECT * FROM jobs WHERE state=? ORDER BY priority ASC, generation ASC LIMIT ?',
+            (state, limit)).fetchall()
+    else:
+        rows = db.execute(
+            'SELECT * FROM jobs ORDER BY CASE state '
+            "WHEN 'pending' THEN 0 WHEN 'claimed' THEN 1 WHEN 'running' THEN 2 "
+            "WHEN 'failed' THEN 3 ELSE 4 END, priority ASC, generation ASC LIMIT ?",
+            (limit,)).fetchall()
+    return [dict(r) for r in rows]
+
+
+def cancel_job(db, job_id, reason='cancelled'):
+    row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
+    if not row:
+        raise KeyError('unknown job')
+    if row['state'] in JOB_TERMINAL:
+        return dict(row), True
+    if row['state'] == JOB_RUNNING:
+        raise RuntimeError('cannot cancel a running job; recover the run first')
+    db.execute(
+        "UPDATE jobs SET state=?,finished_at=?,result=? WHERE id=?",
+        (JOB_CANCELLED, time.time(), json.dumps({'reason': reason}), job_id))
+    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()), False
+
+
+def claim_next_job(db, worker_id, kinds=None):
+    """Atomically claim the highest-priority pending job. Returns None if empty."""
+    clauses = ["state=?", JOB_PENDING]
+    sql = 'SELECT * FROM jobs WHERE state=? '
+    args = [JOB_PENDING]
+    if kinds:
+        placeholders = ','.join('?' for _ in kinds)
+        sql += f'AND kind IN ({placeholders}) '
+        args.extend(kinds)
+    sql += 'ORDER BY priority ASC, generation ASC LIMIT 1'
+    row = db.execute(sql, args).fetchone()
+    if not row:
+        return None
+    db.execute(
+        "UPDATE jobs SET state=?,claimed_at=?,claimed_by=? WHERE id=? AND state=?",
+        (JOB_CLAIMED, time.time(), worker_id, row['id'], JOB_PENDING))
+    claimed = db.execute('SELECT * FROM jobs WHERE id=?', (row['id'],)).fetchone()
+    if not claimed or claimed['state'] != JOB_CLAIMED or claimed['claimed_by'] != worker_id:
+        return None
+    return dict(claimed)
+
+
+def mark_job(db, job_id, state, result=None, run_id=None):
+    if state not in (JOB_RUNNING, JOB_COMPLETED, JOB_FAILED, JOB_PENDING, JOB_CLAIMED):
+        raise ValueError('invalid job state')
+    finished = time.time() if state in (JOB_COMPLETED, JOB_FAILED) else None
+    db.execute(
+        'UPDATE jobs SET state=?,result=?,run_id=COALESCE(?,run_id),finished_at=COALESCE(?,finished_at) WHERE id=?',
+        (state, json.dumps(result) if result is not None else None, run_id, finished, job_id))
+    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())
+
+
+def plan_deploy(db, task_id, env, expected_sha, operation_id=None, extra=None):
+    if env not in ('dev', 'test', 'main'):
+        raise ValueError('env must be dev|test|main')
+    if not re.fullmatch(r'[0-9a-f]{7,40}', expected_sha):
+        raise ValueError('expected_sha must be a git object name')
+    operation_id = operation_id or str(uuid.uuid4())
+    now = time.time()
+    receipt = {
+        'task': task_id,
+        'env': env,
+        'expected_sha': expected_sha,
+        'events': [{'at': now, 'phase': 'planned'}],
+        **(extra or {}),
+    }
+    existing = db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()
+    if existing:
+        if existing['task'] != task_id or existing['expected_sha'] != expected_sha:
+            raise ValueError('operation id reused with different identity')
+        return dict(existing), True
+    db.execute(
+        '''INSERT INTO operations(id,task,kind,phase,env,expected_sha,runtime_sha,receipt,created,updated)
+           VALUES(?,?,?,?,?,?,?,?,?,?)''',
+        (operation_id, task_id, 'deploy', 'planned', env, expected_sha, None,
+         json.dumps(receipt, ensure_ascii=False), now, now))
+    return dict(db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()), False
+
+
+def advance_deploy(db, operation_id, phase, runtime_sha=None, detail=None):
+    if phase not in DEPLOY_PHASES:
+        raise ValueError(f'invalid phase: {phase}')
+    row = db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()
+    if not row:
+        raise KeyError('unknown operation')
+    allowed = {
+        'planned': {'deploying', 'failed', 'recovery_required'},
+        'deploying': {'accepting', 'failed', 'recovery_required'},
+        'accepting': {'passed', 'failed', 'recovery_required'},
+        'failed': {'recovery_required', 'recovered'},
+        'recovery_required': {'recovered', 'deploying'},
+        'passed': set(),
+        'recovered': {'deploying'},
+    }
+    if phase not in allowed.get(row['phase'], set()) and phase != row['phase']:
+        raise RuntimeError(f'illegal transition {row["phase"]} -> {phase}')
+    receipt = json.loads(row['receipt'])
+    now = time.time()
+    receipt.setdefault('events', []).append({'at': now, 'phase': phase, 'detail': detail})
+    if runtime_sha:
+        receipt['runtime_sha'] = runtime_sha
+    db.execute(
+        'UPDATE operations SET phase=?,runtime_sha=COALESCE(?,runtime_sha),receipt=?,updated=? WHERE id=?',
+        (phase, runtime_sha, json.dumps(receipt, ensure_ascii=False), now, operation_id))
+    return dict(db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone())
+
+
+def list_operations(db, task=None, limit=50):
+    if task:
+        rows = db.execute(
+            'SELECT * FROM operations WHERE task=? ORDER BY updated DESC LIMIT ?',
+            (task, limit)).fetchall()
+    else:
+        rows = db.execute(
+            'SELECT * FROM operations ORDER BY updated DESC LIMIT ?', (limit,)).fetchall()
+    return [dict(r) for r in rows]
+
+
+def list_doc_registry(db):
+    ensure_schema(db)
+    return [dict(r) for r in db.execute('SELECT * FROM doc_registry ORDER BY path').fetchall()]
+
+
+def managed_doc_paths(db):
+    return {row['path'] for row in list_doc_registry(db)}
+
+
+def project_cleanup_markdown(entries):
+    """Render cleanup visibility projection from structured entries."""
+    lines = [
+        '# Cleanup visibility projection',
+        '',
+        '> Generated by `agentctl docs project --kind cleanup`. Not a lock; machine `.locks` is authoritative.',
+        '',
+        '| path | branch | reason | cleanup_owner |',
+        '| --- | --- | --- | --- |',
+    ]
+    if not entries:
+        lines.append('| — | — | none_pending | — |')
+    else:
+        for item in entries:
+            path = str(item.get('path') or '').replace('|', '\\|')
+            branch = str(item.get('branch') or '').replace('|', '\\|')
+            reason = str(item.get('reason') or '').replace('|', '\\|')
+            owner = str(item.get('cleanup_owner') or 'independent-maintainer').replace('|', '\\|')
+            lines.append(f'| {path} | {branch} | {reason} | {owner} |')
+    lines.append('')
+    return '\n'.join(lines)
+
+
+def assess_cleanup_entry(main, entry, control_fn=None, process_cwd_fn=None):
+    """Return a visibility dict; never deletes. Failures become reason strings."""
+    from pathlib import Path as P
+    branch_ref = entry.get('branch') or ''
+    branch = branch_ref.replace('refs/heads/', '') if branch_ref.startswith('refs/heads/') else ''
+    path = P(entry.get('worktree') or '')
+    result = {
+        'path': str(path),
+        'branch': branch or None,
+        'eligible': False,
+        'reason': 'unknown',
+        'checks': {},
+    }
+    if not branch or not path:
+        result['reason'] = 'missing_branch_or_path'
+        return result
+    if 'locked' in entry:
+        result['reason'] = 'git_worktree_locked'
+        return result
+    if 'prunable' in entry:
+        result['reason'] = 'git_worktree_prunable'
+        return result
+    if not path.is_dir():
+        result['reason'] = 'path_missing'
+        return result
+
+    def git_c(*args):
+        import subprocess
+        return subprocess.run(['git', '-C', str(path if args[0] != '--main' else main), *([a for a in args if a != '--main'])],
+                              capture_output=True, text=True)
+
+    # Use explicit git -C for main vs path.
+    import subprocess
+
+    def run(cwd, *args):
+        return subprocess.run(['git', '-C', str(cwd), *args], capture_output=True, text=True)
+
+    actual = run(path, 'symbolic-ref', '--quiet', 'HEAD')
+    if actual.returncode or actual.stdout.strip() != f'refs/heads/{branch}':
+        result['reason'] = 'branch_identity_mismatch'
+        result['checks']['head_ref'] = actual.stdout.strip()
+        return result
+    head = run(path, 'rev-parse', 'HEAD').stdout.strip()
+    ancestor = run(main, 'merge-base', '--is-ancestor', head, 'refs/heads/dev')
+    if ancestor.returncode:
+        result['reason'] = 'head_not_ancestor_of_dev'
+        result['checks']['head'] = head
+        return result
+    result['checks']['head'] = head
+    for marker in ('MERGE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD', 'rebase-merge', 'rebase-apply', 'BISECT_START'):
+        marker_path = run(path, 'rev-parse', '--path-format=absolute', '--git-path', marker).stdout.strip()
+        if marker_path and Path(marker_path).exists():
+            result['reason'] = f'unfinished_git_op:{marker}'
+            return result
+    status = run(path, 'status', '--porcelain=v1', '--untracked-files=all', '--ignored=matching')
+    if status.stdout.strip():
+        result['reason'] = 'dirty_or_ignored_retained'
+        result['checks']['status_bytes'] = len(status.stdout)
+        return result
+    if control_fn is not None:
+        try:
+            control_fn(path)
+            result['checks']['registry'] = 'ok'
+        except Exception as exc:  # noqa: BLE001 — visibility aggregates any refusal
+            result['reason'] = f'registry:{exc}'
+            return result
+    if process_cwd_fn is not None:
+        try:
+            process_cwd_fn(path)
+            result['checks']['process_cwd'] = 'ok'
+        except Exception as exc:  # noqa: BLE001
+            result['reason'] = f'process_cwd:{exc}'
+            return result
+    elif not process_platform_ready():
+        result['reason'] = 'process_cwd:platform_deferred_macos_or_no_proc'
+        result['checks']['process_cwd'] = 'deferred'
+        return result
+    result['eligible'] = True
+    result['reason'] = 'eligible'
+    return result
+
+
+def build_command_for_job(kind, payload):
+    """Map a durable job to an argv list relative to agentctl semantics."""
+    if kind == 'command':
+        command = payload.get('command')
+        if not isinstance(command, list) or not command or any(not isinstance(c, str) for c in command):
+            raise ValueError('command job requires string argv list')
+        return command
+    if kind == 'validate':
+        command = payload.get('command')
+        if not isinstance(command, list) or not command:
+            raise ValueError('validate job requires command argv')
+        return ['validate-proxy', *command]  # resolved by dispatcher
+    if kind == 'integrate':
+        expected = payload.get('expected_dev')
+        if not isinstance(expected, str) or not expected:
+            raise ValueError('integrate job requires expected_dev')
+        return ['integrate-proxy', expected]
+    if kind == 'deploy':
+        return ['deploy-proxy', payload.get('operation_id') or '']
+    if kind == 'docs_apply':
+        request = payload.get('request_path')
+        if not isinstance(request, str):
+            raise ValueError('docs_apply requires request_path')
+        return ['docs-proxy', request]
+    if kind == 'cleanup_report':
+        return ['cleanup-report-proxy']
+    raise ValueError(f'unsupported kind {kind}')

scripts/agentctl_w4.py

@@ -0,0 +1,399 @@
+#!/usr/bin/env python3
+"""W4 automation helpers for agentctl: durable jobs, doc projection, deploy loop.
+
+Linux-first. Process liveness reuses the host agentctl proc/alive/boot helpers
+passed in as callbacks so a future macOS process layer can plug in without
+rewriting queue semantics.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import sqlite3
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+SCHEMA_SQL = '''
+CREATE TABLE IF NOT EXISTS jobs(
+  id TEXT PRIMARY KEY,
+  task TEXT NOT NULL,
+  kind TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  state TEXT NOT NULL,
+  priority INTEGER NOT NULL DEFAULT 100,
+  not_before REAL NOT NULL,
+  created REAL NOT NULL,
+  updated REAL NOT NULL,
+  lease_run TEXT,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_error TEXT,
+  result TEXT
+);
+CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, not_before, priority, created);
+CREATE TABLE IF NOT EXISTS deployments(
+  id TEXT PRIMARY KEY,
+  task TEXT NOT NULL,
+  env TEXT NOT NULL,
+  expected_commit TEXT NOT NULL,
+  manifest TEXT,
+  phase TEXT NOT NULL,
+  dry_run INTEGER NOT NULL DEFAULT 0,
+  created REAL NOT NULL,
+  updated REAL NOT NULL,
+  receipt TEXT,
+  last_error TEXT
+);
+CREATE INDEX IF NOT EXISTS deployments_task ON deployments(task, phase);
+CREATE TABLE IF NOT EXISTS projections(
+  file TEXT NOT NULL,
+  record_id TEXT NOT NULL,
+  task TEXT NOT NULL,
+  source TEXT NOT NULL,
+  content_hash TEXT NOT NULL,
+  updated REAL NOT NULL,
+  PRIMARY KEY(file, record_id)
+);
+'''
+
+JOB_KINDS = frozenset({
+    'validate', 'integrate', 'with', 'deploy_prepare', 'deploy_run',
+    'deploy_accept', 'docs_project', 'cleanup_inspect', 'noop',
+})
+JOB_ACTIVE = ('pending', 'leased', 'running')
+JOB_TERMINAL = ('completed', 'failed', 'cancelled')
+DEPLOY_PHASES = (
+    'prepared', 'deploying', 'accepting', 'passed', 'failed', 'recovery_required',
+)
+# Beyond CURRENT/MATRIX: any Markdown table under .doc/ may be projected when
+# registered here. Layout changes still require docs:admin via docs apply.
+PROJECTABLE_DOCS = frozenset({
+    '.doc/ops/CURRENT.md',
+    '.doc/status/FEATURE-MATRIX.md',
+    '.doc/ops/CURRENT.md',  # duplicate ok in set
+})
+PROJECTABLE_PREFIXES = ('.doc/ops/', '.doc/status/', '.doc/remediation/')
+
+
+def ensure_schema(db: sqlite3.Connection) -> None:
+    db.executescript(SCHEMA_SQL)
+
+
+def _now() -> float:
+    return time.time()
+
+
+def queue_enqueue(db, *, task: str, kind: str, payload: dict, priority: int = 100,
+                  not_before: Optional[float] = None, job_id: Optional[str] = None) -> dict:
+    if kind not in JOB_KINDS:
+        raise ValueError(f'unsupported job kind: {kind}')
+    if not isinstance(payload, dict):
+        raise ValueError('payload must be an object')
+    jid = job_id or str(uuid.uuid4())
+    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', jid):
+        raise ValueError('invalid job id')
+    ts = _now()
+    when = float(not_before) if not_before is not None else ts
+    existing = db.execute('SELECT * FROM jobs WHERE id=?', (jid,)).fetchone()
+    if existing:
+        if (existing['task'] != task or existing['kind'] != kind
+                or existing['payload'] != json.dumps(payload, sort_keys=True)):
+            raise ValueError('job id reused with different payload')
+        return dict(existing)
+    db.execute(
+        '''INSERT INTO jobs(id,task,kind,payload,state,priority,not_before,created,updated)
+           VALUES(?,?,?,?,?,?,?,?,?)''',
+        (jid, task, kind, json.dumps(payload, sort_keys=True), 'pending',
+         int(priority), when, ts, ts))
+    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (jid,)).fetchone())
+
+
+def queue_list(db, *, task: Optional[str] = None, state: Optional[str] = None,
+               limit: int = 50) -> list:
+    clauses, args = [], []
+    if task:
+        clauses.append('task=?')
+        args.append(task)
+    if state:
+        clauses.append('state=?')
+        args.append(state)
+    where = (' WHERE ' + ' AND '.join(clauses)) if clauses else ''
+    rows = db.execute(
+        f'SELECT * FROM jobs{where} ORDER BY priority ASC, created ASC LIMIT ?',
+        (*args, max(1, min(int(limit), 500)))).fetchall()
+    return [dict(r) for r in rows]
+
+
+def queue_cancel(db, job_id: str, reason: str) -> dict:
+    row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
+    if not row:
+        raise KeyError(job_id)
+    if row['state'] in JOB_TERMINAL:
+        return dict(row)
+    if row['state'] == 'running':
+        raise RuntimeError('cannot cancel a running job; recover the lease first')
+    db.execute(
+        "UPDATE jobs SET state='cancelled',last_error=?,updated=? WHERE id=?",
+        (reason, _now(), job_id))
+    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())
+
+
+def claim_next_job(db, *, worker: str) -> Optional[dict]:
+    """Atomically lease the next due pending job. Caller must be in a write txn."""
+    ts = _now()
+    row = db.execute(
+        '''SELECT * FROM jobs WHERE state='pending' AND not_before<=?
+           ORDER BY priority ASC, created ASC LIMIT 1''', (ts,)).fetchone()
+    if not row:
+        return None
+    db.execute(
+        '''UPDATE jobs SET state='leased',lease_run=?,attempts=attempts+1,updated=?
+           WHERE id=? AND state='pending' ''',
+        (worker, ts, row['id']))
+    claimed = db.execute('SELECT * FROM jobs WHERE id=?', (row['id'],)).fetchone()
+    if not claimed or claimed['state'] != 'leased':
+        return None
+    return dict(claimed)
+
+
+def finish_job(db, job_id: str, *, ok: bool, result: Optional[dict] = None,
+               error: Optional[str] = None) -> dict:
+    state = 'completed' if ok else 'failed'
+    db.execute(
+        '''UPDATE jobs SET state=?,result=?,last_error=?,updated=? WHERE id=?''',
+        (state, json.dumps(result or {}, sort_keys=True), error, _now(), job_id))
+    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())
+
+
+def mark_job_running(db, job_id: str) -> None:
+    db.execute("UPDATE jobs SET state='running',updated=? WHERE id=?", (_now(), job_id))
+
+
+def projectable_file(path: str) -> bool:
+    if path in PROJECTABLE_DOCS or path in {'.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'}:
+        return True
+    return any(path.startswith(prefix) for prefix in PROJECTABLE_PREFIXES) and path.endswith('.md')
+
+
+def build_projection_row(*, task_id: str, owner: str, branch: str, worktree: str,
+                         mutex: str, status: str, extra: Optional[dict] = None) -> str:
+    """Canonical CURRENT-style row used for machine→doc projection."""
+    cells = [owner, task_id, branch, worktree, mutex, status]
+    if extra:
+        for key in sorted(extra):
+            cells.append(str(extra[key]))
+    return '| ' + ' | '.join(cells) + ' |'
+
+
+def record_projection(db, *, file: str, record_id: str, task: str, source: str,
+                      content_hash: str) -> None:
+    db.execute(
+        '''INSERT INTO projections(file,record_id,task,source,content_hash,updated)
+           VALUES(?,?,?,?,?,?)
+           ON CONFLICT(file,record_id) DO UPDATE SET
+             task=excluded.task, source=excluded.source,
+             content_hash=excluded.content_hash, updated=excluded.updated''',
+        (file, record_id, task, source, content_hash, _now()))
+
+
+def deployment_create(db, *, deploy_id: str, task: str, env: str, expected_commit: str,
+                      manifest: Optional[str], dry_run: bool) -> dict:
+    if env not in ('dev', 'test', 'main'):
+        raise ValueError('env must be dev|test|main')
+    if not re.fullmatch(r'[0-9a-f]{7,40}', expected_commit):
+        raise ValueError('expected_commit must be a hex SHA')
+    ts = _now()
+    existing = db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone()
+    if existing:
+        if (existing['task'] != task or existing['env'] != env
+                or existing['expected_commit'] != expected_commit):
+            raise ValueError('deploy id reused with different identity')
+        return dict(existing)
+    db.execute(
+        '''INSERT INTO deployments(id,task,env,expected_commit,manifest,phase,dry_run,created,updated)
+           VALUES(?,?,?,?,?,?,?,?,?)''',
+        (deploy_id, task, env, expected_commit, manifest, 'prepared',
+         1 if dry_run else 0, ts, ts))
+    return dict(db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone())
+
+
+def deployment_set_phase(db, deploy_id: str, phase: str, *, receipt: Optional[dict] = None,
+                         error: Optional[str] = None) -> dict:
+    if phase not in DEPLOY_PHASES:
+        raise ValueError(f'invalid deploy phase: {phase}')
+    db.execute(
+        '''UPDATE deployments SET phase=?,receipt=?,last_error=?,updated=? WHERE id=?''',
+        (phase, json.dumps(receipt or {}, sort_keys=True) if receipt is not None else None,
+         error, _now(), deploy_id))
+    return dict(db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone())
+
+
+def deployment_get(db, deploy_id: str) -> Optional[dict]:
+    row = db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone()
+    return dict(row) if row else None
+
+
+def dry_run_deploy_steps(deploy: dict) -> dict:
+    """Closed-loop steps without touching shared docker. Used by tests and design."""
+    return {
+        'env': deploy['env'],
+        'expected_commit': deploy['expected_commit'],
+        'manifest': deploy.get('manifest'),
+        'steps': [
+            {'name': 'guard_env_lease', 'status': 'ok'},
+            {'name': 'verify_main_clean', 'status': 'ok'},
+            {'name': 'apply_or_rebuild', 'status': 'skipped_dry_run'},
+            {'name': 'health_ready', 'status': 'skipped_dry_run'},
+            {'name': 'pi_image_identity', 'status': 'skipped_dry_run'},
+        ],
+        'dry_run': True,
+    }
+
+
+def cleanup_visibility_report(
+    *,
+    path: Path,
+    branch: str,
+    head: str,
+    ancestor_of_dev: bool,
+    dirty: bool,
+    unfinished_git: list,
+    registry: dict,
+    process: dict,
+) -> dict:
+    """Structured cleanup gate report. Never deletes; visibility only."""
+    blockers = []
+    if not ancestor_of_dev:
+        blockers.append({'code': 'NOT_MERGED', 'message': 'HEAD is not an ancestor of dev'})
+    if dirty:
+        blockers.append({'code': 'DIRTY_TREE', 'message': 'tracked/untracked/ignored files remain'})
+    for marker in unfinished_git:
+        blockers.append({'code': 'GIT_IN_PROGRESS', 'message': f'unfinished Git operation: {marker}'})
+    if registry.get('active_task'):
+        blockers.append({'code': 'TASK_ACTIVE', 'message': 'registered task still active'})
+    if registry.get('live_run'):
+        blockers.append({'code': 'LOCK_BUSY', 'message': 'live or unresolved operation',
+                         'run_id': registry.get('live_run')})
+    status = process.get('status')
+    if status == 'busy':
+        blockers.append({'code': 'PROCESS_BUSY', 'message': 'process still uses worktree cwd',
+                         'pids': process.get('pids', [])})
+    elif status == 'unreadable':
+        blockers.append({'code': 'PROCESS_UNREADABLE', 'message': 'cannot verify same-user cwd',
+                         'pids': process.get('unreadable', []),
+                         'platform_note': process.get('platform_note')})
+    elif status == 'unsupported':
+        blockers.append({'code': 'PROCESS_UNSUPPORTED',
+                         'message': process.get('message', 'process inspection unsupported'),
+                         'platform_note': process.get('platform_note')})
+    return {
+        'status': 'ready' if not blockers else 'cleanup_pending',
+        'path': str(path),
+        'branch': branch,
+        'head': head,
+        'gates': {
+            'ancestor_of_dev': ancestor_of_dev,
+            'dirty': dirty,
+            'unfinished_git': unfinished_git,
+            'registry': registry,
+            'process': process,
+        },
+        'blockers': blockers,
+    }
+
+
+def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
+    """Fail-closed cwd scan using /proc. macOS should use lsof via sibling worktree."""
+    import sys
+    if not sys.platform.startswith('linux'):
+        return {
+            'status': 'unsupported',
+            'message': f'/proc cwd scan unavailable on {sys.platform}',
+            'platform_note': 'Use agentctl-macos lsof enumerator; treat unknown as cleanup_pending',
+            'pids': [],
+            'unreadable': [],
+        }
+    uid = os.getuid() if uid is None else uid
+    busy, unreadable = [], []
+    try:
+        entries = list(Path('/proc').iterdir())
+    except OSError:
+        return {
+            'status': 'unreadable',
+            'message': 'cannot list /proc',
+            'platform_note': 'fail-closed',
+            'pids': [],
+            'unreadable': ['*'],
+        }
+    for proc in entries:
+        if not proc.name.isdigit():
+            continue
+        try:
+            if proc.stat().st_uid != uid:
+                continue
+            cwd = Path(os.readlink(proc / 'cwd'))
+            if cwd == path or path in cwd.parents:
+                busy.append(proc.name)
+        except (FileNotFoundError, ProcessLookupError):
+            continue
+        except PermissionError:
+            try:
+                if (proc / 'comm').read_text().strip() == 'systemd':
+                    continue
+            except (OSError, UnicodeError):
+                pass
+            unreadable.append(proc.name)
+    if busy:
+        return {'status': 'busy', 'pids': busy, 'unreadable': unreadable}
+    if unreadable:
+        return {'status': 'unreadable', 'pids': [], 'unreadable': unreadable,
+                'platform_note': 'fail-closed; do not treat as exited'}
+    return {'status': 'clear', 'pids': [], 'unreadable': []}
+
+
+def pi_source_bind_paths() -> tuple:
+    """Host paths that must not be bind-mounted when PI image delivery is enforced."""
+    return (
+        './deploy/pi-agent/runner.mjs',
+        './deploy/pi-agent/pi-args.mjs',
+        './deploy/pi-agent/result-v1.mjs',
+        './deploy/pi-agent/extensions',
+        'deploy/pi-agent/runner.mjs',
+        'deploy/pi-agent/pi-args.mjs',
+        'deploy/pi-agent/result-v1.mjs',
+        'deploy/pi-agent/extensions',
+    )
+
+
+def inspect_compose_for_pi_binds(compose_text: str) -> dict:
+    """Static check: flag source bind mounts under pi-agent service."""
+    forbidden = pi_source_bind_paths()
+    hits = []
+    in_pi = False
+    for line in compose_text.splitlines():
+        stripped = line.strip()
+        if re.match(r'^pi-agent\s*:', stripped):
+            in_pi = True
+            continue
+        if in_pi and re.match(r'^[A-Za-z0-9_.-]+:\s*$', stripped):
+            in_pi = False
+        if in_pi:
+            for item in forbidden:
+                if item in stripped and (':/app/' in stripped or stripped.startswith('-')):
+                    hits.append(stripped)
+                    break
+    return {
+        'ok': not hits,
+        'source_binds': hits,
+        'recommendation': (
+            'Apply docker-compose.dev.pi-image.yml overlay so runner source ships in the image'
+            if hits else 'no pi-agent source binds detected'
+        ),
+    }
+
+
+# Type aliases for agentctl callbacks (avoid circular imports at runtime).
+EmitFn = Callable[[Any], None]
+DenyFn = Callable[..., None]