weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
3f63753b6337bd12b155c6f73bf5b0c02797ffe5
Agent-System/scripts/agentctl.py
T
weichen f9f055bee1 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.
2026-09-24 14:13:27 +00:00

3391 lines
158 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Local, cooperative agent coordination. OS isolation is intentionally out of scope."""
import argparse
import contextlib
import errno
import fcntl
import hashlib
import json
import os
from dataclasses import dataclass
from pathlib import Path
import re
import shutil
import signal
import sqlite3
import subprocess
import sys
import tempfile
import time
from typing import Literal, Optional
import uuid
# Pure W4 helpers (deploy/projection/PI checks) live beside this file for tests.
sys.path.insert(0, str(Path(__file__).resolve().parent))
# Governance tools must not leave __pycache__ that dirties clean Git trees.
sys.dont_write_bytecode = True
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'
DEFAULT_POLICY = {
'protocol_version': 1,
'min_protocol_version': 1,
'policy_revision': 'agent-governance-2026-09-21-r1',
'shared_docs': ['.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'],
'projectable_docs': ['.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'],
'projectable_prefixes': ['.doc/ops/', '.doc/status/', '.doc/remediation/'],
'job_kinds': [
'validate', 'integrate', 'with', 'deploy_prepare', 'deploy_run',
'deploy_accept', 'docs_project', 'cleanup_inspect', 'noop',
],
'doc_rules': {
'.doc/ops/CURRENT.md': {
'own_row_only': True,
'slot_protected': True,
'header': '| owner | task |',
},
'.doc/status/FEATURE-MATRIX.md': {
'evidence_for_done': True,
'header': '| record_id | 模块 |',
},
},
'reserved_record_ids': ['holder', 'idle', 'Cursor', 'Codex', '-'],
'protected_sections': {
'.doc/ops/CURRENT.md': [
'## 集成槽位',
'## 焦点',
'## 清理接管与遗留现场',
'## 受控约束',
'## 活跃入口',
],
},
'heartbeat_ttl_seconds': 300,
'suspect_ttl_seconds': 600,
# Optional catalog keys (not in required set-equality checks):
'catalog_gate': {
'hard_kinds': ['same_mutex', 'same_feature', 'same_scope', 'same_req_id'],
'semantic_top_k': 10,
'bm25_threshold': -8.0,
'ack_required': True,
},
'catalog_vector': {
'enabled': False,
'timeout_seconds': 8,
'fail_open': True,
'cosine_threshold': 0.82,
},
}
PROJECTION_STATUS_VOCAB = frozenset({'planned', 'in_progress', 'blocked', 'done', 'cancelled'})
PROJECTION_STATUS_FROM_MACHINE = {'queued': 'in_progress', 'idle': 'planned'}
SHARED_DOCS = set(DEFAULT_POLICY['shared_docs'])
RESERVED_RECORD_IDS = set(DEFAULT_POLICY['reserved_record_ids'])
DOC_RULES = dict(DEFAULT_POLICY['doc_rules'])
PROTECTED_SECTIONS = {k: list(v) for k, v in DEFAULT_POLICY['protected_sections'].items()}
HEARTBEAT_TTL_SECONDS = DEFAULT_POLICY['heartbeat_ttl_seconds']
SUSPECT_TTL_SECONDS = DEFAULT_POLICY['suspect_ttl_seconds']
ACTIVE = ('running', 'recovery_required', 'suspect', 'quarantine')
UNRESOLVED = ('running', 'recovery_required', 'queued', 'suspect', 'quarantine')
STATUS_RUNS = ('running', 'queued', 'recovery_required', 'busy', 'suspect', 'quarantine')
TERMINAL = ('completed', 'failed', 'recovered')
EXTERNAL_VERIFY_MARKER = 'verified:'
RUN_QUARANTINE_COLUMNS = (
('fence', 'INTEGER NOT NULL DEFAULT 0'),
('suspect_since', 'REAL'),
('quarantined_at', 'REAL'),
('quarantine_reason', 'TEXT'),
('verified_by', 'TEXT'),
)
QUEUE_TERMINAL = ('completed', 'failed', 'cancelled')
GIT_CONTEXT = ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE', 'GIT_PREFIX')
QUEUE_SCHEMA = '''
CREATE TABLE IF NOT EXISTS queue_items(
id TEXT PRIMARY KEY,
task TEXT NOT NULL,
head_sha TEXT NOT NULL,
expected_dev TEXT NOT NULL,
priority INTEGER NOT NULL DEFAULT 0,
state TEXT NOT NULL,
created REAL NOT NULL,
updated REAL NOT NULL,
claim_run TEXT,
reason TEXT,
UNIQUE(task, head_sha)
);
CREATE INDEX IF NOT EXISTS queue_pending ON queue_items(state, priority DESC, created);
'''
CLAIM_SCHEMA = '''
CREATE TABLE IF NOT EXISTS claims(
id TEXT PRIMARY KEY,
resource TEXT NOT NULL,
mode TEXT NOT NULL,
task TEXT NOT NULL,
owner TEXT NOT NULL,
request_id TEXT,
state TEXT NOT NULL,
reason TEXT,
created REAL NOT NULL,
updated REAL NOT NULL,
UNIQUE(resource, task)
);
CREATE INDEX IF NOT EXISTS claims_active ON claims(state, resource);
CREATE TABLE IF NOT EXISTS claim_queue(
id TEXT PRIMARY KEY,
resource TEXT NOT NULL,
task TEXT NOT NULL,
request_id TEXT,
state TEXT NOT NULL,
created REAL NOT NULL
);
'''
Lookup = Literal['ok', 'gone', 'unreadable']
AliveStatus = Literal['alive', 'dead', 'unknown']
DescentStatus = Literal['yes', 'no', 'unknown']
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)
catalog.bind_deny(deny)
def emit(value, stream=sys.stdout):
print(json.dumps(value, ensure_ascii=False, sort_keys=True), file=stream, flush=True)
def emit_payload(payload, fmt='json', renderer=None, stream=sys.stdout):
"""Emit JSON (default) or a human table via an optional renderer callable."""
if (fmt or 'json') == 'table' and renderer is not None:
text = renderer(payload)
if not text.endswith('\n'):
text += '\n'
stream.write(text)
stream.flush()
return
emit(payload, stream)
def digest(value):
return hashlib.sha256(value.encode() if isinstance(value, str) else value).hexdigest()
def _policy_path(main: Path) -> Path:
return Path(main) / POLICY_REL
def _validate_policy_shape(data) -> dict:
if not isinstance(data, dict):
deny('POLICY_INVALID', 'Governance policy must be a JSON object')
required = (
'protocol_version', 'min_protocol_version', 'policy_revision',
'shared_docs', 'projectable_docs', 'projectable_prefixes', 'job_kinds',
'doc_rules', 'reserved_record_ids', 'protected_sections',
)
missing = [key for key in required if key not in data]
if missing:
deny('POLICY_INVALID', 'Governance policy missing required keys', keys=missing)
if not isinstance(data['protocol_version'], int) or not isinstance(data['min_protocol_version'], int):
deny('POLICY_INVALID', 'protocol_version and min_protocol_version must be integers')
if data['protocol_version'] < data['min_protocol_version']:
deny('POLICY_MISMATCH', 'protocol_version below min_protocol_version',
protocol_version=data['protocol_version'],
min_protocol_version=data['min_protocol_version'])
for key in ('shared_docs', 'projectable_docs', 'projectable_prefixes',
'job_kinds', 'reserved_record_ids'):
if not isinstance(data[key], list) or not all(isinstance(item, str) for item in data[key]):
deny('POLICY_INVALID', f'{key} must be a list of strings')
if not isinstance(data['doc_rules'], dict) or not isinstance(data['protected_sections'], dict):
deny('POLICY_INVALID', 'doc_rules and protected_sections must be objects')
if not isinstance(data['policy_revision'], str) or not data['policy_revision']:
deny('POLICY_INVALID', 'policy_revision must be a nonempty string')
# Structure must stay compatible with the builtin DEFAULT_POLICY contract.
if set(data['shared_docs']) != set(DEFAULT_POLICY['shared_docs']):
deny('POLICY_MISMATCH', 'shared_docs diverges from DEFAULT_POLICY contract')
if set(data['job_kinds']) != set(DEFAULT_POLICY['job_kinds']):
deny('POLICY_MISMATCH', 'job_kinds diverges from DEFAULT_POLICY contract')
if set(data['projectable_docs']) != set(DEFAULT_POLICY['projectable_docs']):
deny('POLICY_MISMATCH', 'projectable_docs diverges from DEFAULT_POLICY contract')
if list(data['projectable_prefixes']) != list(DEFAULT_POLICY['projectable_prefixes']):
deny('POLICY_MISMATCH', 'projectable_prefixes diverges from DEFAULT_POLICY contract')
# Optional TTL keys: validate when present; missing → DEFAULT_POLICY at apply time.
for key in ('heartbeat_ttl_seconds', 'suspect_ttl_seconds'):
if key in data and (not isinstance(data[key], int) or data[key] <= 0):
deny('POLICY_INVALID', f'{key} must be a positive integer')
# Optional catalog keys: shape checked in catalog.apply_policy; reject non-objects early.
for key in ('catalog_gate', 'catalog_vector'):
if key in data and data[key] is not None and not isinstance(data[key], dict):
deny('POLICY_INVALID', f'{key} must be an object')
for filename, rules in data['doc_rules'].items():
if not isinstance(rules, dict):
deny('POLICY_INVALID', 'doc_rules entries must be objects', file=filename)
if rules.get('slot_protected'):
sections = data['protected_sections'].get(filename)
if not isinstance(sections, list) or not sections:
deny('POLICY_INVALID',
'slot_protected requires a nonempty protected_sections entry',
file=filename)
if not all(isinstance(item, str) and item for item in sections):
deny('POLICY_INVALID', 'protected_sections values must be nonempty strings',
file=filename)
return data
def apply_policy(policy: dict) -> dict:
"""Install a validated policy into module globals used by write paths."""
global SHARED_DOCS, RESERVED_RECORD_IDS, DOC_RULES, PROTECTED_SECTIONS
global HEARTBEAT_TTL_SECONDS, SUSPECT_TTL_SECONDS
SHARED_DOCS = set(policy['shared_docs'])
RESERVED_RECORD_IDS = set(policy['reserved_record_ids'])
DOC_RULES = {key: dict(value) for key, value in policy['doc_rules'].items()}
PROTECTED_SECTIONS = {key: list(value) for key, value in policy['protected_sections'].items()}
HEARTBEAT_TTL_SECONDS = int(policy.get('heartbeat_ttl_seconds', DEFAULT_POLICY['heartbeat_ttl_seconds']))
SUSPECT_TTL_SECONDS = int(policy.get('suspect_ttl_seconds', DEFAULT_POLICY['suspect_ttl_seconds']))
w4.apply_policy(policy)
catalog.apply_policy(policy)
return policy
def load_policy(main: Path) -> dict:
"""Load governance policy from the main worktree. Fail-closed; never creates the file."""
path = _policy_path(main)
if not path.is_file():
deny('POLICY_MISSING', f'Governance policy missing at {POLICY_REL}', path=str(path))
try:
raw = path.read_text(encoding='utf-8')
data = json.loads(raw)
except (OSError, UnicodeError) as exc:
deny('POLICY_INVALID', f'Cannot read governance policy: {exc}', path=str(path))
except json.JSONDecodeError as exc:
deny('POLICY_INVALID', f'Governance policy is not valid JSON: {exc}', path=str(path))
policy = _validate_policy_shape(data)
return apply_policy(policy)
def policy_status(main: Path) -> dict:
"""Read-only policy probe for status/doctor. Never creates files."""
path = _policy_path(main)
if not path.is_file():
return {'status': 'missing', 'code': 'POLICY_MISSING', 'path': str(path)}
try:
data = json.loads(path.read_text(encoding='utf-8'))
policy = _validate_policy_shape(data)
apply_policy(policy)
return {
'status': 'ok',
'path': str(path),
'policy_revision': policy['policy_revision'],
'protocol_version': policy['protocol_version'],
}
except Denied as exc:
return {'status': 'denied', 'code': exc.code, 'message': exc.message, 'path': str(path),
**exc.details}
except (OSError, UnicodeError) as exc:
return {'status': 'denied', 'code': 'POLICY_INVALID', 'message': str(exc), 'path': str(path)}
def require_policy(repo) -> dict:
return load_policy(repo.main)
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
@dataclass(frozen=True)
class ProcInfo:
"""Process identity; start is an opaque token comparable within the same boot."""
pid: int
ppid: int
pgid: int
state: str
start: str
# ps columns: pid ppid pgid lstart state — lstart has spaces, state is last token.
_PS_LINE = re.compile(r'\s*(\d+)\s+(\d+)\s+(\d+)\s+(.+?)\s+(\S+)\s*$')
def _normalize_state(state):
return (state or '?')[0]
def _linux_read_proc(pid):
path = Path('/proc', str(pid), 'stat')
try:
raw = path.read_text().rsplit(')', 1)[1].split()
return 'ok', ProcInfo(pid=pid, ppid=int(raw[1]), pgid=int(raw[2]),
state=_normalize_state(raw[0]), start=raw[19])
except FileNotFoundError:
return 'gone', None
except (PermissionError, IsADirectoryError):
return 'unreadable', None
except OSError as exc:
if getattr(exc, 'errno', None) in (errno.ENOENT, errno.ESRCH):
return 'gone', None
return 'unreadable', None
except (ValueError, IndexError):
return 'unreadable', None
def _parse_ps_line(line):
match = _PS_LINE.match(line)
if not match:
return None
pid, ppid, pgid, start, state = match.groups()
return ProcInfo(pid=int(pid), ppid=int(ppid), pgid=int(pgid),
state=_normalize_state(state), start=start.strip())
def _darwin_read_proc(pid):
try:
result = subprocess.run(
['ps', '-o', 'pid=,ppid=,pgid=,lstart=,state=', '-p', str(pid)],
capture_output=True, text=True)
except OSError:
return 'unreadable', None
# BSD ps: nonexistent pid → nonzero + empty stdout. Any other opacity → unreadable.
if result.returncode != 0 and not result.stdout.strip():
# Errno-style chatter (e.g. "process id too large") still means no such process.
return 'gone', None
if not result.stdout.strip():
return 'unreadable', None
info = _parse_ps_line(result.stdout.splitlines()[0])
if not info or info.pid != int(pid):
return 'unreadable', None
return 'ok', info
def boot_epoch():
"""Opaque boot generation token; DB field `boot` stays compatible."""
if sys.platform == 'darwin':
try:
raw = subprocess.check_output(['sysctl', '-n', 'kern.boottime'], text=True)
except (OSError, subprocess.CalledProcessError) as exc:
deny('PROCESS_UNREADABLE', f'Unable to read kern.boottime: {exc}')
match = re.search(r'sec\s*=\s*(\d+)', raw)
if not match:
deny('PROCESS_UNREADABLE', 'Unable to parse kern.boottime')
return match.group(1)
if sys.platform.startswith('linux'):
try:
return Path('/proc/sys/kernel/random/boot_id').read_text().strip()
except OSError as exc:
deny('PROCESS_UNREADABLE', f'Unable to read boot_id: {exc}')
deny('UNSUPPORTED_PLATFORM', f'Boot epoch unsupported on {sys.platform}')
def read_proc(pid) -> tuple[Lookup, Optional[ProcInfo]]:
"""Lookup a single process. unreadable must never be treated as exited."""
if sys.platform == 'darwin':
return _darwin_read_proc(pid)
if sys.platform.startswith('linux'):
return _linux_read_proc(pid)
deny('UNSUPPORTED_PLATFORM', f'Process inspection unsupported on {sys.platform}')
def _enumerate_group(pgid):
"""Yield ProcInfo in pgid. Returns (saw_member_alive, enum_status).
enum_status is 'ok' when every observable entry was readable, else 'unknown'.
Fail-closed: enum failure or any unreadable member never reports 'all dead'.
"""
if sys.platform == 'darwin':
try:
result = subprocess.run(
['ps', '-axo', 'pid=,ppid=,pgid=,lstart=,state='],
capture_output=True, text=True)
except OSError:
return False, 'unknown'
if result.returncode != 0:
return False, 'unknown'
alive_member = False
for line in result.stdout.splitlines():
if not line.strip():
continue
info = _parse_ps_line(line)
if info is None:
return alive_member, 'unknown'
if info.pgid == pgid and info.state != 'Z':
alive_member = True
return alive_member, 'ok'
if sys.platform.startswith('linux'):
try:
items = list(Path('/proc').iterdir())
except OSError:
return False, 'unknown'
alive_member, saw_unreadable = False, False
for item in items:
if not item.name.isdigit():
continue
lookup, info = _linux_read_proc(int(item.name))
if lookup == 'unreadable':
saw_unreadable = True
continue
if lookup == 'ok' and info and info.pgid == pgid and info.state != 'Z':
alive_member = True
return alive_member, ('unknown' if saw_unreadable else 'ok')
deny('UNSUPPORTED_PLATFORM', f'Process enumeration unsupported on {sys.platform}')
def alive_status(pid, start, boot_id) -> AliveStatus:
if boot_id != boot_epoch():
return 'dead'
lookup, info = read_proc(pid)
if lookup == 'gone':
return 'dead'
if lookup == 'unreadable':
return 'unknown'
if not info or info.state == 'Z' or info.start != start:
return 'dead'
return 'alive'
def group_status(pgid) -> AliveStatus:
if not pgid:
return 'dead'
alive_member, enum_status = _enumerate_group(pgid)
if alive_member:
return 'alive'
if enum_status != 'ok':
return 'unknown'
return 'dead'
def descent_status(ancestor) -> DescentStatus:
current = os.getpid()
seen = set()
while current > 1 and current not in seen:
if current == ancestor:
return 'yes'
seen.add(current)
lookup, info = read_proc(current)
if lookup == 'unreadable':
return 'unknown'
if lookup == 'gone' or not info:
return 'no'
current = info.ppid
return 'no'
def boot():
return boot_epoch()
def proc(pid):
"""Compat: ok→dict(start/ppid/group/state); gone→None; unreadable→deny (fail-closed)."""
lookup, info = read_proc(pid)
if lookup == 'gone':
return None
if lookup != 'ok' or not info:
deny('PROCESS_UNREADABLE', f'Unable to inspect process {pid}')
return {'pid': info.pid, 'ppid': info.ppid, 'group': info.pgid, 'state': info.state, 'start': info.start}
def alive(pid, start, boot_id):
"""True when the process may still hold resources (alive or unknown)."""
return alive_status(pid, start, boot_id) != 'dead'
def group_alive(group):
"""True when the group may still hold resources (alive or unknown)."""
return group_status(group) != 'dead'
def descendants_of(pid):
return descent_status(pid) == 'yes'
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.startswith('claim:'):
return 'claim:' + relative(value[6:])
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:])
if a.startswith('claim:') and b.startswith('claim:'):
return covers(a[6:], b[6:]) or covers(b[6:], a[6:])
# 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')
# DDL inside ensure_schema implicitly commits any open txn; take the
# exclusive lock only after schema is stable so claim/lease CAS holds.
db.execute('BEGIN IMMEDIATE')
ensure_schema(db)
if db.in_transaction:
db.execute('COMMIT')
db.execute('BEGIN IMMEDIATE')
yield db
if write and db.in_transaction:
db.commit()
except BaseException:
if write and db.in_transaction:
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(True) as db:
db.execute('SELECT count(*) FROM runs').fetchone()
db.execute('SELECT count(*) FROM queue_items').fetchone()
codegraph = codegraph_ensure_fail_open(repo.main)
emit({'status': 'initialized', 'lock_root': str(repo.root),
'codegraph': codegraph, **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);
''' + QUEUE_SCHEMA)
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))
codegraph = codegraph_ensure_fail_open(repo.main)
emit({'status': 'initialized', 'lock_root': str(repo.root),
'codegraph': codegraph, **identity})
def ensure_schema(db):
"""Additive tables for durable queue + W4 deploy/projection + claims; write txn only."""
for statement in filter(None, (part.strip() for part in QUEUE_SCHEMA.split(';'))):
db.execute(statement)
for statement in filter(None, (part.strip() for part in CLAIM_SCHEMA.split(';'))):
db.execute(statement)
existing = {row[1] for row in db.execute('PRAGMA table_info(runs)')}
for name, decl in RUN_QUARANTINE_COLUMNS:
if name not in existing:
db.execute(f'ALTER TABLE runs ADD COLUMN {name} {decl}')
w4.ensure_schema(db)
catalog.ensure_schema(db)
def codegraph_is_ignored(path):
"""True when git will not track .codegraph under path (safe to init)."""
target = Path(path).resolve()
try:
# Pattern `.codegraph/` only matches directories; probe with trailing slash.
completed = subprocess.run(
['git', '-C', str(target), 'check-ignore', '-q', '--', '.codegraph/'],
capture_output=True, check=False)
return completed.returncode == 0
except OSError:
return False
def codegraph_binary():
return shutil.which('codegraph')
def codegraph_sync_fail_open(path=None, *, timeout=60):
"""Run `codegraph sync -q`. Always returns; never raises into git hooks."""
binary = codegraph_binary()
target = str(Path(path).resolve()) if path else None
if not binary:
return {'status': 'not_installed', 'exit_code': 0}
cmd = [binary, 'sync', '-q']
if target:
cmd.append(target)
try:
completed = subprocess.run(
cmd, capture_output=True, text=True, timeout=timeout, check=False)
return {
'status': 'ok' if completed.returncode == 0 else 'sync_failed',
'exit_code': completed.returncode,
'stderr': (completed.stderr or '')[:500],
}
except (OSError, subprocess.TimeoutExpired) as exc:
return {'status': 'sync_error', 'exit_code': 0, 'error': str(exc)}
def codegraph_ensure_fail_open(path, *, timeout=300):
"""Init if missing, else sync. Fail-open for agentctl init / session start.
Skips creating a new index when `.codegraph/` is not gitignored, so init
cannot dirty a clean source tree (test fixtures / partial checkouts).
"""
binary = codegraph_binary()
target = Path(path).resolve()
if not binary:
return {'status': 'not_installed'}
db = target / '.codegraph' / 'codegraph.db'
if db.exists():
result = codegraph_sync_fail_open(target, timeout=min(timeout, 60))
result['action'] = 'sync'
return result
if not codegraph_is_ignored(target):
return {'status': 'skipped', 'action': 'init',
'reason': '.codegraph not gitignored; refusing to dirty worktree'}
try:
completed = subprocess.run(
[binary, 'init', '-y', str(target)],
capture_output=True, text=True, timeout=timeout, check=False)
return {
'status': 'ok' if completed.returncode == 0 else 'init_failed',
'action': 'init',
'exit_code': completed.returncode,
'stderr': (completed.stderr or '')[:500],
}
except (OSError, subprocess.TimeoutExpired) as exc:
return {'status': 'init_error', 'action': 'init', 'error': str(exc)}
def record_audit(repo, kind, payload):
"""Append a read-only audit ledger entry. Never mutates shared docs."""
with repo.db(True) as db:
ensure_schema(db)
db.execute(
'INSERT INTO audit_ledger(kind, created, payload) VALUES(?,?,?)',
(kind, time.time(), json.dumps(payload, ensure_ascii=False, sort_keys=True)))
return payload
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('claim:'):
p = value[6:]
if p != '.' and secret_path(p):
deny('PATH_OUT_OF_SCOPE', 'Runtime and Git administration paths are not claimable')
if not any(covers(s, p) for s in scopes):
deny('PATH_OUT_OF_SCOPE', 'Claim 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):
lookup, info = read_proc(os.getpid())
if lookup != 'ok' or not info:
deny('PROCESS_UNREADABLE', 'Cannot observe supervisor process', lookup=lookup)
with self.repo.db(True) as db:
ensure_schema(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_epoch(), 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_epoch(), 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','suspect','quarantine') ORDER BY generation",
(self.request_id,)).fetchall()
for other in candidates:
if other['state'] == 'queued':
# Only reclaim when proven dead; unknown keeps the slot.
if alive_status(other['pid'], other['start'], other['boot']) == 'dead':
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':
# unknown/alive both require recovery; only proven-dead releases cleanly.
remaining = group_status(state['pgid']) != 'dead'
self.finish(1, bool(doc and doc['state'] == 'prepared') or remaining)
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')
# Fail-closed: unknown supervisor or uncertain descent denies the lease identity.
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 alive_status(row['pid'], row['start'], row['boot']) != 'alive'
or descent_status(row['pid']) != 'yes'):
deny('STALE_GENERATION', 'Lease identity is stale or caller is not a supervised descendant')
# Unset env equates to fence 0 so pre-upgrade supervisors / self-bootstrap
# ff hooks (which re-exec the updated agentctl.py mid-merge) stay compatible.
expected_fence = str(row['fence'] if 'fence' in row.keys() else 0)
observed_fence = os.getenv('AGENTCTL_FENCE', '0')
if observed_fence != expected_fence:
deny('STALE_GENERATION', 'Lease fence is stale or missing', expected=expected_fence,
observed=os.getenv('AGENTCTL_FENCE'))
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)
lookup, info = read_proc(child.pid)
child_start = info.start if lookup == 'ok' and info else ''
with repo.db(True) as db:
db.execute('UPDATE runs SET child_pid=?,child_start=?,pgid=? WHERE id=?',
(child.pid, child_start, 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()
# remaining=True when alive or unknown — never treat opaque enum as all-dead.
return (code if code >= 0 else 128 - code), group_status(child.pid) != 'dead'
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 --')
# Soft-overlap acks must be clear before exclusive writes (shared reads exempt).
if not args.shared:
with repo.db() as db:
catalog.assert_acks_clear(db, args.task)
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'],
AGENTCTL_FENCE=str(lease.row.get('fence') or 0))
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 not cells:
continue
first = cells[0].casefold()
# record_id/id first → stable key in cells[0]; legacy 模块/module/feature → key is module name.
if first not in ('record_id', 'id', '模块', '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 matrix_module_states(content):
"""Map module display name → status for evidence gates across record_id migrations."""
states, status_index, module_index, width = {}, None, 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 not cells:
continue
first = cells[0].casefold()
if first not in ('record_id', 'id', '模块', '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')
module_cols = [i for i, c in enumerate(labels) if c in ('模块', 'module', 'feature')]
if first in ('record_id', 'id'):
if len(module_cols) != 1:
deny('INVALID_DOCUMENT', 'Matrix with record_id needs one module column')
module_index = module_cols[0]
else:
module_index = 0
status_index, width = columns[0], len(cells)
continue
if all(re.fullmatch(r':?-+:?', cell.replace(' ', '')) for cell in cells):
continue
if len(cells) != width:
deny('INVALID_DOCUMENT', 'Matrix records must match its columns')
module = cells[module_index]
if module in states:
deny('INVALID_DOCUMENT', 'Matrix module names must be unique')
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[module] = 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 (DOC_RULES.get(filename, {}).get('own_row_only') 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 RESERVED_RECORD_IDS:
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')
for section in PROTECTED_SECTIONS.get(filename, []):
def slot(text, marker=section):
match = re.search(r'(?m)^' + re.escape(marker) + r'\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.
# Compare by module display name so additive record_id migration does not
# look like newly promoted done rows (keys would otherwise diverge).
if DOC_RULES.get(filename, {}).get('evidence_for_done') or filename.endswith('FEATURE-MATRIX.md'):
old_states = matrix_module_states(before[filename])
new_states = matrix_module_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'):
if cells[0] != record_id:
deny('INVALID_REQUEST', 'FEATURE-MATRIX replacement must keep record_id in the first column')
if '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 = DOC_RULES.get(filename, {}).get('header')
if not header:
header = '| owner | task |' if filename.endswith('CURRENT.md') else '| record_id | 模块 |'
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','suspect','quarantine')",
(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 _external_side_effect_resources(resources):
return [r for r in resources if r.startswith('env:') or r == 'git:dev']
def _require_external_verification(resources, reason):
external = _external_side_effect_resources(resources)
if external and EXTERNAL_VERIFY_MARKER not in (reason or ''):
deny('EVIDENCE_REQUIRED',
'env:* or git:dev recovery requires an explicit verified: marker in --reason',
resources=external)
return external
def _acquire_physical_locks(repo, resources):
handles = []
try:
for name in physical_names(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')
return handles
except BaseException:
for handle in reversed(handles):
handle.close()
raise
def _process_recovery_gates(row):
# Deny unless supervisor and (same-boot) child group are proven dead.
if alive_status(row['pid'], row['start'], row['boot']) != 'dead':
deny('LOCK_BUSY', 'Supervisor or child process group is still alive; no takeover')
if row['boot'] == boot_epoch() and group_status(row['pgid']) != 'dead':
deny('LOCK_BUSY', 'Supervisor or child process group is still alive; no takeover')
def _apply_prepared_document(repo, row, document):
if not document or document['state'] != 'prepared':
return
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:
ensure_schema(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=?", (row['id'],))
def recover(repo, args):
with repo.db(True) as db:
ensure_schema(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
resources = json.loads(row['resources'])
external = _require_external_verification(resources, args.reason)
_process_recovery_gates(row)
handles = _acquire_physical_locks(repo, resources)
try:
_apply_prepared_document(repo, row, document)
with repo.db(True) as db:
ensure_schema(db)
db.execute(
"UPDATE runs SET state='recovered',reason=?,exit_code=?,fence=COALESCE(fence,0)+1 WHERE id=?",
(args.reason, 0 if document else 1, args.run))
event(db, args.run, {
'event': 'operator_recovery',
'reason': args.reason,
'external_side_effects': bool(external),
'resources': external,
})
emit({'status': 'recovered', 'run_id': args.run})
finally:
for handle in reversed(handles):
handle.close()
def quarantine_sweep(repo):
"""Demote stale/unobservable runs to suspect/quarantine. Never auto-releases."""
now = time.time()
with repo.db(True) as db:
ensure_schema(db)
rows = [dict(r) for r in db.execute(
"SELECT * FROM runs WHERE state IN ('running','recovery_required','suspect')").fetchall()]
# Observe processes outside the write transaction (no flock/git/subprocess in txn).
decisions = []
for row in rows:
alive = alive_status(row['pid'], row['start'], row['boot'])
group = group_status(row['pgid'])
heartbeat = float(row['heartbeat'] or 0)
age = now - heartbeat
state = row['state']
if state in ('running', 'recovery_required'):
if age > HEARTBEAT_TTL_SECONDS and alive != 'alive':
decisions.append({
'id': row['id'], 'from': state, 'to': 'suspect',
'suspect_since': now, 'alive': alive, 'group': group,
'reason': 'heartbeat_ttl',
})
continue
if state == 'suspect':
since = float(row['suspect_since'] or row['heartbeat'] or 0)
if now - since <= SUSPECT_TTL_SECONDS:
continue
# Fail-closed: unknown supervisor or non-dead group stays at suspect.
if alive == 'unknown' or group != 'dead':
continue
decisions.append({
'id': row['id'], 'from': state, 'to': 'quarantine',
'quarantined_at': now, 'alive': alive, 'group': group,
'reason': 'suspect_ttl',
})
applied = []
with repo.db(True) as db:
ensure_schema(db)
for item in decisions:
current = db.execute('SELECT state FROM runs WHERE id=?', (item['id'],)).fetchone()
if not current or current['state'] != item['from']:
continue
if item['to'] == 'suspect':
db.execute(
"UPDATE runs SET state='suspect',suspect_since=?,quarantine_reason=? WHERE id=?",
(item['suspect_since'], item['reason'], item['id']))
else:
db.execute(
"UPDATE runs SET state='quarantine',quarantined_at=?,quarantine_reason=? WHERE id=?",
(item['quarantined_at'], item['reason'], item['id']))
event(db, item['id'], {
'event': 'quarantine_demote', 'from': item['from'], 'to': item['to'],
'reason': item['reason'], 'alive_status': item['alive'], 'group_status': item['group'],
})
applied.append({'run_id': item['id'], 'from': item['from'], 'to': item['to']})
emit({'status': 'ok', 'demoted': applied, 'heartbeat_ttl_seconds': HEARTBEAT_TTL_SECONDS,
'suspect_ttl_seconds': SUSPECT_TTL_SECONDS})
return 0
def quarantine_list(repo):
with repo.db(True) as db:
ensure_schema(db)
rows = [dict(r) for r in db.execute(
"SELECT * FROM runs WHERE state IN ('suspect','quarantine') ORDER BY generation").fetchall()]
emit({'status': 'ok', 'runs': rows})
return 0
def quarantine_release(repo, args):
if not args.verified_by or not str(args.verified_by).strip():
deny('INVALID_REQUEST', 'quarantine release requires --verified-by')
with repo.db(True) as db:
ensure_schema(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 0
if row['state'] not in ('suspect', 'quarantine', 'recovery_required', 'running', 'queued', 'busy'):
deny('INVALID_REQUEST', 'Run is not in a releasable unresolved state', state=row['state'])
resources = json.loads(row['resources'])
external = _require_external_verification(resources, args.reason)
_process_recovery_gates(row)
handles = _acquire_physical_locks(repo, resources)
try:
_apply_prepared_document(repo, row, document)
with repo.db(True) as db:
ensure_schema(db)
db.execute(
"UPDATE runs SET state='recovered',reason=?,exit_code=?,fence=COALESCE(fence,0)+1,"
"verified_by=? WHERE id=?",
(args.reason, 0 if document else 1, args.verified_by, args.run))
event(db, args.run, {
'event': 'quarantine_release',
'reason': args.reason,
'verified_by': args.verified_by,
'external_side_effects': bool(external),
'resources': external,
})
emit({'status': 'recovered', 'run_id': args.run, 'verified_by': args.verified_by})
finally:
for handle in reversed(handles):
handle.close()
return 0
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)
# post-merge / post-checkout only sync codegraph (fail-open); they must never
# write tracked files or deny integrate's ff-only merge.
commands = {
'pre-commit': 'hooks check',
'pre-merge-commit': 'hooks check --merge',
'reference-transaction': 'hooks ref "$1"',
'post-merge': 'hooks codegraph-sync',
'post-checkout': 'hooks codegraph-sync',
}
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 hooks_codegraph_sync(repo):
"""Git post-merge / post-checkout entry: sync index, always exit 0."""
result = codegraph_sync_fail_open(repo.current)
emit({'status': 'ok', 'hook': 'codegraph-sync', 'codegraph': result})
return 0
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'],
AGENTCTL_FENCE=str(lease.row.get('fence') or 0))
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 queue_row(row):
return dict(row) if row is not None else None
def queue_enqueue(repo, args):
task = task_row(repo, args.task)
if repo.current != Path(task['worktree']):
deny('WORKTREE_MISMATCH', 'Enqueue from the registered task worktree')
head = git(task['worktree'], 'rev-parse', 'HEAD')
if args.head and git(task['worktree'], 'rev-parse', args.head + '^{commit}') != head:
deny('BASE_CHANGED', 'Requested head is not the owning task HEAD')
expected = git(repo.main, 'rev-parse', args.expected_dev + '^{commit}')
admit_candidate(repo, task, head, expected)
item_id = args.request_id or str(uuid.uuid4())
if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', item_id):
deny('INVALID_REQUEST', 'Invalid queue request id')
now = time.time()
with repo.db(True) as db:
existing = db.execute(
"SELECT * FROM queue_items WHERE task=? AND head_sha=? AND state NOT IN ('cancelled')",
(args.task, head)).fetchone()
if existing:
if existing['state'] in ('pending', 'claimed') and existing['expected_dev'] == expected:
emit({'status': 'duplicate', 'item': queue_row(existing)})
return 0
deny('QUEUE_CONFLICT', 'A non-cancelled queue item already covers this task head',
item=queue_row(existing))
if db.execute('SELECT id FROM queue_items WHERE id=?', (item_id,)).fetchone():
deny('REQUEST_ID_REUSED', 'Queue request id already exists')
db.execute(
'''INSERT INTO queue_items(id,task,head_sha,expected_dev,priority,state,created,updated,claim_run,reason)
VALUES(?,?,?,?,?,'pending',?,?,NULL,NULL)''',
(item_id, args.task, head, expected, max(0, int(args.priority)), now, now))
event(db, item_id, {'event': 'queue_enqueued', 'task': args.task, 'head_sha': head})
row = db.execute('SELECT * FROM queue_items WHERE id=?', (item_id,)).fetchone()
emit({'status': 'enqueued', 'item': queue_row(row)})
return 0
def queue_list(repo, args):
states = None if args.all else ('pending', 'claimed')
with repo.db(True) as db:
if states:
rows = db.execute(
"SELECT * FROM queue_items WHERE state IN (?,?) ORDER BY priority DESC, created ASC",
states).fetchall()
else:
rows = db.execute('SELECT * FROM queue_items ORDER BY created ASC').fetchall()
payload = {'status': 'ok', 'items': [queue_row(r) for r in rows]}
emit_payload(payload, getattr(args, 'format', 'json'), visibility.render_queue_table)
return 0
def queue_cancel(repo, args):
task_row(repo, args.task)
with repo.db(True) as db:
row = db.execute('SELECT * FROM queue_items WHERE id=?', (args.id,)).fetchone()
if not row:
deny('UNKNOWN_QUEUE_ITEM', 'Unknown queue item id')
if row['task'] != args.task:
deny('PATH_OUT_OF_SCOPE', 'Queue item belongs to another task')
if row['state'] in QUEUE_TERMINAL:
emit({'status': 'replayed', 'item': queue_row(row)})
return 0
if row['state'] == 'claimed':
deny('LOCK_BUSY', 'Queue item is claimed by an in-flight schedule; recover or wait')
db.execute("UPDATE queue_items SET state='cancelled',reason=?,updated=? WHERE id=?",
(args.reason, time.time(), args.id))
event(db, args.id, {'event': 'queue_cancelled', 'reason': args.reason})
row = db.execute('SELECT * FROM queue_items WHERE id=?', (args.id,)).fetchone()
emit({'status': 'cancelled', 'item': queue_row(row)})
return 0
def queue_schedule(repo, args):
"""Claim the next durable queue item and run integrate once (or until drain/stop)."""
if repo.current != repo.main or git(repo.main, 'branch', '--show-current') != 'dev':
deny('MAIN_WORKTREE_REQUIRED', 'Schedule integrations only from the main dev worktree')
processed, results = 0, []
while True:
with repo.db(True) as db:
item = db.execute(
"SELECT * FROM queue_items WHERE state='pending' ORDER BY priority DESC, created ASC LIMIT 1"
).fetchone()
if not item:
break
db.execute("UPDATE queue_items SET state='claimed',updated=?,claim_run=? WHERE id=? AND state='pending'",
(time.time(), None, item['id']))
claimed = db.execute('SELECT * FROM queue_items WHERE id=?', (item['id'],)).fetchone()
item = queue_row(claimed)
try:
admit_candidate(repo, task_row(repo, item['task']), item['head_sha'], item['expected_dev'])
except Denied as exc:
with repo.db(True) as db:
db.execute("UPDATE queue_items SET state='failed',reason=?,updated=? WHERE id=?",
(exc.code + ': ' + exc.message, time.time(), item['id']))
event(db, item['id'], {'event': 'queue_failed', 'code': exc.code})
results.append({'id': item['id'], 'status': 'failed', 'code': exc.code, 'message': exc.message})
processed += 1
if not args.drain:
break
continue
request_id = 'queue-' + item['id']
integrate_args = argparse.Namespace(task=item['task'], expected_dev=item['expected_dev'],
request_id=request_id)
try:
code = integrate(repo, integrate_args)
except Denied as exc:
with repo.db(True) as db:
state = 'pending' if exc.code == 'LOCK_BUSY' else 'failed'
db.execute("UPDATE queue_items SET state=?,reason=?,updated=?,claim_run=NULL WHERE id=?",
(state, exc.code + ': ' + exc.message, time.time(), item['id']))
event(db, item['id'], {'event': 'queue_rescheduled' if state == 'pending' else 'queue_failed',
'code': exc.code})
results.append({'id': item['id'], 'status': state, 'code': exc.code, 'message': exc.message})
processed += 1
if exc.code == 'LOCK_BUSY' or not args.drain:
break
continue
with repo.db(True) as db:
if code == 0:
db.execute("UPDATE queue_items SET state='completed',updated=?,claim_run=?,reason=NULL WHERE id=?",
(time.time(), request_id, item['id']))
event(db, item['id'], {'event': 'queue_completed', 'run_id': request_id})
results.append({'id': item['id'], 'status': 'completed', 'run_id': request_id})
else:
db.execute("UPDATE queue_items SET state='failed',updated=?,claim_run=?,reason=? WHERE id=?",
(time.time(), request_id, 'integrate exit ' + str(code), item['id']))
event(db, item['id'], {'event': 'queue_failed', 'exit_code': code})
results.append({'id': item['id'], 'status': 'failed', 'exit_code': code, 'run_id': request_id})
processed += 1
if not args.drain:
break
emit({'status': 'ok', 'processed': processed, 'results': results})
return 0 if all(r.get('status') == 'completed' for r in results) or not results else 1
def docs_project(repo, args):
"""Project registry/queue into a docs-apply request (row or whole-document CAS)."""
task = task_row(repo, args.task)
if repo.current != Path(task['worktree']):
deny('WORKTREE_MISMATCH', 'Project documents from the registered task worktree')
filename = args.file
if not w4.projectable_file(filename):
deny('PATH_OUT_OF_SCOPE', 'File is not registered as projectable')
mode = getattr(args, 'mode', None) or 'row'
if mode not in ('row', 'document'):
deny('INVALID_REQUEST', 'Projection mode must be row or document')
# Controlled apply still requires shared-governance docs (docs apply gate).
if getattr(args, 'apply', False) and filename not in SHARED_DOCS:
deny('PATH_OUT_OF_SCOPE', 'Controlled apply is limited to shared governance documents')
content = safe_path(repo.current, filename).read_text()
with repo.db(True) as db:
ensure_schema(db)
pending = [queue_row(r) for r in db.execute(
"SELECT * FROM queue_items WHERE task=? AND state IN ('pending','claimed') ORDER BY priority DESC, created",
(args.task,)).fetchall()]
active_runs = [dict(r) for r in db.execute(
"SELECT id,state,resources,kind FROM runs WHERE task=? AND state IN ('running','queued','recovery_required','busy','suspect','quarantine')",
(args.task,)).fetchall()]
queue_state = pending[0]['state'] if pending else 'idle'
if args.status is not None:
if args.status not in PROJECTION_STATUS_VOCAB:
deny('INVALID_REQUEST', 'Projection status must be planned|in_progress|blocked|done|cancelled',
status=args.status)
status = args.status
else:
machine_status = 'queued' if pending else 'idle'
status = PROJECTION_STATUS_FROM_MACHINE.get(machine_status, machine_status)
if status not in PROJECTION_STATUS_VOCAB:
deny('INVALID_REQUEST', 'Mapped projection status is not in vocabulary', status=status)
worktree_path = Path(task['worktree'])
try:
worktree_cell = str(worktree_path.relative_to(repo.main))
except ValueError:
worktree_cell = str(worktree_path)
if mode == 'document':
if filename not in SHARED_DOCS and not getattr(args, 'replacement', None):
deny('INVALID_REQUEST', 'Document projection outside shared docs requires --replacement')
replacement = args.replacement
if replacement is None:
# Default document projection keeps content and only emits CAS baseline.
replacement = content
change = w4.build_document_projection_change(
file=filename, current_text=content, replacement=replacement)
record_id = '@document'
source = 'document'
else:
if filename not in SHARED_DOCS:
deny('INVALID_REQUEST', 'Row projection currently supports shared governance docs only')
record_id = getattr(args, 'record', None) or args.task
existing = doc_rows(content, record_id)
expected_row = existing[0][1] if existing else None
if args.replacement:
replacement = args.replacement
elif filename.endswith('CURRENT.md'):
replacement = w4.build_projection_row(
task_id=task['id'], owner=task['owner'], branch=task['branch'],
worktree=worktree_cell, mutex=args.mutex or '-', status=status)
elif filename.endswith('FEATURE-MATRIX.md'):
# MATRIX statuses (done|partial|planned|…) differ from CURRENT projection vocab;
# callers must supply a full row including stable record_id via --replacement.
deny('INVALID_REQUEST',
'FEATURE-MATRIX row projection requires --replacement with full row including record_id')
else:
deny('INVALID_REQUEST', 'Row projection for this file requires --replacement')
change = w4.build_row_projection_change(
file=filename, record_id=record_id, current_row=expected_row,
replacement=replacement)
source = 'row'
with repo.db(True) as db:
ensure_schema(db)
w4.record_projection(
db, file=filename, record_id=record_id, task=args.task,
source=source, content_hash=change['expected_sha256'] or digest(content))
machine = {
'task': {
'id': task['id'], 'owner': task['owner'], 'branch': task['branch'],
'worktree': task['worktree'], 'state': task['state'],
'scopes': json.loads(task['scopes']), 'capabilities': json.loads(task['capabilities']),
'head_sha': git(task['worktree'], 'rev-parse', 'HEAD'),
},
'queue': pending,
'runs': active_runs,
'queue_phase': queue_state,
'dev_sha': git(repo.main, 'rev-parse', 'dev'),
'mode': mode,
'status': status,
'record_id': record_id,
'mutex': args.mutex or '-',
}
request_id = args.request_id or (
f"{args.task}-project-{digest(json.dumps(machine, sort_keys=True, default=str))[:12]}")
machine['snapshot_time'] = time.time()
request = {
'request_id': request_id,
'changes': [change],
}
applied = None
if getattr(args, 'apply', False):
# Controlled apply via the same docs-apply entrypoint (lease + receipt).
req_path = Path(tempfile.mkdtemp(prefix='agentctl-project-')) / 'request.json'
req_path.write_text(json.dumps(request), encoding='utf-8')
apply_args = argparse.Namespace(task=args.task, request=str(req_path))
code = docs_apply(repo, apply_args)
applied = {'exit_code': code, 'request': str(req_path)}
if code != 0:
emit({'status': 'project_apply_failed', 'request': request, 'machine': machine,
'applied': applied})
return code
emit({
'status': 'projected' if not applied else 'projected_and_applied',
'request': request,
'machine': machine,
'applied': applied,
})
return 0
def docs_projections(repo, args):
"""Read-only projection ledger readout from the projections table."""
require_policy(repo)
with repo.db(True) as db:
ensure_schema(db)
if getattr(args, 'file', None):
rows = db.execute(
'SELECT file,record_id,task,source,content_hash,updated FROM projections '
'WHERE file=? ORDER BY updated, record_id',
(args.file,)).fetchall()
else:
rows = db.execute(
'SELECT file,record_id,task,source,content_hash,updated FROM projections '
'ORDER BY file, updated, record_id').fetchall()
emit({'status': 'ok', 'projections': [dict(row) for row in rows]})
return 0
def deploy_cmd(repo, args):
"""W4 deploy/recover helpers; never touches shared docker from this entrypoint."""
task = task_row(repo, args.task) if getattr(args, 'task', None) else None
action = args.deploy_action
with repo.db(True) as db:
ensure_schema(db)
if action == 'run-loop':
if not task:
deny('INVALID_REQUEST', 'deploy run-loop requires --task')
command = list(args.command or [])
if command and command[0] == '--':
command = command[1:]
if not command and not args.dry_run:
deny('INVALID_REQUEST', 'Provide a deploy command argv or --dry-run')
if args.dry_run and not command:
command = [sys.executable, '-c', 'print("dry-run-placeholder")']
result = w4.execute_deploy_recover_loop(
db,
deploy_id=args.id,
task=args.task,
env=args.env,
expected_commit=args.expected_commit,
command=command,
dry_run=bool(args.dry_run),
manifest=args.manifest,
cwd=str(repo.current),
)
emit({'status': result['status'], **result})
return 0 if result['status'] == 'passed' else 1
if action == 'status':
row = w4.deployment_get(db, args.id)
if not row:
deny('UNKNOWN_DEPLOY', 'Deploy id not found')
emit({'status': 'ok', 'deploy': row})
return 0
if action == 'recover':
row = w4.recover_deploy(db, args.id, reason=args.reason)
emit({'status': 'recovered', 'deploy': row})
return 0
deny('INVALID_REQUEST', f'Unknown deploy action {action}')
def cleanup_process_gate(path):
"""Report process-cwd visibility without deleting. Fail-closed unknowns are explicit."""
path = Path(path).resolve()
visibility = w4.cwd_visibility(path)
status = visibility.get('status')
if status == 'busy':
return {
'status': 'blocked', 'platform': sys.platform,
'holders': visibility.get('pids', []),
'unreadable': visibility.get('unreadable', []),
'reason': 'live process cwd uses worktree',
}
if status == 'clear':
return {
'status': 'clear', 'platform': sys.platform,
'holders': [], 'unreadable': [], 'reason': None,
'platform_note': visibility.get('platform_note'),
}
if status == 'unsupported':
return {
'status': 'unknown', 'platform': sys.platform,
'reason': visibility.get('message') or 'process cwd enumeration unsupported; cleanup_pending',
'holders': [], 'unreadable': [],
}
return {
'status': 'unknown', 'platform': sys.platform,
'holders': [],
'unreadable': visibility.get('unreadable', []),
'reason': visibility.get('message') or 'cannot verify cwd; cleanup_pending',
'platform_note': visibility.get('platform_note'),
}
def cleanup_status(repo, args):
"""Structured cleanup gate report; never removes a worktree."""
if args.path:
target = Path(args.path).resolve()
else:
matches = [e for e in repo.worktrees() if e.get('branch') == f'refs/heads/{args.branch}']
if len(matches) != 1:
deny('UNKNOWN_WORKTREE', 'Expected exactly one worktree for branch', branch=args.branch)
target = Path(matches[0]['worktree']).resolve()
gates = []
def gate(name, status, detail=None, **extra):
item = {'name': name, 'status': status, 'detail': detail}
item.update(extra)
gates.append(item)
if not target.is_dir():
gate('path_exists', 'fail', 'worktree directory missing')
emit({'status': 'blocked', 'path': str(target), 'gates': gates, 'cleanup_allowed': False})
return 0
branch_result = git(target, 'branch', '--show-current', check=False)
head_result = git(target, 'rev-parse', 'HEAD', check=False)
branch = branch_result.stdout.strip() if branch_result.returncode == 0 else ''
head = head_result.stdout.strip() if head_result.returncode == 0 else ''
if not branch or not head:
gate('git_identity', 'fail', 'unable to read branch/HEAD')
else:
gate('git_identity', 'pass', branch=branch, head=head)
if git(repo.main, 'merge-base', '--is-ancestor', head, 'refs/heads/dev', check=False).returncode:
gate('dev_ancestor', 'fail', 'task HEAD is not an ancestor of dev; cleanup_pending')
else:
gate('dev_ancestor', 'pass')
unfinished = []
for marker in ('MERGE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD', 'rebase-merge', 'rebase-apply', 'BISECT_START'):
marker_abs = Path(git(target, 'rev-parse', '--path-format=absolute', '--git-path', marker))
if marker_abs.exists():
unfinished.append(marker)
if unfinished:
gate('git_operation', 'fail', 'unfinished Git operation; cleanup_pending', markers=unfinished)
else:
gate('git_operation', 'pass')
porcelain = git(target, 'status', '--porcelain=v1', '--untracked-files=all', '--ignored=matching')
if porcelain:
gate('worktree_clean', 'fail', 'tracked, untracked or ignored files remain',
sample=porcelain.splitlines()[:20])
else:
gate('worktree_clean', 'pass')
with repo.db(True) as db:
tasks = [dict(r) for r in db.execute('SELECT id,state FROM tasks WHERE worktree=?', (str(target),)).fetchall()]
busy = []
for task in tasks:
row = db.execute(
"SELECT id,state FROM runs WHERE task=? AND state IN ('running','recovery_required','queued','suspect','quarantine')",
(task['id'],)).fetchone()
if row:
busy.append(dict(row))
if any(t['state'] == 'active' for t in tasks):
gate('registry_task', 'fail', 'active registered task owns this worktree', tasks=tasks)
else:
gate('registry_task', 'pass', tasks=tasks)
if busy:
gate('registry_runs', 'fail', 'live or unresolved operation', runs=busy)
else:
gate('registry_runs', 'pass')
key = digest(str(target))
lock_path = repo.root / 'os' / f'worktree-{key}.lock'
lifecycle = 'unknown'
if lock_path.exists():
try:
with lock_path.open('a+') as handle:
try:
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
fcntl.flock(handle, fcntl.LOCK_UN)
lifecycle = 'free'
except BlockingIOError:
lifecycle = 'held'
except OSError as exc:
lifecycle = f'error:{exc}'
else:
lifecycle = 'absent'
gate('lifecycle_lock', 'fail' if lifecycle == 'held' else ('pass' if lifecycle in ('free', 'absent') else 'unknown'),
lifecycle)
process = cleanup_process_gate(target)
gate('process_cwd', 'pass' if process['status'] == 'clear' else (
'fail' if process['status'] == 'blocked' else 'unknown'), process.get('reason'), **process)
allowed = all(g['status'] == 'pass' for g in gates)
emit({
'status': 'allowed' if allowed else 'blocked',
'path': str(target),
'cleanup_allowed': allowed,
'gates': gates,
})
return 0
def cleanup_report(repo, args=None):
"""Batch cleanup readiness for all .worktrees entries. Never deletes."""
# Lazy import keeps agentctl usable when agent_worktree is mid-edit in tests.
import agent_worktree as aw # noqa: E402
payload = aw.build_cleanup_report(repo.main, repo.main)
# Enrich with registry owners when available.
with repo.db() as db:
try:
ensure_schema(db)
except Exception:
pass
for item in payload.get('worktrees', []):
path = item.get('path')
if not path:
continue
rows = [dict(r) for r in db.execute(
'SELECT id,owner,state FROM tasks WHERE worktree=?', (path,)).fetchall()]
item['tasks'] = rows
active = [r for r in rows if r.get('state') == 'active']
item['cleanup_owner'] = (active[0]['owner'] if active
else (rows[0]['owner'] if rows else None))
record_audit(repo, 'cleanup_report', payload)
emit_payload(payload, getattr(args, 'format', 'json') if args is not None else 'json',
visibility.render_cleanup_table)
return 0
def docs_drift(repo, args=None):
"""Compare FEATURE-MATRIX status columns against last snapshot.
Hard alerts only for done → non-done regressions (ISSUE-20260923 toolization).
Open (non-terminal) rows are listed as soft candidates for human judgment.
Read-only: never writes CURRENT/MATRIX.
"""
matrix_path = repo.main / '.doc/status/FEATURE-MATRIX.md'
if not matrix_path.is_file():
deny('INVALID_DOCUMENT', 'FEATURE-MATRIX.md missing on main worktree')
# Prefer committed tip so dirty local MATRIX edits do not pollute drift.
tip = git(repo.main, 'show', 'refs/heads/dev:.doc/status/FEATURE-MATRIX.md', check=False)
content = tip.stdout if tip.returncode == 0 else matrix_path.read_text()
current = matrix_states(content)
previous = {}
with repo.db(True) as db:
ensure_schema(db)
for row in db.execute('SELECT record_id, status FROM matrix_state_snapshot'):
previous[row['record_id']] = row['status']
regressions = []
for record_id, prev in previous.items():
now = current.get(record_id)
if prev == 'done' and now is not None and now != 'done':
regressions.append({
'record_id': record_id,
'previous': prev,
'current': now,
'severity': 'hard',
'message': 'MATRIX status regressed from done',
})
elif prev == 'done' and record_id not in current:
regressions.append({
'record_id': record_id,
'previous': prev,
'current': None,
'severity': 'hard',
'message': 'MATRIX done row disappeared',
})
now_ts = time.time()
db.execute('DELETE FROM matrix_state_snapshot')
for record_id, status in current.items():
db.execute(
'INSERT INTO matrix_state_snapshot(record_id, status, updated) VALUES(?,?,?)',
(record_id, status, now_ts))
open_rows = sorted(
[{'record_id': rid, 'status': st} for rid, st in current.items()
if st not in ('done', 'removed')],
key=lambda item: item['record_id'])
payload = {
'status': 'ok',
'file': '.doc/status/FEATURE-MATRIX.md',
'record_count': len(current),
'previous_count': len(previous),
'regressions': regressions,
'open_capabilities': open_rows,
'baseline': 'first_snapshot' if not previous else 'compared',
'note': ('Hard alerts are done→non-done only; open_capabilities are '
'candidates for human MATRIX updates via docs apply.'),
}
record_audit(repo, 'docs_drift', payload)
emit(payload)
return 0
def run_audit_cycle(repo):
"""Daemon-side read-only audit: cleanup report + docs drift. Never docs_apply."""
results = {}
try:
cleanup_report(repo)
results['cleanup_report'] = 'ok'
except Exception as exc: # noqa: BLE001 — daemon must continue
results['cleanup_report'] = {'status': 'error', 'error': str(exc)}
try:
docs_drift(repo)
results['docs_drift'] = 'ok'
except Exception as exc: # noqa: BLE001
results['docs_drift'] = {'status': 'error', 'error': str(exc)}
with repo.db(True) as db:
ensure_schema(db)
event(db, 'daemon', {'event': 'audit_cycle', **results})
return results
def _claim_fingerprint(task_id, resources, mode='X'):
return digest(json.dumps({'task': task_id, 'resources': sorted(resources), 'mode': mode}, sort_keys=True))
def claim_acquire(repo, args):
task = task_row(repo, args.task)
resources = sorted({resource(item) for item in args.resource})
if not resources or any(not item.startswith('claim:') for item in resources):
deny('INVALID_RESOURCE', 'claim acquire requires one or more claim: resources')
check_resources(repo, task, resources)
request_id = args.request_id or str(uuid.uuid4())
if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', request_id):
deny('INVALID_REQUEST', 'Invalid request id')
fingerprint = _claim_fingerprint(task['id'], resources)
wait = max(0.0, min(float(args.wait or 0), 60.0))
deadline = time.monotonic() + wait
while True:
busy = None
with repo.db(True) as db:
prior = [dict(r) for r in db.execute(
'SELECT * FROM claims WHERE request_id=?', (request_id,)).fetchall()]
if prior:
held = [row for row in prior if row['state'] == 'held']
payload = sorted(row['resource'] for row in held) if held else sorted(
row['resource'] for row in prior)
same_task = all(row['task'] == task['id'] for row in prior)
same_fp = _claim_fingerprint(task['id'], payload) == fingerprint and same_task
if held and same_fp and set(payload) == set(resources):
emit({'status': 'held', 'replayed': True, 'request_id': request_id,
'claims': held})
return 0
deny('REQUEST_ID_REUSED', 'Claim request id belongs to a different payload',
request_id=request_id)
frozen = []
for row in db.execute('SELECT * FROM freezes'):
if any(overlaps(item, row['resource']) for item in resources):
frozen.append(dict(row))
if frozen:
deny('FROZEN', 'Resource is frozen', resource=frozen[0]['resource'],
holder=frozen[0]['owner'])
holders = []
for row in db.execute("SELECT * FROM claims WHERE state='held'"):
if any(overlaps(item, row['resource']) for item in resources):
# Self re-acquire of identical resource is covered by UNIQUE(resource,task)
if row['task'] == task['id'] and row['resource'] in resources:
continue
holders.append(dict(row))
if holders:
tickets = []
now = time.time()
for item in resources:
ticket = str(uuid.uuid4())
db.execute(
'INSERT INTO claim_queue(id,resource,task,request_id,state,created) VALUES(?,?,?,?,?,?)',
(ticket, item, task['id'], request_id, 'queued', now))
tickets.append(ticket)
event(db, ticket, {'event': 'claim_queued', 'resource': item,
'task': task['id'], 'holder': holders[0]['task']})
busy = {
'holder': holders[0]['task'],
'ticket': tickets[0],
'tickets': tickets,
'retry_after': 1,
'resources': resources,
}
else:
now = time.time()
claims = []
for item in resources:
existing = db.execute(
'SELECT * FROM claims WHERE resource=? AND task=?',
(item, task['id'])).fetchone()
if existing and existing['state'] == 'held':
claims.append(dict(existing))
continue
claim_id = request_id if len(resources) == 1 else f'{request_id}:{digest(item)[:12]}'
if existing:
db.execute(
'''UPDATE claims SET id=?, mode=?, owner=?, request_id=?, state='held',
reason=?, updated=? WHERE resource=? AND task=?''',
(claim_id, 'X', task['owner'], request_id, fingerprint, now,
item, task['id']))
else:
db.execute(
'''INSERT INTO claims(id,resource,mode,task,owner,request_id,state,reason,created,updated)
VALUES(?,?,?,?,?,?,'held',?,?,?)''',
(claim_id, item, 'X', task['id'], task['owner'], request_id,
fingerprint, now, now))
row = dict(db.execute(
'SELECT * FROM claims WHERE resource=? AND task=?',
(item, task['id'])).fetchone())
claims.append(row)
event(db, claim_id, {'event': 'claim_acquired', 'resource': item,
'task': task['id'], 'request_id': request_id})
emit({'status': 'held', 'request_id': request_id, 'claims': claims})
return 0
if busy:
if time.monotonic() >= deadline:
deny('LOCK_BUSY', 'Claim resource is held; retry with ticket or do independent work',
**busy)
time.sleep(min(0.05, max(0.0, deadline - time.monotonic())))
continue
deny('RECOVERY_REQUIRED', 'Claim acquire exited without result')
def claim_release(repo, args):
task = task_row(repo, args.task)
target = resource(args.resource)
if not target.startswith('claim:'):
deny('INVALID_RESOURCE', 'claim release requires a claim: resource')
with repo.db(True) as db:
row = db.execute(
"SELECT * FROM claims WHERE resource=? AND task=? AND state='held'",
(target, task['id'])).fetchone()
if not row:
deny('UNKNOWN_CLAIM', 'No held claim for this task/resource')
now = time.time()
db.execute("UPDATE claims SET state='released', updated=?, reason=? WHERE id=?",
(now, args.reason or 'released', row['id']))
event(db, row['id'], {'event': 'claim_released', 'resource': target, 'task': task['id']})
# Promote earliest queued waiter to visibility only; acquire still races fairly.
waiting = db.execute(
"SELECT * FROM claim_queue WHERE resource=? AND state='queued' ORDER BY created LIMIT 1",
(target,)).fetchone()
emit({'status': 'released', 'claim': dict(row), 'next_ticket': waiting['id'] if waiting else None})
return 0
def claim_status(repo, args):
target = resource(args.resource)
if not target.startswith('claim:'):
deny('INVALID_RESOURCE', 'claim status requires a claim: resource')
with repo.db() as db:
try:
held = [dict(r) for r in db.execute(
"SELECT * FROM claims WHERE state='held'").fetchall()
if overlaps(target, r['resource'])]
queued = [dict(r) for r in db.execute(
"SELECT * FROM claim_queue WHERE state='queued' ORDER BY created").fetchall()
if overlaps(target, r['resource'])]
except sqlite3.Error:
held, queued = [], []
emit({'status': 'ok', 'resource': target, 'held': held, 'queued': queued})
return 0
def claim_list(repo, args):
with repo.db() as db:
try:
if args.all:
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 sqlite3.Error:
rows = []
payload = {'status': 'ok', 'claims': rows}
emit_payload(payload, getattr(args, 'format', 'json'), visibility.render_claim_table)
return 0
def claim_archive(repo, args):
"""Migrate done CURRENT claim rows to archive via docs apply @document CAS."""
task = task_row(repo, args.task)
if repo.current != Path(task['worktree']) or repo.current == repo.main:
deny('WORKTREE_MISMATCH', 'claim archive must run inside the task worktree')
caps = json.loads(task['capabilities']) if isinstance(task['capabilities'], str) else task['capabilities']
if 'docs:admin' not in caps:
deny('CAPABILITY_REQUIRED', 'claim archive rewrites CURRENT via @document and needs docs:admin')
current_rel = '.doc/ops/CURRENT.md'
archive_rel = '.doc/archive/ops/2026-09-21-current-history.md'
current_path = safe_path(repo.current, current_rel)
content = current_path.read_text(encoding='utf-8')
rows = visibility.parse_current_claim_rows(content)
selected = visibility.select_archive_candidates(
rows,
retain_days=getattr(args, 'retain_days', 7),
include_undated=bool(getattr(args, 'include_undated', False)),
)
eligible = [r for r in selected if r.get('eligible')]
payload = {
'status': 'dry_run',
'retain_days': getattr(args, 'retain_days', 7),
'include_undated': bool(getattr(args, 'include_undated', False)),
'done_scanned': len(selected),
'eligible_count': len(eligible),
'candidates': eligible,
'skipped': [r for r in selected if not r.get('eligible')],
'archive_file': archive_rel,
}
if not getattr(args, 'apply', False):
emit(payload)
return 0
if not eligible:
emit({**payload, 'status': 'noop', 'message': 'no eligible done rows to archive'})
return 0
remove_ids = [r['task'] for r in eligible]
new_content = visibility.build_archived_current(content, remove_ids)
archive_date = time.strftime('%Y-%m-%d', time.localtime())
section = visibility.build_archive_section(eligible, archive_date)
request = {
'request_id': f"{args.task}-claim-archive-{archive_date.replace('-', '')}",
'changes': [{
'file': current_rel,
'record_id': '@document',
'expected_sha256': digest(content),
'replacement': new_content,
}],
}
req_path = Path(tempfile.mkstemp(prefix='agentctl-claim-archive-', suffix='.json')[1])
try:
req_path.write_text(json.dumps(request, ensure_ascii=False, indent=2), encoding='utf-8')
apply_args = argparse.Namespace(task=args.task, request=str(req_path))
rc = docs_apply(repo, apply_args)
if rc not in (0, None):
return rc
finally:
with contextlib.suppress(OSError):
req_path.unlink()
archive_path = safe_path(repo.current, archive_rel)
with Lease(repo, task, [f'path:{archive_rel}'], f'{args.task}-archive-append',
digest(section)) as lease:
if lease.replay:
emit({'status': 'replayed', 'run_id': lease.request_id})
return lease.row['exit_code'] or 0
existing = archive_path.read_text(encoding='utf-8') if archive_path.is_file() else ''
atomic_write(archive_path, existing.rstrip() + '\n' + section)
lease.finish()
emit({
'status': 'archived',
'archived_count': len(eligible),
'tasks': remove_ids,
'archive_file': archive_rel,
'current_sha256': digest(new_content),
})
return 0
def collect_status_payload(repo):
identity = repo.identity()
policy = policy_status(repo.main)
try:
with repo.db(True) as db:
ensure_schema(db)
except Denied:
pass
with repo.db() as db:
tasks = [dict(r) for r in db.execute('SELECT * FROM tasks')]
runs = [dict(r) for r in db.execute(
"SELECT * FROM runs WHERE state IN ('running','queued','recovery_required','busy','suspect','quarantine')")]
try:
queue = [dict(r) for r in db.execute(
"SELECT * FROM queue_items WHERE state IN ('pending','claimed') ORDER BY priority DESC, created")]
except sqlite3.Error:
queue = []
for row in runs:
row['supervisor_alive'] = alive_status(row['pid'], row['start'], row['boot'])
return {
'status': 'ok',
'lock_root': str(repo.root),
'main': str(repo.main),
'version': VERSION,
'epoch': identity['epoch'],
'tasks': tasks,
'runs': runs,
'queue': queue,
'policy': policy,
}
def collect_cleanup_payload(repo):
import agent_worktree as aw # noqa: E402
payload = aw.build_cleanup_report(repo.main, repo.main)
with repo.db() as db:
try:
ensure_schema(db)
except Exception:
pass
for item in payload.get('worktrees', []):
path = item.get('path')
if not path:
continue
rows = [dict(r) for r in db.execute(
'SELECT id,owner,state FROM tasks WHERE worktree=?', (path,)).fetchall()]
item['tasks'] = rows
active = [r for r in rows if r.get('state') == 'active']
item['cleanup_owner'] = (active[0]['owner'] if active
else (rows[0]['owner'] if rows else None))
return payload
def collect_docs_drift_payload(repo):
"""Same logic as docs_drift but returns payload without emitting."""
matrix_path = repo.main / '.doc/status/FEATURE-MATRIX.md'
if not matrix_path.is_file():
return {'status': 'error', 'error': 'FEATURE-MATRIX.md missing'}
tip = git(repo.main, 'show', 'refs/heads/dev:.doc/status/FEATURE-MATRIX.md', check=False)
content = tip.stdout if tip.returncode == 0 else matrix_path.read_text()
current = matrix_states(content)
previous = {}
with repo.db(True) as db:
ensure_schema(db)
for row in db.execute('SELECT record_id, status FROM matrix_state_snapshot'):
previous[row['record_id']] = row['status']
regressions = []
for record_id, prev in previous.items():
now = current.get(record_id)
if prev == 'done' and now is not None and now != 'done':
regressions.append({
'record_id': record_id, 'previous': prev, 'current': now,
'severity': 'hard', 'message': 'MATRIX status regressed from done',
})
elif prev == 'done' and record_id not in current:
regressions.append({
'record_id': record_id, 'previous': prev, 'current': None,
'severity': 'hard', 'message': 'MATRIX done row disappeared',
})
now_ts = time.time()
db.execute('DELETE FROM matrix_state_snapshot')
for record_id, status in current.items():
db.execute(
'INSERT INTO matrix_state_snapshot(record_id, status, updated) VALUES(?,?,?)',
(record_id, status, now_ts))
open_rows = sorted(
[{'record_id': rid, 'status': st} for rid, st in current.items()
if st not in ('done', 'removed')],
key=lambda item: item['record_id'])
return {
'status': 'ok',
'file': '.doc/status/FEATURE-MATRIX.md',
'record_count': len(current),
'previous_count': len(previous),
'regressions': regressions,
'open_capabilities': open_rows,
'baseline': 'first_snapshot' if not previous else 'compared',
}
def collect_daemon_payload(repo):
try:
identity = repo.identity()
except Denied as exc:
return {'status': 'denied', 'code': exc.code, 'message': exc.message}
lock = manager_lock_status(repo)
heartbeat = read_daemon_heartbeat(repo)
heartbeat_age = None
if heartbeat and isinstance(heartbeat.get('updated'), (int, float)):
heartbeat_age = max(0.0, time.time() - float(heartbeat['updated']))
pending = claimed = 0
try:
with repo.db() as db:
try:
pending = db.execute(
"SELECT count(*) FROM queue_items WHERE state='pending'").fetchone()[0]
claimed = db.execute(
"SELECT count(*) FROM queue_items WHERE state='claimed'").fetchone()[0]
except sqlite3.Error:
pending = claimed = 0
except Denied:
pass
return {
'status': 'ok',
'lock_root': str(repo.root),
'epoch': identity.get('epoch'),
'manager_lock': lock,
'heartbeat': heartbeat,
'heartbeat_age_seconds': heartbeat_age,
'queue': {'pending': pending, 'claimed': claimed},
}
def collect_catalog_drift_payload(repo):
matrix_path = repo.main / '.doc/status/FEATURE-MATRIX.md'
try:
content = git(repo.main, 'show', 'refs/heads/dev:.doc/status/FEATURE-MATRIX.md')
except Exception:
content = matrix_path.read_text(encoding='utf-8') if matrix_path.is_file() else ''
states = matrix_states(content) if content else {}
with repo.db(True) as db:
return catalog.catalog_drift(db, matrix_states=states)
def cmd_brief(repo, args):
"""One-page Markdown briefing aggregating status/drift/cleanup/catalog/daemon."""
status_payload = collect_status_payload(repo)
try:
docs = collect_docs_drift_payload(repo)
except Exception as exc: # noqa: BLE001
docs = {'status': 'error', 'error': str(exc)}
try:
cleanup = collect_cleanup_payload(repo)
except Exception as exc: # noqa: BLE001
cleanup = {'status': 'error', 'error': str(exc), 'worktrees': []}
try:
cat = collect_catalog_drift_payload(repo)
except Exception as exc: # noqa: BLE001
cat = {'status': 'error', 'error': str(exc)}
try:
daemon = collect_daemon_payload(repo)
except Exception as exc: # noqa: BLE001
daemon = {'status': 'error', 'error': str(exc)}
text = visibility.build_brief(
status=status_payload,
docs_drift=docs,
cleanup=cleanup,
catalog_drift=cat,
daemon=daemon,
)
if getattr(args, 'format', 'markdown') == 'json':
emit({
'status': 'ok',
'brief': text,
'sources': {
'status': {
'active_tasks': len([
t for t in (status_payload.get('tasks') or [])
if t.get('state') not in ('done', 'cancelled')
]),
'runs': len(status_payload.get('runs') or []),
'queue': len(status_payload.get('queue') or []),
},
'docs_drift': docs,
'cleanup': {'worktrees': len((cleanup or {}).get('worktrees') or [])},
'catalog_drift': cat,
'daemon': daemon,
},
})
else:
sys.stdout.write(text)
sys.stdout.flush()
return 0
def claim_cancel(repo, args):
task = task_row(repo, args.task)
with repo.db(True) as db:
row = db.execute('SELECT * FROM claim_queue WHERE id=?', (args.id,)).fetchone()
if not row:
deny('UNKNOWN_TICKET', 'Unknown claim queue ticket')
if row['task'] != task['id']:
deny('PATH_OUT_OF_SCOPE', 'Only the ticket owner can cancel')
if row['state'] != 'queued':
deny('INVALID_REQUEST', f'Ticket is already {row["state"]}')
db.execute("UPDATE claim_queue SET state='cancelled' WHERE id=?", (args.id,))
event(db, args.id, {'event': 'claim_cancelled', 'reason': args.reason, 'task': task['id']})
emit({'status': 'cancelled', 'ticket': args.id, 'reason': args.reason})
return 0
MANAGER_LOCK_NAME = 'manager.lock'
DAEMON_HEARTBEAT_NAME = 'daemon.heartbeat'
def _read_manager_holder_pid(path):
try:
raw = path.read_text().strip()
except OSError:
return None
if not raw:
return None
try:
return int(raw.split()[0])
except ValueError:
return None
@contextlib.contextmanager
def acquire_manager_lock(repo):
"""Exclusive flock on manager.lock; not part of physical_names() lease set."""
path = repo.os_file(MANAGER_LOCK_NAME)
handle = path.open('a+')
try:
try:
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
holder_pid = None
for _ in range(20):
holder_pid = _read_manager_holder_pid(path)
if holder_pid is not None:
break
time.sleep(0.01)
deny('LOCK_BUSY', 'Daemon manager lock is held',
holder_pid=holder_pid, retry_after=1)
handle.seek(0)
handle.truncate()
handle.write(f'{os.getpid()}\n')
handle.flush()
os.fsync(handle.fileno())
yield handle
finally:
try:
fcntl.flock(handle, fcntl.LOCK_UN)
except OSError:
pass
handle.close()
def write_daemon_heartbeat(repo, cycle):
payload = {
'pid': os.getpid(),
'updated': time.time(),
'cycle': cycle,
}
atomic_write(repo.os_file(DAEMON_HEARTBEAT_NAME), json.dumps(payload, sort_keys=True))
def read_daemon_heartbeat(repo):
path = repo.root / 'os' / DAEMON_HEARTBEAT_NAME
if not path.exists() or path.is_symlink():
return None
try:
data = json.loads(path.read_text())
except (OSError, json.JSONDecodeError, UnicodeDecodeError):
return None
if not isinstance(data, dict):
return None
return data
def manager_lock_status(repo):
"""Read-only probe: never create files or take EX locks."""
path = repo.root / 'os' / MANAGER_LOCK_NAME
if not path.exists() or path.is_symlink():
return {'held': False, 'holder_pid': None, 'path': str(path)}
try:
with path.open('r') as handle:
try:
fcntl.flock(handle, fcntl.LOCK_SH | fcntl.LOCK_NB)
except BlockingIOError:
return {
'held': True,
'holder_pid': _read_manager_holder_pid(path),
'path': str(path),
}
fcntl.flock(handle, fcntl.LOCK_UN)
except OSError as exc:
return {'held': False, 'holder_pid': None, 'path': str(path), 'error': str(exc)}
return {'held': False, 'holder_pid': None, 'path': str(path)}
def daemon_run(repo, args):
"""Foreground coordinator loop: exclusive manager.lock + queue_schedule cycles."""
interval = float(getattr(args, 'interval', 1.0) or 1.0)
if interval < 0:
deny('INVALID_REQUEST', 'daemon interval must be >= 0')
max_cycles = getattr(args, 'max_cycles', None)
if max_cycles is not None:
max_cycles = int(max_cycles)
if max_cycles < 1:
deny('INVALID_REQUEST', 'daemon max-cycles must be >= 1')
audit_every = int(getattr(args, 'audit_every', 30) or 0)
if audit_every < 0:
deny('INVALID_REQUEST', 'daemon audit-every must be >= 0')
stop = {'flag': False}
def _request_stop(signum, frame):
stop['flag'] = True
previous_term = signal.signal(signal.SIGTERM, _request_stop)
previous_int = signal.signal(signal.SIGINT, _request_stop)
cycles = 0
try:
with acquire_manager_lock(repo):
emit({
'status': 'daemon_started',
'pid': os.getpid(),
'interval': interval,
'max_cycles': max_cycles,
'audit_every': audit_every,
})
while not stop['flag']:
cycles += 1
write_daemon_heartbeat(repo, cycles)
try:
queue_schedule(repo, argparse.Namespace(drain=False))
except Denied as exc:
detail = {
'event': 'daemon_schedule_denied',
'code': exc.code,
'message': exc.message,
**exc.details,
}
if exc.code != 'LOCK_BUSY':
with repo.db(True) as db:
ensure_schema(db)
event(db, 'daemon', detail)
emit({'status': 'denied', 'code': exc.code, 'message': exc.message,
**exc.details}, sys.stderr)
# Read-only audit: never docs_apply (main cwd would WORKTREE_MISMATCH).
if audit_every and cycles % audit_every == 0:
try:
run_audit_cycle(repo)
except Exception as exc: # noqa: BLE001
emit({'status': 'audit_cycle_error', 'error': str(exc)}, sys.stderr)
if max_cycles is not None and cycles >= max_cycles:
break
if stop['flag']:
break
deadline = time.monotonic() + interval
while not stop['flag'] and time.monotonic() < deadline:
time.sleep(min(0.05, max(0.0, deadline - time.monotonic())))
write_daemon_heartbeat(repo, cycles)
emit({'status': 'daemon_stopped', 'pid': os.getpid(), 'cycles': cycles,
'reason': 'signal' if stop['flag'] else 'max_cycles'})
return 0
finally:
signal.signal(signal.SIGTERM, previous_term)
signal.signal(signal.SIGINT, previous_int)
def daemon_status(repo, args):
"""Read-only daemon probe: lock holder, heartbeat age, queue counts."""
try:
identity = repo.identity()
except Denied as exc:
emit({'status': 'denied', 'code': exc.code, 'message': exc.message, **exc.details})
return 1
lock = manager_lock_status(repo)
heartbeat = read_daemon_heartbeat(repo)
heartbeat_age = None
if heartbeat and isinstance(heartbeat.get('updated'), (int, float)):
heartbeat_age = max(0.0, time.time() - float(heartbeat['updated']))
pending = claimed = 0
try:
with repo.db() as db:
try:
pending = db.execute(
"SELECT count(*) FROM queue_items WHERE state='pending'").fetchone()[0]
claimed = db.execute(
"SELECT count(*) FROM queue_items WHERE state='claimed'").fetchone()[0]
except sqlite3.Error:
pending = claimed = 0
except Denied:
pass
emit({
'status': 'ok',
'lock_root': str(repo.root),
'epoch': identity.get('epoch'),
'manager_lock': lock,
'heartbeat': heartbeat,
'heartbeat_age_seconds': heartbeat_age,
'queue': {'pending': pending, 'claimed': claimed},
})
return 0
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'):
sp=sub.add_parser(name)
if name in ('status','doctor'):
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')
reg.add_argument('--intent',help='Design intent text for overlap probe')
reg.add_argument('--feature',help='FEATURE-MATRIX record_id / catalog feature')
reg.add_argument('--mutex',help='Mutex key for hard same_mutex gate')
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)
proj=d.add_parser('project');proj.add_argument('--task',required=True);proj.add_argument('--file',default='.doc/ops/CURRENT.md')
proj.add_argument('--mutex',default='');proj.add_argument('--status');proj.add_argument('--request-id')
proj.add_argument('--mode',choices=['row','document'],default='row')
proj.add_argument('--record');proj.add_argument('--replacement')
proj.add_argument('--apply',action='store_true',help='Apply projected change via docs apply')
dproj=d.add_parser('projections');dproj.add_argument('--file')
d.add_parser('drift', help='Read-only MATRIX status regression audit')
q=sub.add_parser('queue').add_subparsers(dest='queue_action',required=True)
qe=q.add_parser('enqueue');qe.add_argument('--task',required=True);qe.add_argument('--expected-dev',required=True)
qe.add_argument('--head');qe.add_argument('--priority',type=int,default=0);qe.add_argument('--request-id')
ql=q.add_parser('list');ql.add_argument('--all',action='store_true');ql.add_argument('--format',choices=['json','table'],default='json')
qc=q.add_parser('cancel');qc.add_argument('--task',required=True);qc.add_argument('--id',required=True);qc.add_argument('--reason',required=True)
qs=q.add_parser('schedule');qs.add_argument('--drain',action='store_true')
cu=sub.add_parser('cleanup').add_subparsers(dest='cleanup_action',required=True)
cs=cu.add_parser('status');cs.add_argument('--path');cs.add_argument('--branch')
crep=cu.add_parser('report', help='Batch cleanup readiness for .worktrees (never deletes)')
crep.add_argument('--format',choices=['json','table'],default='json')
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')
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')
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)
dr.add_argument('--env',choices=['dev','test','main'],required=True)
dr.add_argument('--expected-commit',required=True);dr.add_argument('--manifest')
dr.add_argument('--dry-run',action='store_true')
dr.add_argument('command',nargs=argparse.REMAINDER)
ds=dep.add_parser('status');ds.add_argument('--id',required=True)
drec=dep.add_parser('recover');drec.add_argument('--id',required=True);drec.add_argument('--reason',required=True)
cl=sub.add_parser('claim').add_subparsers(dest='claim_action',required=True)
ca=cl.add_parser('acquire');ca.add_argument('--task',required=True);ca.add_argument('--resource',action='append',required=True)
ca.add_argument('--request-id');ca.add_argument('--wait',type=float,default=0)
crl=cl.add_parser('release');crl.add_argument('--task',required=True);crl.add_argument('--resource',required=True);crl.add_argument('--reason',default='')
cst=cl.add_parser('status');cst.add_argument('--resource',required=True);cst.add_argument('--json',action='store_true')
clist=cl.add_parser('list');clist.add_argument('--all',action='store_true')
clist.add_argument('--format',choices=['json','table'],default='json')
carc=cl.add_parser('archive', help='Archive done CURRENT claim rows via docs apply')
carc.add_argument('--task',required=True)
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')
qtn.add_parser('list')
qrel=qtn.add_parser('release');qrel.add_argument('--run',required=True);qrel.add_argument('--reason',required=True)
qrel.add_argument('--verified-by',required=True,dest='verified_by')
dm=sub.add_parser('daemon').add_subparsers(dest='daemon_action',required=True)
drun=dm.add_parser('run')
drun.add_argument('--interval',type=float,default=1.0,help='Seconds between schedule cycles')
drun.add_argument('--max-cycles',type=int,dest='max_cycles',
help='Stop after N cycles (tests / bounded runs)')
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)
cadd.add_argument('--title', required=True)
cadd.add_argument('--statement', required=True)
cadd.add_argument('--owner', required=True)
cadd.add_argument('--feature')
cadd.add_argument('--source-task')
cadd.add_argument('--source-doc')
cadd.add_argument('--state', default='proposed')
cadd.add_argument('--conversation-ref')
clist = cat.add_parser('list'); clist.add_argument('--state'); clist.add_argument('--limit', type=int, default=100)
cshow = cat.add_parser('show'); cshow.add_argument('--id', required=True)
csearch = cat.add_parser('search'); csearch.add_argument('--query', required=True); csearch.add_argument('--top', type=int, default=10)
cov = cat.add_parser('overlap')
cov.add_argument('--task', required=True)
cov.add_argument('--statement', required=True)
cov.add_argument('--requirement')
cov.add_argument('--feature')
cov.add_argument('--mutex')
cov.add_argument('--top', type=int, default=10)
cack = cat.add_parser('ack')
cack.add_argument('--task', required=True)
cack.add_argument('--overlap', required=True)
cack.add_argument('--decision', required=True, choices=sorted(catalog.DECISIONS))
cack.add_argument('--reason', required=True)
cack.add_argument('--decided-by', default='Cursor')
cack.add_argument('--authorization-ref')
cat.add_parser('status')
cat.add_parser('drift', help='Refresh features from MATRIX; report done→non-done regressions')
return p
def main():
args=parser().parse_args();repo=Repo(args.repo)
write_actions = {
'init', 'task', 'with', 'docs', 'validate', 'integrate', 'queue',
'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup', 'claim',
'quarantine', 'daemon', 'catalog', 'serve', 'pi',
}
# status/doctor remain readable without a policy file and never create one.
if args.action in write_actions:
if args.action == 'docs' and getattr(args, 'doc_action', None) == 'snapshot':
pass # snapshot is read-only CAS helper
elif args.action == 'docs' and getattr(args, 'doc_action', None) == 'projections':
pass # projection ledger readout is read-only
elif args.action == 'docs' and getattr(args, 'doc_action', None) == 'drift':
require_policy(repo) # drift needs schema; never writes shared docs
elif args.action == 'queue' and getattr(args, 'queue_action', None) == 'list':
pass
elif args.action == 'claim' and getattr(args, 'claim_action', None) in ('status', 'list'):
require_policy(repo) # status/list need schema constants but are non-mutating
elif args.action == 'claim' and getattr(args, 'claim_action', None) == 'archive' and not getattr(args, 'apply', False):
require_policy(repo) # dry-run is non-mutating
elif args.action == 'quarantine' and getattr(args, 'quarantine_action', None) == 'list':
require_policy(repo) # list needs TTL/schema constants
elif args.action == 'daemon' and getattr(args, 'daemon_action', None) == 'status':
pass # read-only probe; never creates manager.lock or heartbeat
elif args.action == 'catalog' and getattr(args, 'catalog_action', None) in (
'list', 'show', 'search', 'status',
):
require_policy(repo) # needs schema; non-mutating reads
elif args.action == 'cleanup':
require_policy(repo)
elif args.action == 'hooks' and getattr(args, 'hook_action', None) == 'codegraph-sync':
pass # fail-open sync; must not require policy or deny merges
else:
require_policy(repo)
if args.action=='init':initialize(repo)
elif args.action in ('status','doctor'):
payload = collect_status_payload(repo)
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')
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)
# Catalog gate BEFORE insert: probe commits in its own txn so soft hits survive deny.
req = getattr(args, 'requirement', None)
intent = getattr(args, 'intent', None)
feature = getattr(args, 'feature', None)
mutex = getattr(args, 'mutex', None)
if req or intent or feature or mutex:
with repo.db(True) as db:
gate_result = catalog.gate_task_register(
db,
task=args.id,
scopes=scopes,
requirement=req,
intent=intent,
feature=feature,
mutex=mutex,
overlaps_fn=overlaps,
statement=intent,
)
catalog.enforce_gate_result(gate_result)
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','suspect','quarantine')",(args.id,)).fetchone():deny('LOCK_BUSY','Task still has an active or unresolved operation')
if db.execute("SELECT id FROM claims WHERE task=? AND state='held'",(args.id,)).fetchone():
deny('LOCK_BUSY','Task still holds claims; release them before close')
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)
elif args.doc_action=='projections':return docs_projections(repo,args)
elif args.doc_action=='project':return docs_project(repo,args)
elif args.doc_action=='drift':return docs_drift(repo,args)
else:return docs_apply(repo,args)
elif args.action=='queue':
if args.queue_action=='enqueue':return queue_enqueue(repo,args)
if args.queue_action=='list':return queue_list(repo,args)
if args.queue_action=='cancel':return queue_cancel(repo,args)
return queue_schedule(repo,args)
elif args.action=='cleanup':
if args.cleanup_action=='report':
return cleanup_report(repo,args)
if not args.path and not args.branch:
deny('INVALID_REQUEST', 'cleanup status requires --path or --branch')
if args.path and args.branch:
deny('INVALID_REQUEST', 'Provide only one of --path or --branch')
return cleanup_status(repo,args)
elif args.action=='recover':recover(repo,args)
elif args.action=='operation':
with repo.db() as db:row=db.execute('SELECT * FROM runs WHERE id=?',(args.run,)).fetchone()
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','suspect','quarantine')"):
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)
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':
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)
if args.claim_action=='release':return claim_release(repo,args)
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)
if args.quarantine_action=='list':return quarantine_list(repo)
if args.quarantine_action=='release':return quarantine_release(repo,args)
elif args.action=='daemon':
if args.daemon_action=='run':return daemon_run(repo,args)
if args.daemon_action=='status':return daemon_status(repo,args)
elif args.action=='catalog':
return catalog_cmd(repo, args)
return 0
def catalog_cmd(repo, args):
"""Dispatch agentctl catalog subcommands."""
action = args.catalog_action
if action == 'add':
with repo.db(True) as db:
result = catalog.requirement_add(
db,
req_id=args.id,
title=args.title,
statement=args.statement,
owner=args.owner,
feature=args.feature,
source_task=args.source_task,
source_doc=args.source_doc,
state=args.state,
conversation_ref=args.conversation_ref,
)
emit(result)
return 0
if action == 'list':
with repo.db() as db:
rows = catalog.requirement_list(db, state=args.state, limit=args.limit)
emit({'status': 'ok', 'requirements': rows})
return 0
if action == 'show':
with repo.db() as db:
row = catalog.requirement_show(db, args.id)
emit({'status': 'ok', 'requirement': row})
return 0
if action == 'search':
with repo.db() as db:
hits = catalog.search(db, args.query, top=args.top)
emit({'status': 'ok', 'hits': hits})
return 0
if action == 'overlap':
with repo.db(True) as db:
scopes = []
try:
task = task_row(repo, args.task)
scopes = json.loads(task['scopes']) if isinstance(task['scopes'], str) else task['scopes']
except Denied:
# Allow pre-register probes; path/scope hard checks skipped without a task row.
pass
result = catalog.probe_overlap(
db,
task=args.task,
statement=args.statement,
requirement=args.requirement,
feature=args.feature,
mutex=args.mutex,
scopes=scopes,
top=args.top,
overlaps_fn=overlaps,
)
emit(result)
return 0
if action == 'ack':
with repo.db(True) as db:
result = catalog.overlap_ack(
db,
task=args.task,
overlap_id=args.overlap,
decision=args.decision,
decided_by=args.decided_by,
reason=args.reason,
authorization_ref=args.authorization_ref,
)
emit(result)
return 0
if action == 'status':
with repo.db() as db:
emit(catalog.catalog_status(db))
return 0
if action == 'drift':
matrix_path = repo.main / '.doc/status/FEATURE-MATRIX.md'
# Prefer clean tip from refs/heads/dev when available.
try:
content = git(repo.main, 'show', 'refs/heads/dev:.doc/status/FEATURE-MATRIX.md')
except Exception:
content = matrix_path.read_text(encoding='utf-8') if matrix_path.is_file() else ''
states = matrix_states(content) if content else {}
with repo.db(True) as db:
result = catalog.catalog_drift(db, matrix_states=states)
event(db, 'catalog', {'event': 'drift', **{k: result[k] for k in ('record_count', 'regressions')}})
emit(result)
return 0
deny('INVALID_REQUEST', f'Unknown catalog action: {action}')
return 1
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)
Reference in New Issue View Git Blame Copy Permalink