#!/usr/bin/env python3 """Governance console helpers: claim_new, integrate_prepare, pi_packet, serve. Shared by CLI and HTTP (same-process). Raises agentctl.Denied on refusal. Returns payload dicts; CLI wrappers call emit(). """ from __future__ import annotations import argparse import contextlib import json import mimetypes import os import re import tempfile import threading import time import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Optional from urllib.parse import parse_qs, urlparse import agentctl as ac import agentctl_visibility as visibility try: import pi_subagents as pi except Exception: # noqa: BLE001 — optional for environments without full PI deps pi = None CURRENT_REL = '.doc/ops/CURRENT.md' DEFAULT_HOST = '127.0.0.1' DEFAULT_PORT = 7432 DEFAULT_STATIC = 'scripts/governance-console' FORBIDDEN_BODY_KEYS = frozenset({'command', 'shell', 'argv'}) HTTP_STATUS = { 'INVALID_REQUEST': 400, 'CAPABILITY_REQUIRED': 403, 'PATH_OUT_OF_SCOPE': 403, 'BASE_CHANGED': 409, 'WORKTREE_MISMATCH': 409, 'LOCK_BUSY': 409, 'UNMANAGED_DOCUMENT_CHANGE': 409, 'DIRTY_TREE': 409, 'FROZEN': 409, 'EVIDENCE_REQUIRED': 422, 'REQUEST_ID_REUSED': 409, 'RECOVERY_REQUIRED': 500, } def _caps(task) -> list: raw = task['capabilities'] return json.loads(raw) if isinstance(raw, str) else list(raw or []) def _rel_worktree(path: Path, main: Path) -> str: try: return str(path.resolve().relative_to(main.resolve())) except ValueError: text = str(path) if '/.worktrees/' in text.replace('\\', '/'): return '.worktrees/' + text.replace('\\', '/').split('/.worktrees/', 1)[1] return text def claim_new(repo: ac.Repo, args) -> dict: """Create or replace a CURRENT claim row via single-row docs apply.""" task = ac.task_row(repo, args.task) if repo.current != Path(task['worktree']) or repo.current == repo.main: ac.deny('WORKTREE_MISMATCH', 'claim new must run inside the task worktree') if 'docs:publish' not in _caps(task) and 'docs:admin' not in _caps(task): ac.deny('CAPABILITY_REQUIRED', 'claim new requires docs:publish') owner = getattr(args, 'owner', None) or task.get('owner') or 'Cursor' branch = getattr(args, 'branch', None) or task.get('branch') or f"feature/{args.task}" worktree = getattr(args, 'worktree', None) or _rel_worktree(Path(task['worktree']), repo.main) mutex = getattr(args, 'mutex', None) or '-' status = getattr(args, 'status', None) or 'in_progress' if status not in ac.PROJECTION_STATUS_VOCAB: ac.deny('INVALID_REQUEST', 'status must be a projection vocab value') content = ac.safe_path(repo.current, CURRENT_REL).read_text(encoding='utf-8') existing = ac.doc_rows(content, args.task) expected = ac.digest(existing[0][1]) if existing else None replacement = ( f"| {owner} | {args.task} | {branch} | {worktree} | {mutex} | {status} |" ) if existing and existing[0][1].strip() == replacement.strip(): return { 'status': 'idempotent', 'record_id': args.task, 'request_id': getattr(args, 'request_id', None), 'row': replacement, } if existing and not getattr(args, 'replace', False): ac.deny('INVALID_REQUEST', 'row exists; pass --replace') request_id = getattr(args, 'request_id', None) or f"{args.task}-claim-new-{uuid.uuid4().hex[:8]}" request = { 'request_id': request_id, 'changes': [{ 'file': CURRENT_REL, 'record_id': args.task, 'expected_sha256': expected, 'replacement': replacement, }], } req_path = Path(tempfile.mkstemp(prefix='agentctl-claim-new-', suffix='.json')[1]) try: req_path.write_text(json.dumps(request, ensure_ascii=False), encoding='utf-8') captured = [] with _capture_emit(captured): ac.docs_apply(repo, argparse.Namespace(task=args.task, request=str(req_path))) applied = next((c for c in captured if isinstance(c, dict)), {}) status = applied.get('status') if status == 'replayed': out_status = 'replayed' else: out_status = 'replaced' if existing else 'created' return { 'status': out_status, 'record_id': args.task, 'request_id': request_id, 'run_id': applied.get('run_id'), 'row': replacement, } finally: with contextlib.suppress(OSError): req_path.unlink() def integrate_prepare(repo: ac.Repo, args) -> dict: """Build integrate candidate worktree and --no-ff merge task branch.""" task = ac.task_row(repo, args.task) expected = getattr(args, 'expected_dev', None) if not expected or not re.fullmatch(r'[0-9a-f]{40}|[0-9a-f]{64}', expected): ac.deny('INVALID_REQUEST', 'expected-dev must be a full commit hash') tip = ac.git(repo.main, 'rev-parse', 'refs/heads/dev') if tip != expected: ac.deny('BASE_CHANGED', 'refs/heads/dev tip differs from expected-dev', tip=tip, expected_dev=expected) integrate_id = getattr(args, 'integrate_id', None) or f"integrate-{args.task}" if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}', integrate_id): ac.deny('INVALID_REQUEST', 'Invalid integrate-id') branch = f'feature/{integrate_id}' integ_path = repo.main / '.worktrees' / integrate_id task_branch = task['branch'] if not task_branch: ac.deny('INVALID_REQUEST', 'Source task has no branch') if integ_path.exists(): integ_repo = ac.Repo(str(integ_path)) head = ac.git(integ_path, 'rev-parse', 'HEAD') first = ac.git(integ_path, 'rev-parse', 'HEAD^1', check=False) if first.returncode == 0 and first.stdout.strip() == expected: # Already prepared on this base. return { 'status': 'prepared', 'integrate_task': integrate_id, 'worktree': str(integ_path), 'head': head, 'expected_dev': expected, 'first_parent': expected, 'idempotent': True, } porcelain = ac.git(integ_path, 'status', '--porcelain', '--untracked-files=all') if porcelain: ac.deny('INVALID_REQUEST', 'Existing integrate worktree is dirty', path=str(integ_path)) if head != expected: # Different base: refuse rather than reset. ac.deny('BASE_CHANGED', 'Integrate worktree exists for a different base; remove it first', path=str(integ_path), head=head) # Clean checkout parked on expected-dev (e.g. after abort): continue to merge. else: integ_path.parent.mkdir(exist_ok=True) ac.git(repo.main, 'worktree', 'add', '-b', branch, str(integ_path), expected) # Register integrate task (must run as checkout identity). _register_integrate_task(integ_path, integrate_id, owner=task.get('owner') or 'Cursor') integ_repo = ac.Repo(str(integ_path)) integ_task = ac.task_row(integ_repo, integrate_id) request_id = getattr(args, 'request_id', None) or f"{integrate_id}-prepare-{uuid.uuid4().hex[:8]}" msg = f"merge: candidate {args.task} into dev" # New request_id when recovering after a failed fence-less attempt. with ac.Lease(integ_repo, integ_task, ['path:.', 'git:index', 'git:dev'], request_id, ac.digest(msg + expected + task_branch), kind='integrate_prepare') as lease: if lease.replay: head = ac.git(integ_path, 'rev-parse', 'HEAD') return { 'status': 'replayed', 'run_id': lease.request_id, 'integrate_task': integrate_id, 'worktree': str(integ_path), 'head': head, 'expected_dev': expected, } # Merge commit must run with AGENTCTL_* fence (pre-commit hook), same as integrate(). env = os.environ.copy() env.update( AGENTCTL_RUN_ID=lease.request_id, AGENTCTL_TASK_ID=integrate_id, AGENTCTL_GENERATION=str(lease.row['generation']), AGENTCTL_EPOCH=integ_repo.identity()['epoch'], AGENTCTL_FENCE=str(lease.row.get('fence') or 0), ) code, orphan = ac.supervise( integ_repo, lease, ['git', 'merge', '--no-ff', '-m', msg, task_branch], integ_path, env, ) if code or orphan: with contextlib.suppress(Exception): ac.git(integ_path, 'merge', '--abort', check=False) lease.finish(code or 1, True) ac.deny('GIT_ERROR', 'merge --no-ff failed', exit_code=code, orphan=bool(orphan)) head = ac.git(integ_path, 'rev-parse', 'HEAD') first_parent = ac.git(integ_path, 'rev-parse', 'HEAD^1') if first_parent != expected: lease.finish(1, True) ac.deny('BASE_CHANGED', 'first parent is not expected-dev', first_parent=first_parent, expected_dev=expected) lease.finish() return { 'status': 'prepared', 'integrate_task': integrate_id, 'worktree': str(integ_path), 'head': head, 'expected_dev': expected, 'first_parent': first_parent, 'run_id': request_id, 'idempotent': False, } def _register_integrate_task(integ_path: Path, integrate_id: str, owner: str): """Register integrate task from that checkout (subprocess avoids cwd gymnastics).""" import subprocess import sys agentctl = Path(__file__).resolve() cmd = [ sys.executable, '-B', str(agentctl.parent / 'agentctl.py'), 'task', 'register', '--id', integrate_id, '--owner', owner, '--scope', '.', '--capability', 'git:dev', '--capability', 'docs:publish', ] proc = subprocess.run(cmd, cwd=str(integ_path), capture_output=True, text=True) if proc.returncode != 0: # Already registered with same identity is ok if output says registered/exists matching. err = (proc.stderr or proc.stdout or '').strip() try: payload = json.loads(err.splitlines()[-1]) except Exception: payload = {} if payload.get('code') == 'TASK_EXISTS': return if payload.get('status') == 'registered': return ac.deny('INVALID_REQUEST', 'Failed to register integrate task', detail=err[:500]) def pi_packet(repo: ac.Repo, args) -> dict: """Build PI dispatch packet JSON; optional write under report assets.""" task = ac.task_row(repo, args.task) wt = Path(task['worktree']) role = getattr(args, 'role', None) or getattr(args, 'role_id', None) known = getattr(pi, 'KNOWN_ROLES', None) if pi else None if not role: ac.deny('INVALID_REQUEST', 'role is required') if known is not None and role not in known: ac.deny('UNKNOWN_ROLE', 'Unknown role_id', role_id=role) goal = getattr(args, 'goal', None) auth = getattr(args, 'authorization_ref', None) if not goal or not auth: ac.deny('INVALID_REQUEST', 'goal and authorization-ref are required') raw_sha = getattr(args, 'source_sha', None) if not raw_sha: raw_sha = ac.git(wt, 'rev-parse', 'HEAD') if pi is not None: try: source_sha = pi.freeze_source_sha(wt, raw_sha) except Exception as exc: # ExecutorError code = getattr(exc, 'code', 'INVALID_SOURCE_SHA') ac.deny(code, getattr(exc, 'message', str(exc)), **getattr(exc, 'details', {})) else: if not re.fullmatch(r'[0-9a-f]{40}|[0-9a-f]{64}', raw_sha): ac.deny('INVALID_SOURCE_SHA', 'source_sha must be a full commit hash') source_sha = raw_sha request_id = getattr(args, 'request_id', None) or str(uuid.uuid4()) rid_re = getattr(pi, 'REQUEST_ID_RE', re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$')) if not rid_re.fullmatch(request_id): ac.deny('INVALID_REQUEST', 'Invalid request_id') input_files = list(getattr(args, 'input_file', None) or getattr(args, 'input_files', None) or []) approved = [] for item in (getattr(args, 'approved_command', None) or []): if isinstance(item, list): approved.append(item) elif isinstance(item, str): approved.append(item.split()) else: ac.deny('INVALID_REQUEST', 'approved-command must be string or list') timeout = int(getattr(args, 'timeout_seconds', None) or 600) packet = { 'task_id': args.task, 'request_id': request_id, 'role_id': role, 'goal': goal, 'authorization_ref': auth, 'source_sha': source_sha, 'input_files': input_files, 'timeout_seconds': timeout, 'approved_commands': approved, } wrote = None if getattr(args, 'write', False): rel_root = getattr(pi, 'REPORT_ROOT_REL', '.doc/test-reports/assets/pi-subagents') rel = f"{rel_root}/{args.task}/{request_id}/packet.json" # Write under task worktree with lease. if repo.current != wt: ac.deny('WORKTREE_MISMATCH', 'pi packet --write must run in the task worktree') dest = ac.safe_path(repo.current, rel) dest.parent.mkdir(parents=True, exist_ok=True) body = json.dumps(packet, ensure_ascii=False, indent=2) + '\n' with ac.Lease(repo, task, [f'path:{rel}'], f"{args.task}-pi-packet-{request_id}", ac.digest(body), kind='pi_packet') as lease: if lease.replay: return {'status': 'replayed', 'run_id': lease.request_id, 'packet': packet, 'path': rel} ac.atomic_write(dest, body) lease.finish() wrote = rel return {'status': 'ok', 'packet': packet, 'path': wrote} @contextlib.contextmanager def _capture_emit(bucket: list): original = ac.emit def _cap(value, stream=None): bucket.append(value) ac.emit = _cap try: yield bucket finally: ac.emit = original def denied_http_status(code: str) -> int: if code in HTTP_STATUS: return HTTP_STATUS[code] if code.startswith('UNKNOWN_'): return 404 return 500 def build_brief_text(repo: ac.Repo) -> str: status = ac.collect_status_payload(repo) try: docs = ac.collect_docs_drift_payload(repo) except Exception as exc: # noqa: BLE001 docs = {'status': 'error', 'error': str(exc)} try: cleanup = ac.collect_cleanup_payload(repo) except Exception as exc: # noqa: BLE001 cleanup = {'status': 'error', 'error': str(exc), 'worktrees': []} try: cat = ac.collect_catalog_drift_payload(repo) except Exception as exc: # noqa: BLE001 cat = {'status': 'error', 'error': str(exc)} try: daemon = ac.collect_daemon_payload(repo) except Exception as exc: # noqa: BLE001 daemon = {'status': 'error', 'error': str(exc)} return visibility.build_brief( status=status, docs_drift=docs, cleanup=cleanup, catalog_drift=cat, daemon=daemon, ) def _queue_payload(repo: ac.Repo, all_items: bool = False) -> dict: states = None if all_items 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() return {'status': 'ok', 'items': [ac.queue_row(r) for r in rows]} def _claims_payload(repo: ac.Repo, all_items: bool = False) -> dict: with repo.db() as db: try: if all_items: rows = [dict(r) for r in db.execute( 'SELECT * FROM claims ORDER BY updated DESC').fetchall()] else: rows = [dict(r) for r in db.execute( "SELECT * FROM claims WHERE state='held' ORDER BY updated DESC").fetchall()] except Exception: rows = [] return {'status': 'ok', 'claims': rows} def serve(repo: ac.Repo, args) -> int: host = getattr(args, 'host', None) or DEFAULT_HOST port = int(getattr(args, 'port', None) or DEFAULT_PORT) if host not in ('127.0.0.1', 'localhost', '::1'): ac.deny('INVALID_REQUEST', 'serve host must be loopback (127.0.0.1)') static_rel = getattr(args, 'static', None) or DEFAULT_STATIC static_root = (repo.main / static_rel).resolve() if not str(static_root).startswith(str(repo.main.resolve())): ac.deny('PATH_OUT_OF_SCOPE', 'static root must stay under main worktree') # Prefer main tip; if not yet integrated, allow the same relative path from # the current checkout when it remains under the main worktree tree. if not (static_root / 'index.html').is_file(): alt = (repo.current / static_rel).resolve() if str(alt).startswith(str(repo.main.resolve())) and (alt / 'index.html').is_file(): static_root = alt class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *a): pass def _send(self, code: int, body: Any, content_type='application/json; charset=utf-8'): if isinstance(body, (dict, list)): data = json.dumps(body, ensure_ascii=False, sort_keys=True).encode('utf-8') elif isinstance(body, str): data = body.encode('utf-8') else: data = body self.send_response(code) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(len(data))) self.send_header('Cache-Control', 'no-store') self.end_headers() self.wfile.write(data) def _read_json(self) -> dict: length = int(self.headers.get('Content-Length') or 0) if length > 1_000_000: ac.deny('INVALID_REQUEST', 'request body too large') raw = self.rfile.read(length) if length else b'{}' try: data = json.loads(raw.decode('utf-8') or '{}') except Exception: ac.deny('INVALID_REQUEST', 'JSON body required') if not isinstance(data, dict): ac.deny('INVALID_REQUEST', 'JSON object required') if FORBIDDEN_BODY_KEYS & set(data): ac.deny('INVALID_REQUEST', 'command/shell/argv fields are forbidden') return data def do_GET(self): # noqa: N802 try: parsed = urlparse(self.path) path = parsed.path qs = parse_qs(parsed.query) fmt = (qs.get('format') or ['json'])[0] all_items = (qs.get('all') or ['0'])[0] in ('1', 'true', 'yes') if path == '/api/status': payload = ac.collect_status_payload(repo) if fmt == 'table': return self._send(200, visibility.render_status_table(payload), 'text/plain; charset=utf-8') return self._send(200, payload) if path == '/api/brief': brief_fmt = (qs.get('format') or ['markdown'])[0] text = build_brief_text(repo) if brief_fmt == 'json': return self._send(200, {'status': 'ok', 'brief': text}) return self._send(200, text, 'text/markdown; charset=utf-8') if path == '/api/queue': payload = _queue_payload(repo, all_items) if fmt == 'table': return self._send(200, visibility.render_queue_table(payload), 'text/plain; charset=utf-8') return self._send(200, payload) if path == '/api/claims': payload = _claims_payload(repo, all_items) if fmt == 'table': return self._send(200, visibility.render_claim_table(payload), 'text/plain; charset=utf-8') return self._send(200, payload) if path == '/api/cleanup': payload = ac.collect_cleanup_payload(repo) if fmt == 'table': return self._send(200, visibility.render_cleanup_table(payload), 'text/plain; charset=utf-8') return self._send(200, payload) if path == '/api/docs/drift': return self._send(200, ac.collect_docs_drift_payload(repo)) if path == '/api/catalog/drift': return self._send(200, ac.collect_catalog_drift_payload(repo)) if path == '/api/daemon': return self._send(200, ac.collect_daemon_payload(repo)) if path in ('/', '/index.html'): return self._static('index.html') if path.startswith('/static/'): return self._static(path[len('/static/'):]) return self._send(404, {'status': 'denied', 'code': 'NOT_FOUND', 'message': path}) except ac.Denied as exc: return self._send(denied_http_status(exc.code), { 'status': 'denied', 'code': exc.code, 'message': exc.message, **exc.details, }) except Exception as exc: # noqa: BLE001 return self._send(500, { 'status': 'denied', 'code': 'RECOVERY_REQUIRED', 'message': str(exc)[:300], }) def do_POST(self): # noqa: N802 try: path = urlparse(self.path).path body = self._read_json() if path == '/api/claim/new': ns = argparse.Namespace( task=body.get('task'), owner=body.get('owner'), mutex=body.get('mutex', '-'), branch=body.get('branch'), worktree=body.get('worktree'), status=body.get('status', 'in_progress'), replace=bool(body.get('replace')), request_id=body.get('request_id'), ) if not ns.task: ac.deny('INVALID_REQUEST', 'task required') # claim_new must run in task worktree task = ac.task_row(repo, ns.task) task_repo = ac.Repo(task['worktree']) return self._send(200, claim_new(task_repo, ns)) if path == '/api/claim/archive': ns = argparse.Namespace( task=body.get('task'), retain_days=int(body.get('retain_days', 7)), include_undated=bool(body.get('include_undated')), apply=True, ) if not ns.task: ac.deny('INVALID_REQUEST', 'task required') task = ac.task_row(repo, ns.task) task_repo = ac.Repo(task['worktree']) captured = [] with _capture_emit(captured): ac.claim_archive(task_repo, ns) return self._send(200, captured[-1] if captured else {'status': 'ok'}) if path == '/api/integrate/prepare': ns = argparse.Namespace( task=body.get('task'), expected_dev=body.get('expected_dev'), integrate_id=body.get('integrate_id'), request_id=body.get('request_id'), ) if not ns.task or not ns.expected_dev: ac.deny('INVALID_REQUEST', 'task and expected_dev required') return self._send(200, integrate_prepare(repo, ns)) if path == '/api/pi/packet': ns = argparse.Namespace( task=body.get('task'), role=body.get('role') or body.get('role_id'), goal=body.get('goal'), authorization_ref=body.get('authorization_ref'), source_sha=body.get('source_sha'), timeout_seconds=body.get('timeout_seconds', 600), input_file=body.get('input_files') or [], approved_command=body.get('approved_commands') or [], request_id=body.get('request_id'), write=bool(body.get('write')), ) if not ns.task: ac.deny('INVALID_REQUEST', 'task required') task = ac.task_row(repo, ns.task) task_repo = ac.Repo(task['worktree']) if ns.write else repo return self._send(200, pi_packet(task_repo, ns)) return self._send(404, {'status': 'denied', 'code': 'NOT_FOUND', 'message': path}) except ac.Denied as exc: return self._send(denied_http_status(exc.code), { 'status': 'denied', 'code': exc.code, 'message': exc.message, **exc.details, }) except Exception as exc: # noqa: BLE001 return self._send(500, { 'status': 'denied', 'code': 'RECOVERY_REQUIRED', 'message': str(exc)[:300], }) def _static(self, name: str): name = name.lstrip('/') if '..' in name or name.startswith('/'): return self._send(400, {'status': 'denied', 'code': 'INVALID_REQUEST', 'message': 'bad path'}) target = (static_root / name).resolve() if not str(target).startswith(str(static_root)) or not target.is_file(): return self._send(404, {'status': 'denied', 'code': 'NOT_FOUND', 'message': name}) ctype = mimetypes.guess_type(str(target))[0] or 'application/octet-stream' return self._send(200, target.read_bytes(), ctype) server = ThreadingHTTPServer((host, port), Handler) ac.emit({ 'status': 'serving', 'host': host, 'port': port, 'static': str(static_root), 'url': f'http://{host}:{port}/', }) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() return 0