feat(agentctl): add governance console P1 serve and action APIs

Implement loopback HTTP console with claim new, integrate prepare, and pi packet sharing the same helpers as CLI; ship SPA and MCP brief/claim_archive wrappers.

scripts/agentctl.py

@@ -28,6 +28,7 @@ os.environ.setdefault('PYTHONDONTWRITEBYTECODE', '1')
 import agentctl_w4 as w4  # noqa: E402
 import agentctl_catalog as catalog  # noqa: E402
 import agentctl_visibility as visibility  # noqa: E402
+import agentctl_console as console  # noqa: E402
 
 VERSION = 1
 POLICY_REL = 'configs/agent-governance.json'
@@ -2972,6 +2973,10 @@ def parser():
             sp.add_argument('--format',choices=['json','table'],default='json')
     brief=sub.add_parser('brief', help='One-page governance briefing (Markdown)')
     brief.add_argument('--format',choices=['markdown','json'],default='markdown')
+    srv=sub.add_parser('serve', help='Loopback governance HTTP console')
+    srv.add_argument('--host', default='127.0.0.1')
+    srv.add_argument('--port', type=int, default=7432)
+    srv.add_argument('--static', default='scripts/governance-console')
     task=sub.add_parser('task').add_subparsers(dest='task_action',required=True)
     reg=task.add_parser('register');reg.add_argument('--id',required=True);reg.add_argument('--owner',required=True);reg.add_argument('--scope',action='append',required=True);reg.add_argument('--capability',action='append',default=[])
     reg.add_argument('--requirement',help='Catalog R-ID to bind at register')
@@ -3013,7 +3018,16 @@ 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')
     hooks.add_parser('codegraph-sync', help='Fail-open codegraph sync for post-merge/checkout')
     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')
+    i=sub.add_parser('integrate')
+    i.add_argument('--task')
+    i.add_argument('--expected-dev')
+    i.add_argument('--request-id')
+    i_sub=i.add_subparsers(dest='integrate_action')
+    ip=i_sub.add_parser('prepare', help='Build integrate candidate worktree + no-ff merge')
+    ip.add_argument('--task',required=True)
+    ip.add_argument('--expected-dev',required=True)
+    ip.add_argument('--integrate-id')
+    ip.add_argument('--request-id')
     dep=sub.add_parser('deploy').add_subparsers(dest='deploy_action',required=True)
     dr=dep.add_parser('run-loop')
     dr.add_argument('--task',required=True);dr.add_argument('--id',required=True)
@@ -3035,6 +3049,15 @@ def parser():
     carc.add_argument('--retain-days',type=int,default=7)
     carc.add_argument('--include-undated',action='store_true')
     carc.add_argument('--apply',action='store_true')
+    cnew=cl.add_parser('new', help='Create CURRENT claim row via docs apply')
+    cnew.add_argument('--task',required=True)
+    cnew.add_argument('--owner')
+    cnew.add_argument('--mutex',default='-')
+    cnew.add_argument('--branch')
+    cnew.add_argument('--worktree')
+    cnew.add_argument('--status',default='in_progress')
+    cnew.add_argument('--replace',action='store_true')
+    cnew.add_argument('--request-id')
     cc=cl.add_parser('cancel');cc.add_argument('--task',required=True);cc.add_argument('--id',required=True);cc.add_argument('--reason',required=True)
     qtn=sub.add_parser('quarantine').add_subparsers(dest='quarantine_action',required=True)
     qtn.add_parser('sweep')
@@ -3049,6 +3072,18 @@ def parser():
     drun.add_argument('--audit-every',type=int,default=30,dest='audit_every',
                       help='Every N cycles run cleanup report + docs drift (0=off)')
     dm.add_parser('status')
+    pi_root=sub.add_parser('pi').add_subparsers(dest='pi_action', required=True)
+    pp=pi_root.add_parser('packet', help='Generate PI dispatch packet JSON')
+    pp.add_argument('--task',required=True)
+    pp.add_argument('--role',required=True)
+    pp.add_argument('--goal',required=True)
+    pp.add_argument('--authorization-ref',required=True)
+    pp.add_argument('--source-sha')
+    pp.add_argument('--timeout-seconds',type=int,default=600)
+    pp.add_argument('--input-file',action='append',default=[])
+    pp.add_argument('--approved-command',action='append',default=[])
+    pp.add_argument('--request-id')
+    pp.add_argument('--write',action='store_true')
     cat = sub.add_parser('catalog').add_subparsers(dest='catalog_action', required=True)
     cadd = cat.add_parser('add')
     cadd.add_argument('--id', required=True)
@@ -3087,7 +3122,7 @@ def main():
     write_actions = {
         'init', 'task', 'with', 'docs', 'validate', 'integrate', 'queue',
         'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup', 'claim',
-        'quarantine', 'daemon', 'catalog',
+        'quarantine', 'daemon', 'catalog', 'serve', 'pi',
     }
     # status/doctor remain readable without a policy file and never create one.
     if args.action in write_actions:
@@ -3123,6 +3158,11 @@ def main():
         emit_payload(payload, getattr(args, 'format', 'json'), visibility.render_status_table)
     elif args.action=='brief':
         return cmd_brief(repo, args)
+    elif args.action=='serve':
+        return console.serve(repo, args)
+    elif args.action=='pi':
+        if args.pi_action=='packet':
+            emit(console.pi_packet(repo, args)); return 0
     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')
@@ -3221,7 +3261,12 @@ def main():
         elif args.hook_action=='codegraph-sync':return hooks_codegraph_sync(repo)
         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=='integrate':
+        if getattr(args, 'integrate_action', None)=='prepare':
+            emit(console.integrate_prepare(repo, args)); return 0
+        if not getattr(args, 'task', None) or not getattr(args, 'expected_dev', None):
+            deny('INVALID_REQUEST', 'integrate requires --task and --expected-dev (or: integrate prepare)')
+        return integrate(repo, args)
     elif args.action=='deploy':return deploy_cmd(repo,args)
     elif args.action=='claim':
         if args.claim_action=='acquire':return claim_acquire(repo,args)
@@ -3229,6 +3274,8 @@ def main():
         if args.claim_action=='status':return claim_status(repo,args)
         if args.claim_action=='list':return claim_list(repo,args)
         if args.claim_action=='archive':return claim_archive(repo,args)
+        if args.claim_action=='new':
+            emit(console.claim_new(repo,args)); return 0
         if args.claim_action=='cancel':return claim_cancel(repo,args)
     elif args.action=='quarantine':
         if args.quarantine_action=='sweep':return quarantine_sweep(repo)

scripts/agentctl_console.py

@@ -0,0 +1,591 @@
+#!/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 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))
+        # 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)
+    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"
+    msg = f"merge: candidate {args.task} into dev"
+    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,
+            }
+        result = ac.git(integ_path, 'merge', '--no-ff', '-m', msg, task_branch, check=False)
+        if result.returncode != 0:
+            with contextlib.suppress(Exception):
+                ac.git(integ_path, 'merge', '--abort', check=False)
+            lease.finish(1, True)
+            ac.deny('GIT_ERROR', 'merge --no-ff failed', stderr=(result.stderr or '')[:500])
+        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

scripts/governance-console/index.html

@@ -0,0 +1,199 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+<meta charset="utf-8"/>
+<meta name="viewport" content="width=device-width, initial-scale=1"/>
+<title>agentctl console</title>
+<style>
+  :root { --bg:#0f1419; --panel:#1a2332; --text:#e7ecf3; --muted:#8b9bb4; --accent:#3d8bfd; --danger:#e35d6a; --ok:#3dd68c; --border:#2a3548; }
+  * { box-sizing: border-box; }
+  body { margin:0; font:14px/1.45 ui-sans-serif, system-ui, sans-serif; background:var(--bg); color:var(--text); }
+  header { display:flex; gap:1rem; align-items:center; padding:.75rem 1rem; border-bottom:1px solid var(--border); background:var(--panel); position:sticky; top:0; z-index:2; }
+  header h1 { font-size:1rem; margin:0; font-weight:600; letter-spacing:.02em; }
+  header .meta { color:var(--muted); flex:1; }
+  button { background:var(--accent); color:#fff; border:0; border-radius:6px; padding:.35rem .7rem; cursor:pointer; }
+  button.ghost { background:transparent; border:1px solid var(--border); color:var(--text); }
+  button.danger { background:var(--danger); }
+  main { display:grid; grid-template-columns:1fr 1fr; gap:1rem; padding:1rem; }
+  section { background:var(--panel); border:1px solid var(--border); border-radius:10px; padding:.75rem 1rem; min-height:220px; }
+  section h2 { margin:0 0 .5rem; font-size:.95rem; }
+  table { width:100%; border-collapse:collapse; font-size:12px; }
+  th, td { text-align:left; padding:.3rem .25rem; border-bottom:1px solid var(--border); vertical-align:top; }
+  th { color:var(--muted); font-weight:500; }
+  .empty { color:var(--muted); font-style:italic; }
+  .toast { position:fixed; right:1rem; bottom:1rem; max-width:28rem; padding:.75rem 1rem; border-radius:8px; background:#222; border:1px solid var(--border); display:none; white-space:pre-wrap; }
+  .toast.err { border-color:var(--danger); color:#ffd0d4; }
+  .toast.ok { border-color:var(--ok); }
+  dialog { border:1px solid var(--border); border-radius:10px; background:var(--panel); color:var(--text); padding:1rem; max-width:36rem; width:90%; }
+  dialog textarea, dialog input, dialog select { width:100%; background:var(--bg); color:var(--text); border:1px solid var(--border); border-radius:6px; padding:.4rem; margin:.25rem 0 .6rem; font:inherit; }
+  #brief { grid-column:1 / -1; }
+  pre.brief { margin:0; white-space:pre-wrap; font-size:12px; color:var(--muted); max-height:12rem; overflow:auto; }
+  @media (max-width:900px) { main { grid-template-columns:1fr; } }
+</style>
+</head>
+<body>
+<header>
+  <h1>agentctl console</h1>
+  <div class="meta" id="meta">loading…</div>
+  <label><input type="checkbox" id="auto" checked/> 10s</label>
+  <button class="ghost" id="refresh">刷新</button>
+  <button id="btnClaim">认领…</button>
+  <button id="btnPrepare">Prepare…</button>
+  <button class="ghost" id="btnArchive" title="dry-run by default in CLI; SPA 仅提示">归档指引</button>
+</header>
+<main>
+  <section>
+    <h2>1. 任务 / 租约</h2>
+    <div id="tasks"></div>
+  </section>
+  <section>
+    <h2>2. 队列与集成</h2>
+    <div id="queue"></div>
+    <p class="empty">Enqueue 请用 CLI：<code>agentctl queue enqueue --task … --expected-dev … --head …</code></p>
+  </section>
+  <section>
+    <h2>3. 文档漂移</h2>
+    <div id="drift"></div>
+  </section>
+  <section>
+    <h2>4. 清理积压</h2>
+    <div id="cleanup"></div>
+  </section>
+  <section id="brief">
+    <h2>brief</h2>
+    <pre class="brief" id="briefBody"></pre>
+  </section>
+</main>
+<div class="toast" id="toast"></div>
+
+<dialog id="dlgClaim">
+  <form method="dialog" id="formClaim">
+    <h3>claim new</h3>
+    <label>task <input name="task" required/></label>
+    <label>owner <input name="owner" value="Cursor" required/></label>
+    <label>mutex <input name="mutex" value="-"/></label>
+    <label><input type="checkbox" name="replace"/> replace</label>
+    <p class="empty">须在已注册任务 worktree 上下文由 serve 代其 Repo 执行。</p>
+    <menu>
+      <button value="cancel" class="ghost">取消</button>
+      <button id="confirmClaim" value="default">确认</button>
+    </menu>
+  </form>
+</dialog>
+
+<dialog id="dlgPrepare">
+  <form method="dialog" id="formPrepare">
+    <h3>integrate prepare</h3>
+    <label>task <input name="task" required/></label>
+    <label>expected_dev <input name="expected_dev" required placeholder="full sha"/></label>
+    <label>integrate_id <input name="integrate_id" placeholder="optional"/></label>
+    <menu>
+      <button value="cancel" class="ghost">取消</button>
+      <button id="confirmPrepare" value="default">确认</button>
+    </menu>
+  </form>
+</dialog>
+
+<script>
+const $ = (s) => document.querySelector(s);
+const toast = (msg, ok) => {
+  const el = $('#toast');
+  el.textContent = msg;
+  el.className = 'toast ' + (ok ? 'ok' : 'err');
+  el.style.display = 'block';
+  setTimeout(() => el.style.display = 'none', 6000);
+};
+async function api(path, opts) {
+  const r = await fetch(path, opts);
+  const ct = r.headers.get('content-type') || '';
+  const body = ct.includes('json') ? await r.json() : await r.text();
+  if (!r.ok) {
+    const code = body && body.code ? body.code : r.status;
+    const msg = body && body.message ? body.message : String(body);
+    throw new Error(code + ': ' + msg);
+  }
+  return body;
+}
+function table(headers, rows) {
+  if (!rows.length) return '<p class="empty">(empty)</p>';
+  let h = '<table><thead><tr>' + headers.map(x => '<th>'+x+'</th>').join('') + '</tr></thead><tbody>';
+  for (const row of rows) h += '<tr>' + row.map(c => '<td>'+esc(c)+'</td>').join('') + '</tr>';
+  return h + '</tbody></table>';
+}
+function esc(s) {
+  return String(s ?? '-').replace(/[&<>"']/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
+}
+async function refresh() {
+  try {
+    const [status, queue, claims, cleanup, docs, cat, daemon, brief] = await Promise.all([
+      api('/api/status'), api('/api/queue'), api('/api/claims'), api('/api/cleanup'),
+      api('/api/docs/drift'), api('/api/catalog/drift'), api('/api/daemon'),
+      api('/api/brief?format=json'),
+    ]);
+    const active = (status.tasks||[]).filter(t => t.state !== 'done' && t.state !== 'cancelled');
+    $('#meta').textContent = `daemon:${daemon.status||'?'} lock:${daemon.manager_lock&&daemon.manager_lock.held} queue:${(queue.items||[]).length} active:${active.length}`;
+    $('#tasks').innerHTML = table(['task','owner','state','worktree'], active.map(t => [t.id, t.owner, t.state, (t.worktree||'').split('/.worktrees/').pop()||t.worktree]))
+      + '<h3 style="font-size:12px;color:var(--muted)">runs</h3>'
+      + table(['run','task','state'], (status.runs||[]).map(r => [r.id, r.task, r.state]))
+      + '<h3 style="font-size:12px;color:var(--muted)">claims</h3>'
+      + table(['task','resource','state'], (claims.claims||[]).map(c => [c.task, c.resource, c.state]));
+    $('#queue').innerHTML = table(['task','state','pri','head'], (queue.items||[]).map(i => [i.task, i.state, i.priority, (i.head_sha||'').slice(0,12)]));
+    const regs = (docs.regressions||[]).concat(cat.regressions||[]);
+    $('#drift').innerHTML = table(['record','prev→cur','msg'], regs.map(r => [r.record_id, (r.previous||r.catalog||'')+'→'+(r.current||r.matrix||''), r.message||'']));
+    const trees = cleanup.worktrees||[];
+    $('#cleanup').innerHTML = table(['path','allowed','owner'], trees.map(t => {
+      const allowed = t.cleanup_allowed || t.status === 'allowed';
+      return [(t.path||'').split('/.worktrees/').pop()||t.path, allowed, t.cleanup_owner||'-'];
+    }));
+    $('#briefBody').textContent = brief.brief || '';
+  } catch (e) {
+    toast(String(e), false);
+  }
+}
+$('#refresh').onclick = refresh;
+let timer = setInterval(refresh, 10000);
+$('#auto').onchange = (e) => {
+  clearInterval(timer);
+  if (e.target.checked) timer = setInterval(refresh, 10000);
+};
+$('#btnClaim').onclick = () => $('#dlgClaim').showModal();
+$('#btnPrepare').onclick = () => $('#dlgPrepare').showModal();
+$('#btnArchive').onclick = () => toast(
+  '归档请用 CLI（需 docs:admin）：\nagentctl claim archive --task <id> [--apply]\n或 MCP claim_archive',
+  true,
+);
+$('#confirmClaim').onclick = async (ev) => {
+  ev.preventDefault();
+  const f = $('#formClaim');
+  const body = {
+    task: f.task.value.trim(),
+    owner: f.owner.value.trim(),
+    mutex: f.mutex.value.trim() || '-',
+    replace: f.replace.checked,
+  };
+  try {
+    const r = await api('/api/claim/new', { method:'POST', headers:{'Content-Type':'application/json'}, body: JSON.stringify(body) });
+    toast(JSON.stringify(r, null, 2), true);
+    $('#dlgClaim').close();
+    refresh();
+  } catch (e) { toast(String(e), false); }
+};
+$('#confirmPrepare').onclick = async (ev) => {
+  ev.preventDefault();
+  const f = $('#formPrepare');
+  const body = {
+    task: f.task.value.trim(),
+    expected_dev: f.expected_dev.value.trim(),
+    integrate_id: f.integrate_id.value.trim() || undefined,
+  };
+  try {
+    const r = await api('/api/integrate/prepare', { method:'POST', headers:{'Content-Type':'application/json'}, body: JSON.stringify(body) });
+    toast(JSON.stringify(r, null, 2), true);
+    $('#dlgPrepare').close();
+    refresh();
+  } catch (e) { toast(String(e), false); }
+};
+refresh();
+</script>
+</body>
+</html>

scripts/mcp/agentctl-mcp.mjs

@@ -262,6 +262,33 @@ const TOOLS = [
     description: 'Refresh catalog features from MATRIX; report done→non-done regressions',
     inputSchema: { type: 'object', properties: {}, additionalProperties: false },
   },
+  {
+    name: 'brief',
+    description: 'One-page governance briefing (markdown or json)',
+    inputSchema: {
+      type: 'object',
+      properties: {
+        format: { type: 'string', enum: ['markdown', 'json'] },
+      },
+      additionalProperties: false,
+    },
+  },
+  {
+    name: 'claim_archive',
+    description: 'Archive done CURRENT claim rows (default dry-run; set apply=true to write)',
+    inputSchema: {
+      type: 'object',
+      properties: {
+        task: { type: 'string' },
+        retain_days: { type: 'number' },
+        include_undated: { type: 'boolean' },
+        apply: { type: 'boolean' },
+        cwd: { type: 'string', description: 'task worktree absolute path' },
+      },
+      required: ['task'],
+      additionalProperties: false,
+    },
+  },
 ];
 
 function callTool(name, args = {}) {
@@ -396,6 +423,20 @@ function callTool(name, args = {}) {
       const r = runAgentctl(['catalog', 'drift']);
       return r.ok ? textResult(r.stdout.trim()) : errResult(r.stderr || r.stdout || r.error);
     }
+    case 'brief': {
+      const argv = ['brief'];
+      if (args.format) argv.push('--format', args.format);
+      const r = runAgentctl(argv);
+      return r.ok ? textResult(r.stdout.trim()) : errResult(r.stderr || r.stdout || r.error);
+    }
+    case 'claim_archive': {
+      const argv = ['claim', 'archive', '--task', args.task];
+      if (args.retain_days != null) argv.push('--retain-days', String(args.retain_days));
+      if (args.include_undated) argv.push('--include-undated');
+      if (args.apply) argv.push('--apply');
+      const r = runAgentctl(argv, { cwd: args.cwd || REPO });
+      return r.ok ? textResult(r.stdout.trim()) : errResult(r.stderr || r.stdout || r.error);
+    }
     default:
       return errResult(`unknown tool: ${name}`);
   }

scripts/tests/test_agent_worktree.py

@@ -24,7 +24,7 @@ class WorktreeTest(unittest.TestCase):
         self.root = Path(self.tmp.name) / "repo"
         self.root.mkdir()
         (self.root / "scripts").mkdir()
-        for name in ("agent-worktree.sh", "agent_worktree.py", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py"):
+        for name in ("agent-worktree.sh", "agent_worktree.py", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py", "agentctl_console.py"):
             shutil.copy2(SOURCE / name, self.root / "scripts" / name)
         (self.root / "configs").mkdir(parents=True, exist_ok=True)
         shutil.copy2(SOURCE.parent / "configs" / "agent-governance.json",

scripts/tests/test_agentctl.py

@@ -39,6 +39,7 @@ class AgentctlTests(unittest.TestCase):
         shutil.copy2(SOURCE.parent / "agentctl_w4.py", self.root / "scripts" / "agentctl_w4.py")
         shutil.copy2(SOURCE.parent / "agentctl_catalog.py", self.root / "scripts" / "agentctl_catalog.py")
         shutil.copy2(SOURCE.parent / "agentctl_visibility.py", self.root / "scripts" / "agentctl_visibility.py")
+        shutil.copy2(SOURCE.parent / "agentctl_console.py", self.root / "scripts" / "agentctl_console.py")
         install_governance_policy(self.root)
         (self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
         for directory in ("src/a", "src/b", ".doc/ops", ".doc/status"):
@@ -1555,6 +1556,7 @@ class ProcessPlatformTests(unittest.TestCase):
         shutil.copy2(SOURCE.parent / "agentctl_w4.py", root / "scripts" / "agentctl_w4.py")
         shutil.copy2(SOURCE.parent / "agentctl_catalog.py", root / "scripts" / "agentctl_catalog.py")
         shutil.copy2(SOURCE.parent / "agentctl_visibility.py", root / "scripts" / "agentctl_visibility.py")
+        shutil.copy2(SOURCE.parent / "agentctl_console.py", root / "scripts" / "agentctl_console.py")
         install_governance_policy(root)
         (root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
         (root / ".doc/ops").mkdir(parents=True)

scripts/tests/test_agentctl_catalog.py

@@ -36,6 +36,9 @@ class CatalogTests(unittest.TestCase):
         vis = SOURCE.parent / "agentctl_visibility.py"
         if vis.exists():
             shutil.copy2(vis, self.root / "scripts" / "agentctl_visibility.py")
+        cons = SOURCE.parent / "agentctl_console.py"
+        if cons.exists():
+            shutil.copy2(cons, self.root / "scripts" / "agentctl_console.py")
         install_governance_policy(self.root)
         (self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
         for directory in ("src/a", "src/b", ".doc/ops", ".doc/status"):

scripts/tests/test_agentctl_serve.py

@@ -0,0 +1,366 @@
+#!/usr/bin/env python3
+"""Black-box tests for agentctl serve / claim new / integrate prepare / pi packet."""
+
+from __future__ import annotations
+
+import json
+import pathlib
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+import urllib.error
+import urllib.request
+
+SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl.py"
+POLICY_SOURCE = SOURCE.parents[1] / "configs" / "agent-governance.json"
+CONSOLE_STATIC = SOURCE.parent / "governance-console"
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class ConsoleServeTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(prefix="agentctl-serve-")
+        self.addCleanup(self.temp.cleanup)
+        self.root = pathlib.Path(self.temp.name) / "repo"
+        self.root.mkdir()
+        self.git("init", "-b", "dev")
+        self.git("config", "user.name", "Agentctl Test")
+        self.git("config", "user.email", "agentctl-test@example.invalid")
+        (self.root / "scripts").mkdir()
+        for name in (
+            "agentctl.py", "agentctl_w4.py", "agentctl_catalog.py",
+            "agentctl_visibility.py", "agentctl_console.py", "pi_subagents.py",
+            "agent_worktree.py",
+        ):
+            src = SOURCE.parent / name
+            if src.exists():
+                shutil.copy2(src, self.root / "scripts" / name)
+        static_dest = self.root / "scripts" / "governance-console"
+        static_dest.mkdir(parents=True, exist_ok=True)
+        if (CONSOLE_STATIC / "index.html").exists():
+            shutil.copy2(CONSOLE_STATIC / "index.html", static_dest / "index.html")
+        else:
+            (static_dest / "index.html").write_text("<html>ok</html>", encoding="utf-8")
+        dest = self.root / "configs"
+        dest.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(POLICY_SOURCE, dest / "agent-governance.json")
+        (self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
+        for directory in ("src/a", ".doc/ops", ".doc/status", ".doc/test-reports"):
+            (self.root / directory).mkdir(parents=True, exist_ok=True)
+        (self.root / "src/a" / ".keep").write_text("keep\n", encoding="utf-8")
+        self._write_current([])
+        (self.root / ".doc/status/FEATURE-MATRIX.md").write_text(
+            "# Features\n\n| record_id | 模块 | 状态 |\n| --- | --- | --- |\n| core | core | planned |\n",
+            encoding="utf-8")
+        self.git("add", ".")
+        self.git("commit", "-m", "fixture")
+        self.cli("init")
+        self._server = None
+        self._thread = None
+
+    def tearDown(self):
+        if self._server is not None:
+            self._server.shutdown()
+            self._server.server_close()
+        if self._thread is not None:
+            self._thread.join(timeout=5)
+
+    def git(self, *args, cwd=None):
+        return subprocess.run(
+            ["git", *args], cwd=cwd or self.root,
+            text=True, capture_output=True, check=True,
+        ).stdout.strip()
+
+    def cli(self, *args, cwd=None, check=True):
+        result = subprocess.run(
+            [sys.executable, "-B", str(self.root / "scripts/agentctl.py"), *args],
+            cwd=cwd or self.root, text=True, capture_output=True, timeout=60,
+        )
+        if check:
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        return result
+
+    def _last_json(self, text: str):
+        lines = [ln for ln in text.splitlines() if ln.strip()]
+        return json.loads(lines[-1])
+
+    def _write_current(self, claim_rows):
+        rows = [
+            "| owner | task | branch | worktree | mutex | status |",
+            "| --- | --- | --- | --- | --- | --- |",
+            *claim_rows,
+        ]
+        body = (
+            "# Coordination\n\n## 焦点\n\nfocus\n\n## 并行认领\n\n"
+            + "\n".join(rows) + "\n\n"
+            "## 清理接管与遗留现场\n\n"
+            "| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n"
+            "| --- | --- | --- | --- |\n"
+            "| none | none | - | - |\n\n"
+            "## 集成槽位\n\n"
+            "| holder | phase | since | scope | queue |\n"
+            "|--------|-------|-------|-------|-------|\n"
+            "| - | idle | 2026-09-24 | test | - |\n\n"
+            "## 受控约束\n\n- c\n\n## 活跃入口\n\n- e\n"
+        )
+        (self.root / ".doc/ops/CURRENT.md").write_text(body, encoding="utf-8")
+
+    def initialized_task(self, task="alpha", capabilities=("docs:publish",)):
+        path = self.root / ".worktrees" / task
+        if not path.exists():
+            self.git("worktree", "add", "-b", "feature/" + task, str(path), "dev")
+        args = ["task", "register", "--id", task, "--owner", "Test Agent",
+                "--scope", "src/a", "--scope", ".doc"]
+        for cap in capabilities:
+            args.extend(("--capability", cap))
+        self.cli(*args, cwd=path)
+        return path
+
+    def start_server(self, host="127.0.0.1", port=None):
+        port = port or _free_port()
+        sys.path.insert(0, str(self.root / "scripts"))
+        import agentctl as ac  # noqa: E402
+        import agentctl_console as console  # noqa: E402
+
+        # Ensure policy loaded for this fixture repo.
+        repo = ac.Repo(str(self.root))
+        ac.require_policy(repo)
+        args = type("A", (), {"host": host, "port": port, "static": "scripts/governance-console"})()
+        # Build server without serve_forever so tests control lifecycle.
+        if host not in ("127.0.0.1", "localhost", "::1"):
+            with self.assertRaises(ac.Denied) as ctx:
+                console.serve(repo, args)
+            return None, ctx.exception
+
+        from http.server import ThreadingHTTPServer
+        # Re-enter serve setup by calling internal pattern: use CLI subprocess for smoke,
+        # and for unit endpoints use Handler via temporary monkey of serve_forever.
+        holder = {}
+
+        def run():
+            class Stoppable(ThreadingHTTPServer):
+                allow_reuse_address = True
+
+            # Reconstruct Handler by invoking console.serve until emit, then hijack.
+            # Simpler: spawn `agentctl serve` subprocess.
+            pass
+
+        proc = subprocess.Popen(
+            [sys.executable, "-B", str(self.root / "scripts/agentctl.py"),
+             "serve", "--host", host, "--port", str(port)],
+            cwd=str(self.root), text=True,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        )
+
+        def _stop_proc(p=proc):
+            if p.poll() is None:
+                p.terminate()
+                try:
+                    p.wait(timeout=3)
+                except subprocess.TimeoutExpired:
+                    p.kill()
+            if p.stdout:
+                p.stdout.close()
+            if p.stderr:
+                p.stderr.close()
+        self.addCleanup(_stop_proc)
+        # Wait until port accepts connections.
+        deadline = time.time() + 8
+        while time.time() < deadline:
+            if proc.poll() is not None:
+                out = (proc.stdout.read() if proc.stdout else "") + (proc.stderr.read() if proc.stderr else "")
+                raise AssertionError(f"serve exited early: {out}")
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+                    break
+            except OSError:
+                time.sleep(0.05)
+        else:
+            proc.kill()
+            raise AssertionError("serve did not bind in time")
+        return port, proc
+
+    def http_json(self, port, path, method="GET", body=None, expect_status=200):
+        data = None if body is None else json.dumps(body).encode("utf-8")
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{port}{path}",
+            data=data, method=method,
+            headers={"Content-Type": "application/json"} if data is not None else {},
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                raw = resp.read().decode("utf-8")
+                status = resp.status
+                ctype = resp.headers.get("Content-Type", "")
+        except urllib.error.HTTPError as exc:
+            raw = exc.read().decode("utf-8")
+            status = exc.code
+            ctype = exc.headers.get("Content-Type", "")
+        self.assertEqual(status, expect_status, raw)
+        if "json" in ctype or raw[:1] in "{[":
+            return json.loads(raw)
+        return raw
+
+    def test_non_loopback_host_denied(self):
+        sys.path.insert(0, str(self.root / "scripts"))
+        import agentctl as ac  # noqa: E402
+        import agentctl_console as console  # noqa: E402
+        repo = ac.Repo(str(self.root))
+        ac.require_policy(repo)
+        args = type("A", (), {"host": "0.0.0.0", "port": 7432, "static": "scripts/governance-console"})()
+        with self.assertRaises(ac.Denied) as ctx:
+            console.serve(repo, args)
+        self.assertEqual(ctx.exception.code, "INVALID_REQUEST")
+
+    def test_get_eight_endpoints_and_spa(self):
+        port, _ = self.start_server()
+        for path in (
+            "/api/status", "/api/brief", "/api/queue", "/api/claims",
+            "/api/cleanup", "/api/docs/drift", "/api/catalog/drift", "/api/daemon",
+        ):
+            payload = self.http_json(port, path)
+            self.assertIsInstance(payload, (dict, str))
+            if path == "/api/brief":
+                self.assertTrue(isinstance(payload, str) and payload.startswith("#"))
+            else:
+                self.assertIsInstance(payload, dict)
+        brief_json = self.http_json(port, "/api/brief?format=json")
+        self.assertIn("brief", brief_json)
+        html = self.http_json(port, "/", expect_status=200)
+        self.assertIn("agentctl console", html if isinstance(html, str) else str(html))
+
+    def test_unknown_post_shell_404(self):
+        port, _ = self.start_server()
+        payload = self.http_json(port, "/api/shell", method="POST", body={}, expect_status=404)
+        self.assertEqual(payload["code"], "NOT_FOUND")
+
+    def test_claim_new_missing_task_400(self):
+        port, _ = self.start_server()
+        payload = self.http_json(port, "/api/claim/new", method="POST", body={}, expect_status=400)
+        self.assertEqual(payload["code"], "INVALID_REQUEST")
+
+    def test_claim_new_without_docs_publish_403(self):
+        self.initialized_task(capabilities=())
+        port, _ = self.start_server()
+        payload = self.http_json(
+            port, "/api/claim/new", method="POST",
+            body={"task": "alpha", "owner": "Test Agent", "mutex": "m1"},
+            expect_status=403,
+        )
+        self.assertEqual(payload["code"], "CAPABILITY_REQUIRED")
+
+    def test_claim_new_cli_create_replace_and_replay(self):
+        task = self.initialized_task(capabilities=("docs:publish",))
+        rid = "alpha-claim-new-test1"
+        created = self._last_json(self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m1", "--request-id", rid, cwd=task,
+        ).stdout)
+        self.assertEqual(created["status"], "created")
+        text = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertIn("| Test Agent | alpha |", text)
+        self.assertIn("| m1 | in_progress |", text)
+        # Identical row → idempotent (no second apply)
+        again = self._last_json(self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m1", "--request-id", rid, cwd=task,
+        ).stdout)
+        self.assertEqual(again["status"], "idempotent")
+        # Without replace while row differs → refuse
+        denied = self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m2", "--request-id", "alpha-claim-new-test2", cwd=task, check=False,
+        )
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("INVALID_REQUEST", denied.stderr + denied.stdout)
+        replaced = self._last_json(self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m2", "--replace", "--request-id", "alpha-claim-new-test3", cwd=task,
+        ).stdout)
+        self.assertEqual(replaced["status"], "replaced")
+        text2 = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertIn("| m2 | in_progress |", text2)
+
+    def test_integrate_prepare_first_parent_and_dirty_refuse(self):
+        feat = self.initialized_task(task="feat1", capabilities=("docs:publish",))
+        (feat / "src/a").mkdir(parents=True, exist_ok=True)
+        (feat / "src/a/note.txt").write_text("hello\n", encoding="utf-8")
+        self.git("add", "src/a/note.txt", cwd=feat)
+        self.git("commit", "-m", "feat change", cwd=feat)
+        expected = self.git("rev-parse", "refs/heads/dev")
+        # Prepare from main cwd is allowed by design.
+        out = self._last_json(self.cli(
+            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
+            "--integrate-id", "integrate-feat1",
+        ).stdout)
+        self.assertEqual(out["status"], "prepared")
+        self.assertEqual(out["first_parent"], expected)
+        integ = self.root / ".worktrees" / "integrate-feat1"
+        self.assertTrue(integ.is_dir())
+        # Idempotent on same base
+        again = self._last_json(self.cli(
+            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
+            "--integrate-id", "integrate-feat1",
+        ).stdout)
+        self.assertTrue(again.get("idempotent"))
+        # Dirty refuse on different attempt id that reuses? Use existing dirty tree.
+        (integ / "dirt.txt").write_text("x", encoding="utf-8")
+        # Advance expected to force non-idempotent path that hits dirty check —
+        # tip still same so idempotent returns early. Force by corrupting first-parent check:
+        # Create second integrate with dirty pre-created empty? Create dirty empty dir as worktree path.
+        dirty_id = "integrate-dirty"
+        dirty_path = self.root / ".worktrees" / dirty_id
+        # Manually add a fake dirty worktree at expected tip then make dirty.
+        self.git("worktree", "add", "-b", f"feature/{dirty_id}", str(dirty_path), expected)
+        (dirty_path / "dirt2.txt").write_text("y", encoding="utf-8")
+        denied = self.cli(
+            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
+            "--integrate-id", dirty_id, check=False,
+        )
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("INVALID_REQUEST", denied.stderr + denied.stdout)
+
+    def test_pi_packet_rejects_symbolic_and_writes(self):
+        task = self.initialized_task(capabilities=("docs:publish",))
+        # Ensure HEAD tracked file for freeze if needed
+        denied = self.cli(
+            "pi", "packet", "--task", "alpha", "--role", "research",
+            "--goal", "g", "--authorization-ref", "user:test",
+            "--source-sha", "HEAD", cwd=task, check=False,
+        )
+        self.assertNotEqual(denied.returncode, 0)
+        blob = denied.stderr + denied.stdout
+        self.assertTrue(
+            "INVALID_SOURCE_SHA" in blob or "INVALID_REQUEST" in blob or "denied" in blob.lower(),
+            blob,
+        )
+        head = self.git("rev-parse", "HEAD", cwd=task)
+        ok = self._last_json(self.cli(
+            "pi", "packet", "--task", "alpha", "--role", "research",
+            "--goal", "research goal", "--authorization-ref", "user:test",
+            "--source-sha", head, "--request-id", "pkt-1",
+            "--approved-command", "python3 -c print(1)",
+            "--write", cwd=task,
+        ).stdout)
+        self.assertEqual(ok["status"], "ok")
+        self.assertEqual(ok["packet"]["source_sha"], head)
+        self.assertEqual(ok["packet"]["role_id"], "research")
+        rel = ok["path"]
+        self.assertTrue(rel.endswith("packet.json"))
+        written = json.loads((task / rel).read_text(encoding="utf-8"))
+        self.assertEqual(written["request_id"], "pkt-1")
+        self.assertEqual(written["approved_commands"], [["python3", "-c", "print(1)"]])
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/tests/test_codegraph_governance.py

@@ -22,7 +22,7 @@ class CodegraphGovernanceTest(unittest.TestCase):
         (self.root / 'scripts').mkdir()
         for name in ('agent-worktree.sh', 'agent_worktree.py', 'agentctl.py',
                      'agentctl', 'agentctl_w4.py', 'agentctl_catalog.py',
-                     'agentctl_visibility.py'):
+                     'agentctl_visibility.py', 'agentctl_console.py'):
             shutil.copy2(SOURCE / name, self.root / 'scripts' / name)
         (self.root / 'configs').mkdir(parents=True, exist_ok=True)
         shutil.copy2(SOURCE.parent / 'configs' / 'agent-governance.json',

scripts/tests/test_ops_guards.py

@@ -19,7 +19,7 @@ class OpsGuardsTest(unittest.TestCase):
         self.root = Path(self.tmp.name) / "repo"
         self.root.mkdir()
         (self.root / "scripts").mkdir()
-        for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py"):
+        for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py", "agentctl_console.py"):
             shutil.copy2(SOURCE / name, self.root / "scripts" / name)
         (self.root / ".gitignore").write_text(".worktrees/\n.locks/\nconfigs/.env.*\n!configs/.env.*.example\ndata/\n")
         (self.root / ".dockerignore").write_text(".doc\n.locks\n.worktrees\n")

scripts/tests/test_pi_subagents.py

@@ -152,6 +152,9 @@ class PiSubagentsTests(unittest.TestCase):
         vis_src = ROOT / "scripts" / "agentctl_visibility.py"
         if vis_src.exists():
             shutil.copy2(vis_src, self.root / "scripts" / "agentctl_visibility.py")
+        cons_src = ROOT / "scripts" / "agentctl_console.py"
+        if cons_src.exists():
+            shutil.copy2(cons_src, self.root / "scripts" / "agentctl_console.py")
         if SOURCE_EXEC.exists():
             shutil.copy2(SOURCE_EXEC, self.root / "scripts" / "pi_subagents.py")
         if SOURCE_MCP.exists():