#!/usr/bin/env python3
"""Local, cooperative agent coordination. OS isolation is intentionally out of scope."""
import argparse
import contextlib
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import sqlite3
import subprocess
import sys
import tempfile
import time
import uuid

VERSION = 1
SHARED_DOCS = {'.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'}
ACTIVE = ('running', 'recovery_required')
TERMINAL = ('completed', 'failed', 'recovered')
GIT_CONTEXT = ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE', 'GIT_PREFIX')


class Denied(Exception):
    def __init__(self, code, message, **details):
        self.code, self.message, self.details = code, message, details
        super().__init__(message)


def deny(code, message, **details):
    raise Denied(code, message, **details)


def emit(value, stream=sys.stdout):
    print(json.dumps(value, ensure_ascii=False, sort_keys=True), file=stream, flush=True)


def digest(value):
    return hashlib.sha256(value.encode() if isinstance(value, str) else value).hexdigest()


def git(path, *args, index=False, check=True, input=None):
    env = os.environ.copy()
    for key in GIT_CONTEXT:
        if key != 'GIT_INDEX_FILE' or not index:
            env.pop(key, None)
    result = subprocess.run(['git', '-C', str(path), *args], env=env,
                            input=input, capture_output=True, text=True)
    if check and result.returncode:
        deny('GIT_ERROR', result.stderr.strip() or 'Git operation failed')
    return result.stdout.rstrip('\n') if check else result


def proc(pid):
    try:
        raw = Path('/proc', str(pid), 'stat').read_text().rsplit(')', 1)[1].split()
        return {'state': raw[0], 'ppid': int(raw[1]), 'group': int(raw[2]), 'start': raw[19]}
    except (OSError, ValueError, IndexError):
        return None


def boot():
    return Path('/proc/sys/kernel/random/boot_id').read_text().strip()


def alive(pid, start, boot_id):
    info = proc(pid)
    return bool(info and info['state'] != 'Z' and info['start'] == start and boot_id == boot())


def descendants_of(pid):
    current = os.getpid()
    seen = set()
    while current > 1 and current not in seen:
        if current == pid:
            return True
        seen.add(current)
        info = proc(current)
        if not info:
            break
        current = info['ppid']
    return False


def group_alive(group):
    if not group:
        return False
    for item in Path('/proc').iterdir():
        if item.name.isdigit():
            info = proc(int(item.name))
            if info and info['group'] == group and info['state'] != 'Z':
                return True
    return False


def relative(value):
    if '\x00' in value or '\n' in value or '\\' in value or value.startswith('/'):
        deny('PATH_OUT_OF_SCOPE', 'Expected a repository-relative path')
    parts = Path(value).parts
    if '..' in parts:
        deny('PATH_OUT_OF_SCOPE', 'Parent traversal is not allowed')
    return Path(value).as_posix().rstrip('/') or '.'


def covers(parent, child):
    return parent == '.' or parent == child or child.startswith(parent + '/')


def resource(value):
    if value.startswith('path:'):
        return 'path:' + relative(value[5:])
    if value in ('docs:publish', 'git:worktrees', 'git:index') or re.fullmatch(r'(env|git):(dev|test|main|shared)', value):
        return value
    if value.startswith('tree:'):
        return value
    deny('INVALID_RESOURCE', 'Unsupported resource', resource=value)


def overlaps(a, b):
    if a == b:
        return True
    if a.startswith('path:') and b.startswith('path:'):
        return covers(a[5:], b[5:]) or covers(b[5:], a[5:])
    # Publishing protected docs conflicts with direct managed path writers, too.
    if a == 'docs:publish' and b.startswith('path:'):
        return any(covers(b[5:], p) or covers(p, b[5:]) for p in SHARED_DOCS)
    if b == 'docs:publish' and a.startswith('path:'):
        return overlaps(b, a)
    return False


def safe_path(root, rel):
    rel = relative(rel)
    target = root / rel
    if not target.resolve().is_relative_to(root.resolve()):
        deny('PATH_OUT_OF_SCOPE', 'Path escapes the worktree')
    cur = root
    for part in Path(rel).parts:
        cur /= part
        if cur.is_symlink():
            deny('PATH_OUT_OF_SCOPE', 'Symlink paths are not managed')
    if target.exists() and target.is_file() and target.stat().st_nlink > 1:
        deny('PATH_OUT_OF_SCOPE', 'Hardlinked working files are not managed')
    return target


def secret_path(name):
    parts = Path(name).parts
    leaf = Path(name).name
    return ('.locks' in parts or '.git' in parts or (parts and 'data' == parts[0])
            or leaf == '.env' or ('.env.' in leaf and not leaf.endswith('.example'))
            or leaf in ('.env.local', 'phase-secrets.local.json')
            or any(p in ('.secrets', 'node_modules') for p in parts))


def clean_source(path, build=False):
    if not build:
        dirty = bool(git(path, 'status', '--porcelain', '--untracked-files=all'))
    else:
        tracked = git(path, 'status', '--porcelain', '--untracked-files=no')
        untracked = git(path, 'ls-files', '--others', '--exclude-standard', '-z').split('\0')
        # These are the existing nightly producer's non-source outputs. git archive
        # excludes all untracked files, independently of Docker ignore patterns.
        evidence_prefix = '.doc/test-reports/assets/promotion-receipts/'
        dirty = bool(tracked or any(n and not n.startswith(evidence_prefix) for n in untracked))
    if dirty:
        deny('DIRTY_TREE', 'Source worktree must be clean; only generated build evidence is excepted')


class Repo:
    def __init__(self, path):
        self.current = Path(git(Path(path).resolve(), 'rev-parse', '--show-toplevel')).resolve()
        self.common = Path(git(self.current, 'rev-parse', '--path-format=absolute', '--git-common-dir')).resolve()
        rows = self.worktrees()
        if not rows or rows[0].get('bare'):
            deny('UNSUPPORTED_REPOSITORY', 'A non-bare main worktree is required')
        self.main = Path(rows[0]['worktree']).resolve()
        if self.common != (self.main / '.git').resolve():
            deny('UNSUPPORTED_REPOSITORY', 'External Git directories require explicit migration')
        self.root = self.main / '.locks'
        if self.root.is_symlink():
            deny('INVALID_LOCK_ROOT', 'The canonical .locks directory must not be a symlink')
        self.dbfile = self.root / 'state.sqlite'

    def worktrees(self):
        records = []
        for block in git(self.current, 'worktree', 'list', '--porcelain', '-z').split('\0\0'):
            record = {}
            for field in block.split('\0'):
                if field:
                    key, _, value = field.partition(' ')
                    record[key] = value
            if record:
                records.append(record)
        return records

    def identity(self):
        try:
            identity = json.loads((self.root / 'identity.json').read_text())
        except FileNotFoundError:
            deny('NOT_INITIALIZED', 'Run agentctl init explicitly from the main worktree')
        except (ValueError, OSError):
            deny('RECOVERY_REQUIRED', 'Lock identity cannot be read')
        if (identity.get('main') != str(self.main) or identity.get('common') != str(self.common)
                or identity.get('version') != VERSION or not self.dbfile.exists()):
            deny('POLICY_MISMATCH', 'Lock identity, schema or repository topology changed')
        old = self.common / 'agent-control' / 'dev-integration.lock'
        bridge = self.root / 'os' / 'dev-integration.lock'
        if not old.exists() or not bridge.exists() or old.is_symlink() or bridge.is_symlink() or not os.path.samefile(old, bridge):
            deny('RECOVERY_REQUIRED', 'Compatibility lock identity changed; stop and inspect')
        if self.dbfile.is_symlink() or (self.root / 'os').is_symlink():
            deny('INVALID_LOCK_ROOT', 'Coordination state must not use symlink aliases')
        return identity

    @contextlib.contextmanager
    def db(self, write=False):
        self.identity()
        # mode=rw never silently creates a fresh authority when state is missing.
        db = sqlite3.connect(self.dbfile.as_uri() + '?mode=' + ('rw' if write else 'ro'),
                             uri=True, timeout=5, isolation_level=None)
        db.row_factory = sqlite3.Row
        try:
            db.execute('PRAGMA busy_timeout=5000')
            if write:
                db.execute('PRAGMA synchronous=FULL')
                db.execute('BEGIN IMMEDIATE')
            yield db
            if write:
                db.commit()
        except BaseException:
            if write:
                db.rollback()
            raise
        finally:
            db.close()

    def os_file(self, name):
        self.identity()
        path = self.root / 'os' / name
        if path.is_symlink():
            deny('INVALID_LOCK_ROOT', 'Stable OS lock must be a regular file')
        return path


def atomic_write(path, content):
    data = content.encode() if isinstance(content, str) else content
    mode = path.stat().st_mode & 0o777 if path.exists() else 0o600
    fd, tmp = tempfile.mkstemp(prefix='.' + path.name + '.agentctl-', dir=path.parent)
    try:
        os.fchmod(fd, mode)
        with os.fdopen(fd, 'wb') as f:
            f.write(data)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, path)
        directory = os.open(path.parent, os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
    finally:
        if os.path.exists(tmp):
            os.unlink(tmp)


def initialize(repo):
    if repo.current != repo.main:
        deny('MAIN_WORKTREE_REQUIRED', 'Initialize from the main worktree')
    legacy = repo.common / 'agent-control'
    legacy.mkdir(exist_ok=True)
    with (legacy / 'init.lock').open('a+') as initialization:
        fcntl.flock(initialization, fcntl.LOCK_EX | fcntl.LOCK_NB)
        if (repo.root / 'identity.json').exists():
            identity = repo.identity()
            with repo.db() as db:
                db.execute('SELECT count(*) FROM runs').fetchone()
            emit({'status': 'initialized', 'lock_root': str(repo.root), **identity})
            return
        old = legacy / 'dev-integration.lock'
        if old.is_symlink():
            deny('RECOVERY_REQUIRED', 'Legacy lock is a symlink; inspect before migration')
        with old.open('a+') as devlock:
            try:
                fcntl.flock(devlock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                deny('LOCK_BUSY', 'Legacy dev integration is currently locked')
            state = legacy / 'active-operation.json'
            if state.exists():
                value = json.loads(state.read_text())
                if value.get('phase') not in ('idle', 'released', 'closed', 'completed'):
                    deny('RECOVERY_REQUIRED', 'Legacy operation is not terminal; inspect its owner first')
            repo.root.mkdir(mode=0o700, exist_ok=True)
            if repo.dbfile.is_symlink() or (repo.root/'identity.json').is_symlink() or (repo.root/'os').is_symlink():
                deny('INVALID_LOCK_ROOT', 'Initialization will not follow existing state symlinks')
            os.chmod(repo.root, 0o700)
            (repo.root / 'os').mkdir(mode=0o700, exist_ok=True)
            bridge = repo.root / 'os' / 'dev-integration.lock'
            if not bridge.exists():
                os.link(old, bridge)
            if old.stat().st_ino != bridge.stat().st_ino or old.stat().st_dev != bridge.stat().st_dev:
                deny('RECOVERY_REQUIRED', 'Old and new dev locks do not share the same inode')
            # Initialization may resume an interrupted schema creation, never a mismatched identity.
            db = sqlite3.connect(repo.dbfile)
            try:
                db.execute('PRAGMA journal_mode=WAL')
                db.execute('PRAGMA synchronous=FULL')
                db.executescript('''
                CREATE TABLE IF NOT EXISTS tasks(id TEXT PRIMARY KEY, owner TEXT NOT NULL,
                  worktree TEXT NOT NULL, branch TEXT NOT NULL, scopes TEXT NOT NULL,
                  capabilities TEXT NOT NULL, state TEXT NOT NULL, base_sha TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS runs(generation INTEGER PRIMARY KEY AUTOINCREMENT,
                  id TEXT UNIQUE NOT NULL, task TEXT NOT NULL, fingerprint TEXT NOT NULL,
                  resources TEXT NOT NULL, mode TEXT NOT NULL, state TEXT NOT NULL,
                  pid INTEGER NOT NULL, start TEXT NOT NULL, boot TEXT NOT NULL,
                  child_pid INTEGER, child_start TEXT, pgid INTEGER, created REAL NOT NULL,
                  heartbeat REAL NOT NULL, exit_code INTEGER, reason TEXT, kind TEXT NOT NULL DEFAULT 'command');
                CREATE INDEX IF NOT EXISTS runs_active ON runs(state, generation);
                CREATE INDEX IF NOT EXISTS runs_task ON runs(task, state);
                CREATE TABLE IF NOT EXISTS documents(run TEXT PRIMARY KEY, task TEXT NOT NULL,
                  worktree TEXT NOT NULL, state TEXT NOT NULL, payload TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS receipts(task TEXT, file TEXT, hash TEXT,
                  PRIMARY KEY(task,file));
                CREATE TABLE IF NOT EXISTS validations(task TEXT, sha TEXT, command_hash TEXT,
                  created REAL, PRIMARY KEY(task,sha));
                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);
                ''')
                db.commit()
            finally:
                db.close()
            os.chmod(repo.dbfile, 0o600)
            identity = {'version': VERSION, 'repo_id': str(uuid.uuid4()), 'epoch': str(uuid.uuid4()),
                        'main': str(repo.main), 'common': str(repo.common)}
            atomic_write(repo.root / 'identity.json', json.dumps(identity))
            emit({'status': 'initialized', 'lock_root': str(repo.root), **identity})


def task_row(repo, task_id):
    with repo.db() as db:
        row = db.execute('SELECT * FROM tasks WHERE id=?', (task_id,)).fetchone()
    if not row:
        deny('UNKNOWN_TASK', 'Register the task before requesting a lease')
    if row['state'] != 'active':
        deny('TASK_CLOSED', 'This task is no longer active')
    if not Path(row['worktree']).exists() or git(row['worktree'], 'branch', '--show-current') != row['branch']:
        deny('WORKTREE_MISMATCH', 'Registered task branch/worktree changed')
    return dict(row)


def check_resources(repo, task, resources):
    scopes, caps = json.loads(task['scopes']), json.loads(task['capabilities'])
    for value in resources:
        if value.startswith('path:'):
            p = value[5:]
            if p != '.' and secret_path(p):
                deny('PATH_OUT_OF_SCOPE', 'Runtime and Git administration paths are not task source scopes')
            safe_path(Path(task['worktree']), p)
            if not any(covers(s, p) for s in scopes):
                deny('PATH_OUT_OF_SCOPE', 'Resource is outside the registered write scope', resource=value)
        elif value.startswith('tree:'):
            if value != 'tree:' + task['worktree']:
                deny('PATH_OUT_OF_SCOPE', 'Task cannot acquire another worktree index')
        elif value not in caps:
            deny('CAPABILITY_REQUIRED', 'Task lacks this registered capability', resource=value)


def event(db, operation, detail):
    db.execute('INSERT INTO events(created,operation,detail) VALUES(?,?,?)',
               (time.time(), operation, json.dumps(detail, ensure_ascii=False)))


def physical_names(resources):
    names = set()
    for item in resources:
        if item in ('git:dev', 'env:dev'):
            names.add('dev-integration.lock')
        if item == 'git:worktrees':
            names.add('worktrees.lock')
        if item == 'docs:publish' or (item.startswith('path:') and any(covers(item[5:], p) for p in SHARED_DOCS)):
            names.add('shared-writeback.lock')
        if item.startswith('tree:'):
            names.add('tree-' + digest(item) + '.lock')
    return sorted(names)


class Lease:
    def __init__(self, repo, task, resources, request_id, fingerprint, wait=0, mode='X', kind='command'):
        self.repo, self.task = repo, task
        self.resources = sorted(set('tree:' + task['worktree'] if r == 'git:index' else resource(r) for r in resources))
        check_resources(repo, task, self.resources)
        self.request_id = request_id or str(uuid.uuid4())
        if not isinstance(self.request_id, str) or not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', self.request_id):
            deny('INVALID_REQUEST', 'Invalid request id')
        self.fingerprint, self.wait, self.mode, self.kind = fingerprint, max(0, min(wait, 60)), mode, kind
        self.handles, self.row, self.replay = [], None, False

    def unlock(self):
        for handle in reversed(self.handles):
            handle.close()
        self.handles = []

    def __enter__(self):
        info = proc(os.getpid())
        with self.repo.db(True) as db:
            old = db.execute('SELECT * FROM runs WHERE id=?', (self.request_id,)).fetchone()
            if old:
                if old['task'] != self.task['id'] or old['fingerprint'] != self.fingerprint or old['kind'] != self.kind:
                    deny('REQUEST_ID_REUSED', 'Request id belongs to a different operation')
                if old['state'] in TERMINAL:
                    self.row, self.replay = dict(old), True
                    return self
                if old['state'] not in ('busy',):
                    deny('RECOVERY_REQUIRED', 'Operation already exists; query or recover it')
                db.execute("UPDATE runs SET state='queued',pid=?,start=?,boot=?,heartbeat=? WHERE id=?",
                           (os.getpid(), info['start'], boot(), time.time(), self.request_id))
            else:
                db.execute('''INSERT INTO runs(id,task,fingerprint,resources,mode,state,pid,start,boot,created,heartbeat,kind)
                              VALUES(?,?,?,?,?,'queued',?,?,?,?,?,?)''',
                           (self.request_id, self.task['id'], self.fingerprint, json.dumps(self.resources),
                            self.mode, os.getpid(), info['start'], boot(), time.time(), time.time(), self.kind))
            self.row = dict(db.execute('SELECT * FROM runs WHERE id=?', (self.request_id,)).fetchone())
        deadline = time.monotonic() + self.wait
        holder = None
        while True:
            try:
                for name in physical_names(self.resources):
                    handle = self.repo.os_file(name).open('a+')
                    self.handles.append(handle)
                    fcntl.flock(handle, (fcntl.LOCK_SH if self.mode == 'S' else fcntl.LOCK_EX) | fcntl.LOCK_NB)
                # Lifecycle lock prevents cleanup between registry validation and execution.
                lifecycle = self.repo.os_file('worktree-' + digest(self.task['worktree']) + '.lock').open('a+')
                self.handles.append(lifecycle)
                fcntl.flock(lifecycle, fcntl.LOCK_SH | fcntl.LOCK_NB)
                task_row(self.repo, self.task['id'])
                with self.repo.db(True) as db:
                    for frozen in db.execute('SELECT * FROM freezes'):
                        if self.mode == 'X' and any(overlaps(r, frozen['resource']) for r in self.resources):
                            deny('FROZEN', 'Resource is frozen', resource=frozen['resource'])
                    candidates = db.execute("SELECT * FROM runs WHERE id!=? AND state IN ('running','recovery_required','queued') ORDER BY generation", (self.request_id,)).fetchall()
                    for other in candidates:
                        if other['state'] == 'queued':
                            if not alive(other['pid'], other['start'], other['boot']):
                                db.execute("UPDATE runs SET state='busy' WHERE id=?", (other['id'],))
                                continue
                            if other['generation'] > self.row['generation']:
                                continue
                        if self.mode == other['mode'] == 'S':
                            continue
                        if any(overlaps(a,b) for a in self.resources for b in json.loads(other['resources'])):
                            holder = {'task': other['task'], 'run_id': other['id'], 'state': other['state']}
                            raise BlockingIOError()
                    db.execute("UPDATE runs SET state='running',heartbeat=? WHERE id=?", (time.time(), self.request_id))
                    event(db, self.request_id, {'event': 'acquired', 'task': self.task['id'], 'resources': self.resources})
                return self
            except BlockingIOError:
                self.unlock()
                if time.monotonic() >= deadline:
                    with self.repo.db(True) as db:
                        db.execute("UPDATE runs SET state='busy' WHERE id=?", (self.request_id,))
                    deny('LOCK_BUSY', 'Resource is busy; retry this request id or do independent work',
                         request_id=self.request_id, holder=holder, retry_after=1)
                time.sleep(min(0.1, max(0, deadline - time.monotonic())))
            except BaseException:
                self.unlock()
                with self.repo.db(True) as db:
                    db.execute("UPDATE runs SET state='failed',exit_code=1 WHERE id=? AND state='queued'", (self.request_id,))
                raise

    def finish(self, code=0, recovery=False):
        if self.replay:
            return
        with self.repo.db(True) as db:
            db.execute('UPDATE runs SET state=?,exit_code=?,heartbeat=? WHERE id=?',
                       ('recovery_required' if recovery else ('completed' if code == 0 else 'failed'),
                        code, time.time(), self.request_id))
            event(db, self.request_id, {'event': 'finished', 'exit_code': code, 'recovery_required': recovery})

    def __exit__(self, typ, error, tb):
        if typ and not self.replay:
            try:
                with self.repo.db(True) as db:
                    doc = db.execute('SELECT state FROM documents WHERE run=?', (self.request_id,)).fetchone()
                    state = db.execute('SELECT * FROM runs WHERE id=?', (self.request_id,)).fetchone()
                if state and state['state'] == 'running':
                    self.finish(1, bool(doc and doc['state'] == 'prepared') or group_alive(state['pgid']))
            finally:
                self.unlock()
        else:
            self.unlock()


def inherited(repo, required=None):
    run_id = os.getenv('AGENTCTL_RUN_ID', '')
    with repo.db() as db:
        row = db.execute('SELECT * FROM runs WHERE id=?', (run_id,)).fetchone()
    if not row or row['state'] != 'running':
        deny('LEASE_REQUIRED', 'Run the operation through agentctl with')
    if (os.getenv('AGENTCTL_TASK_ID') != row['task']
            or os.getenv('AGENTCTL_GENERATION') != str(row['generation'])
            or os.getenv('AGENTCTL_EPOCH') != repo.identity()['epoch']
            or not alive(row['pid'], row['start'], row['boot'])
            or not descendants_of(row['pid'])):
        deny('STALE_GENERATION', 'Lease identity is stale or caller is not a supervised descendant')
    if required:
        resources = json.loads(row['resources'])
        required = 'tree:' + task_row(repo, row['task'])['worktree'] if required == 'git:index' else resource(required)
        permitted = required in resources or (required.startswith('path:') and any(r.startswith('path:') and covers(r[5:], required[5:]) for r in resources))
        if not permitted or row['mode'] != 'X':
            deny('LEASE_REQUIRED', 'Required exclusive resource is not held', resource=required)
    task = task_row(repo, row['task'])
    return dict(row), task


def supervise(repo, lease, command, cwd, env=None):
    """Retain the lease until the child group stops or recovery is recorded."""
    child = None
    stopping = []

    def stop(signum, frame):
        stopping.append(signum)
        if child is not None:
            with contextlib.suppress(ProcessLookupError):
                os.killpg(child.pid, signal.SIGTERM)

    previous = {sig: signal.signal(sig, stop) for sig in (signal.SIGINT, signal.SIGTERM)}
    try:
        if stopping:
            return 128 + stopping[0], False
        child = subprocess.Popen(command, cwd=cwd, env=env, start_new_session=True)
        info = proc(child.pid)
        with repo.db(True) as db:
            db.execute('UPDATE runs SET child_pid=?,child_start=?,pgid=? WHERE id=?',
                       (child.pid, info['start'] if info else '', child.pid, lease.request_id))
        last_beat, stop_at = time.monotonic(), None
        while child.poll() is None:
            if stopping:
                stop_at = stop_at or time.monotonic()
                with contextlib.suppress(ProcessLookupError):
                    os.killpg(child.pid, signal.SIGKILL if time.monotonic() - stop_at > 5 else signal.SIGTERM)
            if time.monotonic() - last_beat > 10:
                with repo.db(True) as db:
                    db.execute('UPDATE runs SET heartbeat=? WHERE id=?', (time.time(), lease.request_id))
                last_beat = time.monotonic()
            time.sleep(0.1)
        code = child.wait()
        return (code if code >= 0 else 128 - code), group_alive(child.pid)
    except BaseException:
        if child is not None:
            with contextlib.suppress(ProcessLookupError):
                os.killpg(child.pid, signal.SIGKILL)
            child.wait()
            lease.finish(1, True)
        raise
    finally:
        for sig, handler in previous.items():
            signal.signal(sig, handler)


def execute(repo, args):
    task = task_row(repo, args.task)
    command = args.command[1:] if args.command[:1] == ['--'] else args.command
    if not command:
        deny('INVALID_REQUEST', 'Expected a child command after --')
    cwd = Path(args.cwd).resolve() if args.cwd else Path(task['worktree'])
    known = {Path(r['worktree']).resolve() for r in repo.worktrees()}
    if cwd not in known or cwd not in (Path(task['worktree']), repo.main):
        deny('WORKTREE_MISMATCH', 'Command cwd must be this task worktree or authorized main operation')
    if cwd == repo.main and not any(r.startswith('env:') or r in ('git:dev','git:test','git:main','git:worktrees') for r in args.resource):
        deny('MAIN_WORKTREE_REQUIRED', 'Main checkout is reserved for registered integration/environment operations')
    if cwd == repo.main and 'git:dev' in args.resource:
        deny('USE_INTEGRATE', 'Protected dev updates use agentctl integrate, not an arbitrary main-worktree command')
    fingerprint = digest(json.dumps([command, str(cwd), sorted(args.resource), args.shared]))
    with Lease(repo, task, args.resource, args.request_id, fingerprint, args.wait,
               'S' if args.shared else 'X') as lease:
        if lease.replay:
            emit({'status': 'replayed', 'run_id': lease.request_id, 'exit_code': lease.row['exit_code']}, sys.stderr)
            return lease.row['exit_code'] or 0
        env = os.environ.copy()
        env.update(AGENTCTL_RUN_ID=lease.request_id, AGENTCTL_TASK_ID=args.task,
                   AGENTCTL_GENERATION=str(lease.row['generation']), AGENTCTL_EPOCH=repo.identity()['epoch'])
        emit({'status': 'running', 'run_id': lease.request_id, 'generation': lease.row['generation']}, sys.stderr)
        code, remaining = supervise(repo, lease, command, cwd, env)
        recovery = remaining or bool(code and any(r.startswith(('env:', 'git:dev')) for r in args.resource))
        lease.finish(code, recovery)
        if recovery:
            emit({'status': 'recovery_required', 'run_id': lease.request_id, 'child_group_alive': remaining}, sys.stderr)
        return code or (1 if remaining else 0)


def doc_rows(content, record_id):
    matches = []
    for number, line in enumerate(content.splitlines()):
        if line.lstrip(' ').startswith('|'):
            cells = [cell.strip() for cell in line.strip().strip('|').split('|')]
            if record_id in cells[:2]:
                matches.append((number, line))
    if len(matches) > 1:
        deny('AMBIGUOUS_RECORD', 'Record id is not unique in this document')
    return matches


def table_cells(line):
    return [cell.strip() for cell in re.split(r'(?<!\\)\|', line.strip().strip('|'))]


def matrix_states(content):
    states, status_index, width = {}, None, None
    for line in content.splitlines():
        if not line.lstrip(' ').startswith('|'):
            if status_index is not None:
                break
            continue
        cells = table_cells(line)
        if status_index is None:
            if cells[0].casefold() not in ('模块', 'module', 'feature'):
                continue
            labels = [c.strip('*').casefold() for c in cells]
            columns = [i for i,c in enumerate(labels) if c in ('状态','state','status')]
            if len(columns) != 1:
                deny('INVALID_DOCUMENT', 'Matrix needs one explicit status column')
            status_index, width = columns[0], len(cells)
            continue
        if all(re.fullmatch(r':?-+:?', cell.replace(' ', '')) for cell in cells):
            continue
        if len(cells) != width or cells[0] in states:
            deny('INVALID_DOCUMENT', 'Matrix records must have unique ids and match its columns')
        value = cells[status_index]
        if value not in ('done','partial','planned','deferred','blocked','removed'):
            deny('INVALID_STATE', 'Matrix status must use an unformatted canonical state')
        states[cells[0]] = value
    if status_index is None:
        deny('INVALID_DOCUMENT', 'Matrix table header is missing')
    return states


def docs_snapshot(repo, args):
    if args.file not in SHARED_DOCS:
        deny('PATH_OUT_OF_SCOPE', 'Only the two shared governance documents are managed')
    content = safe_path(repo.current, args.file).read_text()
    row = [(0, content)] if args.record == '@document' else doc_rows(content, args.record)
    emit({'file': args.file, 'record_id': args.record, 'sha256': digest(row[0][1]) if row else None,
          'row': row[0][1] if row else None, 'document_sha256': digest(content)})


def docs_apply(repo, args):
    task = task_row(repo, args.task)
    if repo.current != Path(task['worktree']) or repo.current == repo.main:
        deny('WORKTREE_MISMATCH', 'Shared documents are edited only in the task worktree')
    request = json.loads(Path(args.request).read_text())
    if not isinstance(request, dict) or not isinstance(request.get('changes'), list) or not request['changes'] or any(not isinstance(c, dict) for c in request['changes']):
        deny('INVALID_REQUEST', 'Expected a nonempty changes array')
    resources = ['docs:publish']
    for change in request['changes']:
        if not isinstance(change.get('file'), str) or change['file'] not in SHARED_DOCS:
            deny('PATH_OUT_OF_SCOPE', 'Unknown shared document')
        resources.append('path:' + change['file'])
    with Lease(repo, task, resources, request.get('request_id'), digest(json.dumps(request, sort_keys=True))) as lease:
        if lease.replay:
            emit({'status': 'replayed', 'run_id': lease.request_id, 'exit_code': lease.row['exit_code']})
            return lease.row['exit_code'] or 0
        before, after, seen = {}, {}, set()
        for change in request['changes']:
            filename, record_id = change['file'], change.get('record_id')
            if not isinstance(record_id, str) or not record_id or (filename, record_id) in seen:
                deny('INVALID_REQUEST', 'Each record must have a unique nonempty id')
            seen.add((filename, record_id))
            if record_id == '@document' and 'docs:admin' not in json.loads(task['capabilities']):
                deny('CAPABILITY_REQUIRED', 'Document layout changes require docs:admin')
            if filename.endswith('CURRENT.md') and record_id != args.task and 'docs:admin' not in json.loads(task['capabilities']):
                deny('PATH_OUT_OF_SCOPE', 'A task can update only its own CURRENT record')
            if record_id in ('holder', 'idle', 'Cursor', 'Codex', '-'):
                deny('PATH_OUT_OF_SCOPE', 'Use task ids; slot ownership is not a document edit')
            path = safe_path(repo.current, filename)
            if filename not in before:
                before[filename] = path.read_text()
                base = git(repo.current, 'show', 'HEAD:' + filename, check=False)
                with repo.db() as db:
                    prior = db.execute('SELECT hash FROM receipts WHERE task=? AND file=?', (args.task, filename)).fetchone()
                if base.returncode != 0 or (before[filename] != base.stdout and (not prior or prior['hash'] != digest(before[filename]))):
                    deny('UNMANAGED_DOCUMENT_CHANGE', 'Shared document differs from HEAD and its last managed receipt; preserve and reconcile first')
                after[filename] = before[filename]
            existing = [(0, before[filename])] if record_id == '@document' else doc_rows(before[filename], record_id)
            expected = digest(existing[0][1]) if existing else None
            if 'expected_sha256' not in change or change['expected_sha256'] != expected:
                deny('BASE_CHANGED', 'Record changed; read the current snapshot before editing', record_id=record_id)
            replacement = change.get('replacement')
            if record_id == '@document':
                if not isinstance(replacement, str):
                    deny('INVALID_REQUEST', 'Expected full document text')
                if any(other.get('file') == filename and other is not change for other in request['changes']):
                    deny('INVALID_REQUEST', 'Whole-document operation cannot mix row edits for the same file')
                if filename.endswith('CURRENT.md'):
                    def slot(text):
                        match = re.search(r'(?m)^## 集成槽位\s*\n(.*?)(?=^## |\Z)', text, re.S | re.M)
                        return match.group(1) if match else None
                    if slot(replacement) != slot(before[filename]):
                        deny('PATH_OUT_OF_SCOPE', 'Document admin cannot overwrite integration slot ownership')
                # A layout update cannot bypass the feature-completion evidence rule.
                if filename.endswith('FEATURE-MATRIX.md'):
                    old_states = matrix_states(before[filename])
                    new_states = matrix_states(replacement)
                    if any(value == 'done' and old_states.get(key) != 'done' for key,value in new_states.items()):
                        deny('EVIDENCE_REQUIRED', 'Change capability completion using a reviewed row update')
                after[filename] = replacement
                continue
            if not isinstance(replacement, str) or len(replacement.splitlines()) != 1 or not replacement.startswith('|'):
                deny('INVALID_REQUEST', 'Replacement must be one Markdown table row')
            cells = table_cells(replacement)
            if record_id not in cells[:2]:
                deny('INVALID_REQUEST', 'Replacement must preserve the record id')
            if filename.endswith('FEATURE-MATRIX.md') and 'done' in cells and matrix_states(before[filename]).get(record_id) != 'done':
                evidence = change.get('evidence', [])
                if 'docs:approve' not in json.loads(task['capabilities']) or not isinstance(evidence, list) or not evidence or any(not isinstance(e, str) for e in evidence):
                    deny('EVIDENCE_REQUIRED', 'Capability completion requires approval capability and committed evidence')
                for evidence_file in evidence:
                    safe_path(repo.current, evidence_file)
                    git(repo.current, 'cat-file', '-e', 'HEAD:' + relative(evidence_file))
            lines = after[filename].splitlines()
            live = doc_rows(after[filename], record_id)
            if live:
                lines[live[0][0]] = replacement
            else:
                header = '| owner | task |' if filename.endswith('CURRENT.md') else '| 模块 |'
                positions = [i for i, line in enumerate(lines) if line.startswith(header)]
                if len(positions) != 1:
                    deny('INVALID_DOCUMENT', 'Target table header is absent or ambiguous')
                lines.insert(positions[0] + 2, replacement)
            after[filename] = '\n'.join(lines) + '\n'
        for name, content in after.items():
            if name.endswith('FEATURE-MATRIX.md'):
                matrix_states(content)
        payload = {name: {'before': before[name], 'after': after[name]} for name in before}
        with repo.db(True) as db:
            db.execute("INSERT INTO documents VALUES(?,?,?,'prepared',?)",
                       (lease.request_id, args.task, str(repo.current), json.dumps(payload)))
        for name, values in payload.items():
            path = safe_path(repo.current, name)
            if path.read_text() != values['before']:
                deny('BASE_CHANGED', 'Document changed outside the managed entrypoint; recovery required')
            atomic_write(path, values['after'])
        with repo.db(True) as db:
            for name, values in payload.items():
                db.execute('INSERT OR REPLACE INTO receipts VALUES(?,?,?)', (args.task, name, digest(values['after'])))
            db.execute("UPDATE documents SET state='applied' WHERE run=?", (lease.request_id,))
        lease.finish()
        emit({'status': 'applied', 'run_id': lease.request_id,
              'files': {name: digest(values['after']) for name, values in payload.items()}})
    return 0


def guard(repo, args):
    if args.kind == 'build':
        sha = git(repo.current, 'rev-parse', '--verify', args.commit + '^{commit}')
        if git(repo.current, 'rev-parse', 'HEAD') != sha:
            deny('BASE_CHANGED', 'Build source HEAD does not match the requested commit')
        clean_source(repo.current, build=True)
    elif args.kind == 'resource':
        inherited(repo, args.resource)
    elif args.kind == 'env':
        if repo.current != repo.main or git(repo.main, 'branch', '--show-current') != 'dev':
            deny('MAIN_WORKTREE_REQUIRED', 'Shared services run only from the main dev worktree')
        targets = ['dev', 'test', 'main', 'shared'] if args.target == 'all' else [args.target]
        for target in targets:
            inherited(repo, 'env:' + target)
        clean_source(repo.main)
    emit({'status': 'allowed'})


def worktree_check(repo, path):
    target = Path(path).resolve()
    with repo.db() as db:
        tasks = db.execute('SELECT id,state FROM tasks WHERE worktree=?', (str(target),)).fetchall()
        for task in tasks:
            row = db.execute("SELECT id FROM runs WHERE task=? AND state IN ('running','recovery_required','queued')", (task['id'],)).fetchone()
            if row:
                deny('LOCK_BUSY', 'Worktree has a live or unresolved operation', run_id=row['id'])
            if task['state'] == 'active':
                deny('TASK_ACTIVE', 'Close the registered task before removing its worktree')
    emit({'status': 'allowed', 'path': str(target)})


def recover(repo, args):
    with repo.db() as db:
        row = db.execute('SELECT * FROM runs WHERE id=?', (args.run,)).fetchone()
        document = db.execute('SELECT * FROM documents WHERE run=?', (args.run,)).fetchone()
    if not row:
        deny('UNKNOWN_OPERATION', 'Unknown run id')
    if row['state'] in TERMINAL:
        emit({'status': row['state'], 'run_id': args.run})
        return
    if alive(row['pid'], row['start'], row['boot']) or (row['boot'] == boot() and group_alive(row['pgid'])):
        deny('LOCK_BUSY', 'Supervisor or child process group is still alive; no takeover')
    handles = []
    try:
        for name in physical_names(json.loads(row['resources'])):
            handle = repo.os_file(name).open('a+')
            handles.append(handle)
            try:
                fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                deny('LOCK_BUSY', 'A physical operation is still in progress')
        if document and document['state'] == 'prepared':
            root = Path(document['worktree'])
            payload = json.loads(document['payload'])
            for name, values in payload.items():
                if safe_path(root, name).read_text() not in (values['before'], values['after']):
                    deny('BASE_CHANGED', 'Recovery would overwrite third-party content; preserve the worktree')
            for name, values in payload.items():
                if safe_path(root, name).read_text() != values['after']:
                    atomic_write(safe_path(root, name), values['after'])
            with repo.db(True) as db:
                for name, values in payload.items():
                    db.execute('INSERT OR REPLACE INTO receipts VALUES(?,?,?)', (row['task'], name, digest(values['after'])))
                db.execute("UPDATE documents SET state='applied' WHERE run=?", (args.run,))
        with repo.db(True) as db:
            db.execute("UPDATE runs SET state='recovered',reason=?,exit_code=? WHERE id=?", (args.reason, 0 if document else 1, args.run))
            event(db, args.run, {'event': 'operator_recovery', 'reason': args.reason})
        emit({'status': 'recovered', 'run_id': args.run})
    finally:
        for handle in reversed(handles):
            handle.close()


def hook_check(repo, merge=False):
    if repo.current == repo.main or git(repo.current, 'branch', '--show-current') in ('dev', 'test', 'main', ''):
        deny('PROTECTED_BRANCH', 'Direct development commits require a registered feature/fix worktree')
    run, task = inherited(repo, 'git:index')
    if str(repo.current) != task['worktree']:
        deny('WORKTREE_MISMATCH', 'Commit worktree differs from registered task')
    names = git(repo.current, 'diff', '--cached', '--name-only', '-z', index=True).split('\0')
    scopes = json.loads(task['scopes'])
    run_resources = json.loads(run['resources'])
    merge_has_dev_lease = merge and 'git:dev' in run_resources
    for name in filter(None, names):
        if secret_path(name):
            deny('PROTECTED_PATH', 'Sensitive/runtime paths cannot enter a task commit', path=name)
        if not any(covers(scope, name) for scope in scopes):
            deny('PATH_OUT_OF_SCOPE', 'Staged file is outside task scope', path=name)
        safe_path(repo.current, name)
        if not merge_has_dev_lease and not any(r.startswith('path:') and covers(r[5:], name) for r in run_resources):
            deny('LEASE_REQUIRED', 'Staged path is not covered by this operation lease', path=name)
        # Integration merge is separately fenced by git:dev and exact candidate validation.
        if name in SHARED_DOCS and not merge_has_dev_lease:
            value = git(repo.current, 'show', ':' + name, index=True, check=False)
            staged = value.stdout if value.returncode == 0 else ''
            with repo.db() as db:
                receipt = db.execute('SELECT hash FROM receipts WHERE task=? AND file=?', (task['id'], name)).fetchone()
            if not receipt or receipt['hash'] != digest(staged):
                deny('DOCUMENT_RECEIPT_REQUIRED', 'Use docs apply for the staged shared document', path=name)


def admit_candidate(repo, task, candidate, expected):
    parents = git(repo.main, 'rev-list', '--parents', '-n', '1', candidate).split()[1:]
    if len(parents) != 2 or parents[0] != expected:
        deny('INVALID_CANDIDATE', 'Candidate must have two parents and the frozen target as its first parent')
    if git(task['worktree'], 'rev-parse', 'HEAD') != candidate:
        deny('BASE_CHANGED', 'Candidate is not the owning task HEAD')
    with repo.db() as db:
        found = db.execute('SELECT sha FROM validations WHERE task=? AND sha=?', (task['id'], candidate)).fetchone()
    if not found:
        deny('VALIDATION_REQUIRED', 'Validate this exact candidate with its owning task first')


def hook_ref(repo, state):
    lines = sys.stdin.read().splitlines()
    if state != 'prepared':
        return
    for line in lines:
        old, new, ref = line.split()
        if ref not in ('refs/heads/dev', 'refs/heads/test', 'refs/heads/main'):
            continue
        target = ref.rsplit('/',1)[1]
        run, task = inherited(repo, 'git:' + target)
        if set(new) == {'0'} or set(old) == {'0'}:
            deny('PROTECTED_BRANCH', 'Protected branch creation/deletion is not a task operation')
        admit_candidate(repo, task, new, old)
        if target == 'dev' and (run['kind'] != 'integrate' or run['fingerprint'] != digest(json.dumps([new, old]))):
            deny('USE_INTEGRATE', 'dev requires the controller operation bound to this old/new pair')
        for name in git(repo.main, 'diff', '--name-only', old, new).splitlines():
            if secret_path(name):
                deny('PROTECTED_PATH', 'Candidate contains a sensitive/runtime path')


def install_hooks(repo):
    if repo.current != repo.main:
        deny('MAIN_WORKTREE_REQUIRED', 'Install hooks from the main worktree after integration')
    repo.identity()
    configured = git(repo.main, 'config', '--get', 'core.hooksPath', check=False)
    if configured.returncode == 0:
        deny('EXISTING_HOOKS', 'Custom hooksPath detected; integrate hooks explicitly instead of overriding it')
    hooks = repo.common / 'hooks'
    hooks.mkdir(exist_ok=True)
    commands = {'pre-commit':'hooks check', 'pre-merge-commit':'hooks check --merge', 'reference-transaction':'hooks ref "$1"'}
    for name in commands:
        p = hooks / name
        if p.exists() and '# agentctl-managed-v1' not in p.read_text():
            deny('EXISTING_HOOKS', 'Existing hook will not be overwritten', hook=name)
    import shlex
    entry = shlex.quote(str(repo.main/'scripts/agentctl.py'))
    for name, command in commands.items():
        p = hooks/name
        atomic_write(p, '#!/bin/sh\n# agentctl-managed-v1\nexec python3 '+entry+' '+command+'\n')
        p.chmod(0o755)
    emit({'status':'installed','hooks':list(commands)})


def validate(repo, args):
    task = task_row(repo, args.task)
    if repo.current != Path(task['worktree']):
        deny('WORKTREE_MISMATCH', 'Validate from the registered task worktree')
    command = args.command[1:] if args.command[:1] == ['--'] else args.command
    if not command:
        deny('INVALID_REQUEST','Validation requires an explicit command')
    # Read all source for exact-HEAD evidence, not just the task's changed paths.
    resources = ['path:' + scope for scope in json.loads(task['scopes'])] + ['git:index']
    with Lease(repo, task, resources, None, digest(json.dumps(['validate', command])), mode='S') as lease:
        clean_source(repo.current)
        sha = git(repo.current, 'rev-parse', 'HEAD')
        with repo.db(True) as db:
            db.execute('DELETE FROM validations WHERE task=? AND sha=?', (args.task, sha))
        code, orphan = supervise(repo, lease, command, repo.current)
        if code or orphan:
            lease.finish(code or 1, orphan)
            return code or 1
        clean_source(repo.current)
        if git(repo.current, 'rev-parse', 'HEAD') != sha:
            deny('BASE_CHANGED','Validation changed HEAD')
        with repo.db(True) as db:
            db.execute('INSERT OR REPLACE INTO validations VALUES(?,?,?,?)',
                       (args.task,sha,digest(json.dumps(command)),time.time()))
        lease.finish()
        emit({'status':'validated','sha':sha,'task':args.task,'run_id':lease.request_id})
    return 0


def integrate(repo, args):
    task = task_row(repo, args.task)
    candidate = git(task['worktree'], 'rev-parse', 'HEAD')
    expected = git(repo.main, 'rev-parse', args.expected_dev+'^{commit}')
    admit_candidate(repo, task, candidate, expected)
    with Lease(repo,task,['git:dev','git:index'],args.request_id,digest(json.dumps([candidate,expected])),kind='integrate') as lease:
        if lease.replay:
            emit({'status':'replayed','run_id':lease.request_id,'exit_code':lease.row['exit_code']})
            return lease.row['exit_code'] or 0
        admit_candidate(repo, task, candidate, expected)
        clean_source(repo.main)
        clean_source(Path(task['worktree']))
        if git(repo.main,'branch','--show-current') != 'dev' or git(repo.main,'rev-parse','dev') != expected:
            deny('BASE_CHANGED','Main dev moved; rebuild and validate the candidate')
        env=os.environ.copy();env.update(AGENTCTL_RUN_ID=lease.request_id,AGENTCTL_TASK_ID=args.task,
            AGENTCTL_GENERATION=str(lease.row['generation']),AGENTCTL_EPOCH=repo.identity()['epoch'])
        code, orphan = supervise(repo, lease, ['git','-C',str(repo.main),'merge','--ff-only',candidate], repo.main, env)
        lease.finish(code,bool(code) or orphan)
        emit({'status':'merged' if code==0 else 'recovery_required','sha':candidate,'run_id':lease.request_id})
        return code


def parser():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--repo',default='.')
    sub=p.add_subparsers(dest='action',required=True)
    for name in ('init','status','doctor'):sub.add_parser(name)
    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=[])
    close=task.add_parser('close');close.add_argument('--id',required=True);close.add_argument('--status',choices=['done','cancelled'],default='done')
    w=sub.add_parser('with');w.add_argument('--task',required=True);w.add_argument('--resource',action='append',required=True);w.add_argument('--shared',action='store_true');w.add_argument('--request-id');w.add_argument('--wait',type=float,default=0);w.add_argument('--cwd');w.add_argument('command',nargs=argparse.REMAINDER)
    g=sub.add_parser('guard').add_subparsers(dest='kind',required=True)
    r=g.add_parser('resource');r.add_argument('resource')
    e=g.add_parser('env');e.add_argument('--target',choices=['dev','test','main','shared','all'],required=True)
    b=g.add_parser('build');b.add_argument('--commit',required=True)
    check=sub.add_parser('worktree-check');check.add_argument('--path',required=True)
    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)
    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'):
        fr=sub.add_parser(action);fr.add_argument('--task',required=True);fr.add_argument('--resource',required=True);fr.add_argument('--reason',required=True)
    hooks=sub.add_parser('hooks').add_subparsers(dest='hook_action',required=True)
    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')
    return p


def main():
    args=parser().parse_args();repo=Repo(args.repo)
    if args.action=='init':initialize(repo)
    elif args.action in ('status','doctor'):
        identity=repo.identity()
        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')")]
        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})
    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')
            branch=git(repo.current,'branch','--show-current')
            if repo.current==repo.main or repo.current.parent!=repo.main/'.worktrees' or not branch.startswith(('feature/','fix/')):
                deny('WORKTREE_REQUIRED','Register from a direct .worktrees feature/fix checkout')
            scopes=sorted(set(relative(v) for v in args.scope))
            for scope in scopes:
                if scope != '.' and secret_path(scope):deny('PATH_OUT_OF_SCOPE','Git and runtime internals cannot be registered as source scopes')
                safe_path(repo.current,scope)
            for cap in args.capability:
                if cap not in ('docs:admin','docs:approve'):resource(cap)
            values=(args.id,args.owner,str(repo.current),branch,json.dumps(scopes),json.dumps(sorted(set(args.capability))),'active',git(repo.current,'rev-parse','HEAD'))
            with repo.os_file('worktree-' + digest(str(repo.current)) + '.lock').open('a+') as lifecycle, repo.db(True) as db:
                fcntl.flock(lifecycle,fcntl.LOCK_SH|fcntl.LOCK_NB)
                if not repo.current.exists():deny('WORKTREE_MISMATCH','Worktree disappeared during registration')
                old=db.execute('SELECT * FROM tasks WHERE id=?',(args.id,)).fetchone()
                if old:
                    if tuple(old)[:6]!=values[:6] or old['state']!='active':deny('TASK_EXISTS','Task registration differs; no implicit authority expansion')
                else:
                    if db.execute("SELECT id FROM tasks WHERE worktree=? AND state='active'", (str(repo.current),)).fetchone():
                        deny('WORKTREE_OCCUPIED', 'Another active task already owns this checkout')
                    db.execute('INSERT INTO tasks VALUES(?,?,?,?,?,?,?,?)',values)
                    event(db,args.id,{'event':'registered','worktree':str(repo.current)})
            emit({'status':'registered','task':args.id,'worktree':str(repo.current)})
        else:
            task_row(repo,args.id)
            with repo.db(True) as db:
                if db.execute("SELECT id FROM runs WHERE task=? AND state IN ('running','queued','recovery_required')",(args.id,)).fetchone():deny('LOCK_BUSY','Task still has an active or unresolved operation')
                db.execute('UPDATE tasks SET state=? WHERE id=?',(args.status,args.id))
            emit({'status':args.status,'task':args.id})
    elif args.action=='with':return execute(repo,args)
    elif args.action=='guard':guard(repo,args)
    elif args.action=='worktree-check':worktree_check(repo,args.path)
    elif args.action=='docs':
        if args.doc_action=='snapshot':docs_snapshot(repo,args)
        else:return docs_apply(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()
        if not row:deny('UNKNOWN_OPERATION','Unknown run id')
        emit(dict(row))
    elif args.action in ('freeze','unfreeze'):
        task=task_row(repo,args.task);r=resource(args.resource);check_resources(repo,task,[r])
        with repo.db(True) as db:
            if args.action=='freeze':
                for row in db.execute("SELECT resources FROM runs WHERE state IN ('running','recovery_required')"):
                    if any(overlaps(r,v) for v in json.loads(row['resources'])):deny('LOCK_BUSY','Cannot freeze a resource with an active writer')
                db.execute('INSERT INTO freezes VALUES(?,?,?)',(r,args.task,args.reason))
            else:
                row=db.execute('SELECT owner FROM freezes WHERE resource=?',(r,)).fetchone()
                if row and row['owner']!=args.task:deny('PATH_OUT_OF_SCOPE','Only the freeze owner can unfreeze')
                db.execute('DELETE FROM freezes WHERE resource=?',(r,))
            event(db,args.task,{'event':args.action,'resource':r,'reason':args.reason})
        emit({'status':args.action,'resource':r})
    elif args.action=='hooks':
        if args.hook_action=='install':install_hooks(repo)
        elif args.hook_action=='check':hook_check(repo,args.merge)
        else:hook_ref(repo,args.state)
    elif args.action=='validate':return validate(repo,args)
    elif args.action=='integrate':return integrate(repo,args)
    return 0


if __name__=='__main__':
    try:
        sys.exit(main() or 0)
    except Denied as exc:
        emit({'status':'denied','code':exc.code,'message':exc.message,**exc.details},sys.stderr)
        sys.exit(75 if exc.code=='LOCK_BUSY' else 1)
    except (OSError,sqlite3.Error,ValueError,KeyError) as exc:
        emit({'status':'denied','code':'RECOVERY_REQUIRED','message':str(exc)},sys.stderr)
        sys.exit(1)