From f01970564a205e74dfecd8d2313e46ae17b17ad2 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Tue, 22 Sep 2026 21:51:17 +0800 Subject: [PATCH] feat(ops): advance governance automation queue and projection increments --- .../2026-09-22-governance-automation.md | 69 +++ scripts/agent_worktree.py | 155 ++++++- scripts/agentctl.py | 420 +++++++++++++++++- scripts/agentctl_automation.py | 413 +++++++++++++++++ scripts/agentctl_w4.py | 399 +++++++++++++++++ 5 files changed, 1448 insertions(+), 8 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-22-governance-automation.md create mode 100644 scripts/agentctl_automation.py create mode 100644 scripts/agentctl_w4.py diff --git a/.doc/ops/short-term/2026-09-22-governance-automation.md b/.doc/ops/short-term/2026-09-22-governance-automation.md new file mode 100644 index 00000000..742c141b --- /dev/null +++ b/.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 + +（实现后回填） diff --git a/scripts/agent_worktree.py b/scripts/agent_worktree.py index 17eec591..1d874441 100644 --- a/scripts/agent_worktree.py +++ b/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": diff --git a/scripts/agentctl.py b/scripts/agentctl.py index 888d90c0..8fa48e24 100755 --- a/scripts/agentctl.py +++ b/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 diff --git a/scripts/agentctl_automation.py b/scripts/agentctl_automation.py new file mode 100644 index 00000000..55501ba1 --- /dev/null +++ b/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}') diff --git a/scripts/agentctl_w4.py b/scripts/agentctl_w4.py new file mode 100644 index 00000000..9607a59b --- /dev/null +++ b/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]