WeiCHEN
3337bf4a42
Add whole-doc and row projection with hash CAS, real enqueue→claim→ fake-deploy→SQLite receipt chain, Darwin lsof cwd visibility, and black-box tests. PI image compose remains deferred.
1758 lines
88 KiB
Python
Executable File
1758 lines
88 KiB
Python
Executable File
#!/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 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))
|
|
import agentctl_w4 as w4 # noqa: E402
|
|
|
|
VERSION = 1
|
|
SHARED_DOCS = {'.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'}
|
|
ACTIVE = ('running', 'recovery_required')
|
|
TERMINAL = ('completed', 'failed', 'recovered')
|
|
QUEUE_TERMINAL = ('completed', 'failed', 'cancelled')
|
|
GIT_CONTEXT = ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE', 'GIT_PREFIX')
|
|
QUEUE_SCHEMA = '''
|
|
CREATE TABLE IF NOT EXISTS queue_items(
|
|
id TEXT PRIMARY KEY,
|
|
task TEXT NOT NULL,
|
|
head_sha TEXT NOT NULL,
|
|
expected_dev TEXT NOT NULL,
|
|
priority INTEGER NOT NULL DEFAULT 0,
|
|
state TEXT NOT NULL,
|
|
created REAL NOT NULL,
|
|
updated REAL NOT NULL,
|
|
claim_run TEXT,
|
|
reason TEXT,
|
|
UNIQUE(task, head_sha)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS queue_pending ON queue_items(state, priority DESC, created);
|
|
'''
|
|
|
|
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)
|
|
|
|
|
|
def emit(value, stream=sys.stdout):
|
|
print(json.dumps(value, ensure_ascii=False, sort_keys=True), file=stream, flush=True)
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(value.encode() if isinstance(value, str) else value).hexdigest()
|
|
|
|
|
|
def git(path, *args, index=False, check=True, input=None):
|
|
env = os.environ.copy()
|
|
for key in GIT_CONTEXT:
|
|
if key != 'GIT_INDEX_FILE' or not index:
|
|
env.pop(key, None)
|
|
result = subprocess.run(['git', '-C', str(path), *args], env=env,
|
|
input=input, capture_output=True, text=True)
|
|
if check and result.returncode:
|
|
deny('GIT_ERROR', result.stderr.strip() or 'Git operation failed')
|
|
return result.stdout.rstrip('\n') if check else result
|
|
|
|
|
|
@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 in ('docs:publish', 'git:worktrees', 'git:index') or re.fullmatch(r'(env|git):(dev|test|main|shared)', value):
|
|
return value
|
|
if value.startswith('tree:'):
|
|
return value
|
|
deny('INVALID_RESOURCE', 'Unsupported resource', resource=value)
|
|
|
|
|
|
def overlaps(a, b):
|
|
if a == b:
|
|
return True
|
|
if a.startswith('path:') and b.startswith('path:'):
|
|
return covers(a[5:], b[5:]) or covers(b[5:], a[5:])
|
|
# Publishing protected docs conflicts with direct managed path writers, too.
|
|
if a == 'docs:publish' and b.startswith('path:'):
|
|
return any(covers(b[5:], p) or covers(p, b[5:]) for p in SHARED_DOCS)
|
|
if b == 'docs:publish' and a.startswith('path:'):
|
|
return overlaps(b, a)
|
|
return False
|
|
|
|
|
|
def safe_path(root, rel):
|
|
rel = relative(rel)
|
|
target = root / rel
|
|
if not target.resolve().is_relative_to(root.resolve()):
|
|
deny('PATH_OUT_OF_SCOPE', 'Path escapes the worktree')
|
|
cur = root
|
|
for part in Path(rel).parts:
|
|
cur /= part
|
|
if cur.is_symlink():
|
|
deny('PATH_OUT_OF_SCOPE', 'Symlink paths are not managed')
|
|
if target.exists() and target.is_file() and target.stat().st_nlink > 1:
|
|
deny('PATH_OUT_OF_SCOPE', 'Hardlinked working files are not managed')
|
|
return target
|
|
|
|
|
|
def secret_path(name):
|
|
parts = Path(name).parts
|
|
leaf = Path(name).name
|
|
return ('.locks' in parts or '.git' in parts or (parts and 'data' == parts[0])
|
|
or leaf == '.env' or ('.env.' in leaf and not leaf.endswith('.example'))
|
|
or leaf in ('.env.local', 'phase-secrets.local.json')
|
|
or any(p in ('.secrets', 'node_modules') for p in parts))
|
|
|
|
|
|
def clean_source(path, build=False):
|
|
if not build:
|
|
dirty = bool(git(path, 'status', '--porcelain', '--untracked-files=all'))
|
|
else:
|
|
tracked = git(path, 'status', '--porcelain', '--untracked-files=no')
|
|
untracked = git(path, 'ls-files', '--others', '--exclude-standard', '-z').split('\0')
|
|
# These are the existing nightly producer's non-source outputs. git archive
|
|
# excludes all untracked files, independently of Docker ignore patterns.
|
|
evidence_prefix = '.doc/test-reports/assets/promotion-receipts/'
|
|
dirty = bool(tracked or any(n and not n.startswith(evidence_prefix) for n in untracked))
|
|
if dirty:
|
|
deny('DIRTY_TREE', 'Source worktree must be clean; only generated build evidence is excepted')
|
|
|
|
|
|
class Repo:
|
|
def __init__(self, path):
|
|
self.current = Path(git(Path(path).resolve(), 'rev-parse', '--show-toplevel')).resolve()
|
|
self.common = Path(git(self.current, 'rev-parse', '--path-format=absolute', '--git-common-dir')).resolve()
|
|
rows = self.worktrees()
|
|
if not rows or rows[0].get('bare'):
|
|
deny('UNSUPPORTED_REPOSITORY', 'A non-bare main worktree is required')
|
|
self.main = Path(rows[0]['worktree']).resolve()
|
|
if self.common != (self.main / '.git').resolve():
|
|
deny('UNSUPPORTED_REPOSITORY', 'External Git directories require explicit migration')
|
|
self.root = self.main / '.locks'
|
|
if self.root.is_symlink():
|
|
deny('INVALID_LOCK_ROOT', 'The canonical .locks directory must not be a symlink')
|
|
self.dbfile = self.root / 'state.sqlite'
|
|
|
|
def worktrees(self):
|
|
records = []
|
|
for block in git(self.current, 'worktree', 'list', '--porcelain', '-z').split('\0\0'):
|
|
record = {}
|
|
for field in block.split('\0'):
|
|
if field:
|
|
key, _, value = field.partition(' ')
|
|
record[key] = value
|
|
if record:
|
|
records.append(record)
|
|
return records
|
|
|
|
def identity(self):
|
|
try:
|
|
identity = json.loads((self.root / 'identity.json').read_text())
|
|
except FileNotFoundError:
|
|
deny('NOT_INITIALIZED', 'Run agentctl init explicitly from the main worktree')
|
|
except (ValueError, OSError):
|
|
deny('RECOVERY_REQUIRED', 'Lock identity cannot be read')
|
|
if (identity.get('main') != str(self.main) or identity.get('common') != str(self.common)
|
|
or identity.get('version') != VERSION or not self.dbfile.exists()):
|
|
deny('POLICY_MISMATCH', 'Lock identity, schema or repository topology changed')
|
|
old = self.common / 'agent-control' / 'dev-integration.lock'
|
|
bridge = self.root / 'os' / 'dev-integration.lock'
|
|
if not old.exists() or not bridge.exists() or old.is_symlink() or bridge.is_symlink() or not os.path.samefile(old, bridge):
|
|
deny('RECOVERY_REQUIRED', 'Compatibility lock identity changed; stop and inspect')
|
|
if self.dbfile.is_symlink() or (self.root / 'os').is_symlink():
|
|
deny('INVALID_LOCK_ROOT', 'Coordination state must not use symlink aliases')
|
|
return identity
|
|
|
|
@contextlib.contextmanager
|
|
def db(self, write=False):
|
|
self.identity()
|
|
# mode=rw never silently creates a fresh authority when state is missing.
|
|
db = sqlite3.connect(self.dbfile.as_uri() + '?mode=' + ('rw' if write else 'ro'),
|
|
uri=True, timeout=5, isolation_level=None)
|
|
db.row_factory = sqlite3.Row
|
|
try:
|
|
db.execute('PRAGMA busy_timeout=5000')
|
|
if write:
|
|
db.execute('PRAGMA synchronous=FULL')
|
|
db.execute('BEGIN IMMEDIATE')
|
|
ensure_schema(db)
|
|
yield db
|
|
if write:
|
|
db.commit()
|
|
except BaseException:
|
|
if write:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
def os_file(self, name):
|
|
self.identity()
|
|
path = self.root / 'os' / name
|
|
if path.is_symlink():
|
|
deny('INVALID_LOCK_ROOT', 'Stable OS lock must be a regular file')
|
|
return path
|
|
|
|
|
|
def atomic_write(path, content):
|
|
data = content.encode() if isinstance(content, str) else content
|
|
mode = path.stat().st_mode & 0o777 if path.exists() else 0o600
|
|
fd, tmp = tempfile.mkstemp(prefix='.' + path.name + '.agentctl-', dir=path.parent)
|
|
try:
|
|
os.fchmod(fd, mode)
|
|
with os.fdopen(fd, 'wb') as f:
|
|
f.write(data)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, path)
|
|
directory = os.open(path.parent, os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
if os.path.exists(tmp):
|
|
os.unlink(tmp)
|
|
|
|
|
|
def initialize(repo):
|
|
if repo.current != repo.main:
|
|
deny('MAIN_WORKTREE_REQUIRED', 'Initialize from the main worktree')
|
|
legacy = repo.common / 'agent-control'
|
|
legacy.mkdir(exist_ok=True)
|
|
with (legacy / 'init.lock').open('a+') as initialization:
|
|
fcntl.flock(initialization, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
if (repo.root / 'identity.json').exists():
|
|
identity = repo.identity()
|
|
with repo.db(True) as db:
|
|
db.execute('SELECT count(*) FROM runs').fetchone()
|
|
db.execute('SELECT count(*) FROM queue_items').fetchone()
|
|
emit({'status': 'initialized', 'lock_root': str(repo.root), **identity})
|
|
return
|
|
old = legacy / 'dev-integration.lock'
|
|
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))
|
|
emit({'status': 'initialized', 'lock_root': str(repo.root), **identity})
|
|
|
|
|
|
def ensure_schema(db):
|
|
"""Additive tables for durable queue + W4 deploy/projection; write txn only."""
|
|
for statement in filter(None, (part.strip() for part in QUEUE_SCHEMA.split(';'))):
|
|
db.execute(statement)
|
|
w4.ensure_schema(db)
|
|
|
|
|
|
def task_row(repo, task_id):
|
|
with repo.db() as db:
|
|
row = db.execute('SELECT * FROM tasks WHERE id=?', (task_id,)).fetchone()
|
|
if not row:
|
|
deny('UNKNOWN_TASK', 'Register the task before requesting a lease')
|
|
if row['state'] != 'active':
|
|
deny('TASK_CLOSED', 'This task is no longer active')
|
|
if not Path(row['worktree']).exists() or git(row['worktree'], 'branch', '--show-current') != row['branch']:
|
|
deny('WORKTREE_MISMATCH', 'Registered task branch/worktree changed')
|
|
return dict(row)
|
|
|
|
|
|
def check_resources(repo, task, resources):
|
|
scopes, caps = json.loads(task['scopes']), json.loads(task['capabilities'])
|
|
for value in resources:
|
|
if value.startswith('path:'):
|
|
p = value[5:]
|
|
if p != '.' and secret_path(p):
|
|
deny('PATH_OUT_OF_SCOPE', 'Runtime and Git administration paths are not task source scopes')
|
|
safe_path(Path(task['worktree']), p)
|
|
if not any(covers(s, p) for s in scopes):
|
|
deny('PATH_OUT_OF_SCOPE', 'Resource is outside the registered write scope', resource=value)
|
|
elif value.startswith('tree:'):
|
|
if value != 'tree:' + task['worktree']:
|
|
deny('PATH_OUT_OF_SCOPE', 'Task cannot acquire another worktree index')
|
|
elif value not in caps:
|
|
deny('CAPABILITY_REQUIRED', 'Task lacks this registered capability', resource=value)
|
|
|
|
|
|
def event(db, operation, detail):
|
|
db.execute('INSERT INTO events(created,operation,detail) VALUES(?,?,?)',
|
|
(time.time(), operation, json.dumps(detail, ensure_ascii=False)))
|
|
|
|
|
|
def physical_names(resources):
|
|
names = set()
|
|
for item in resources:
|
|
if item in ('git:dev', 'env:dev'):
|
|
names.add('dev-integration.lock')
|
|
if item == 'git:worktrees':
|
|
names.add('worktrees.lock')
|
|
if item == 'docs:publish' or (item.startswith('path:') and any(covers(item[5:], p) for p in SHARED_DOCS)):
|
|
names.add('shared-writeback.lock')
|
|
if item.startswith('tree:'):
|
|
names.add('tree-' + digest(item) + '.lock')
|
|
return sorted(names)
|
|
|
|
|
|
class Lease:
|
|
def __init__(self, repo, task, resources, request_id, fingerprint, wait=0, mode='X', kind='command'):
|
|
self.repo, self.task = repo, task
|
|
self.resources = sorted(set('tree:' + task['worktree'] if r == 'git:index' else resource(r) for r in resources))
|
|
check_resources(repo, task, self.resources)
|
|
self.request_id = request_id or str(uuid.uuid4())
|
|
if not isinstance(self.request_id, str) or not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', self.request_id):
|
|
deny('INVALID_REQUEST', 'Invalid request id')
|
|
self.fingerprint, self.wait, self.mode, self.kind = fingerprint, max(0, min(wait, 60)), mode, kind
|
|
self.handles, self.row, self.replay = [], None, False
|
|
|
|
def unlock(self):
|
|
for handle in reversed(self.handles):
|
|
handle.close()
|
|
self.handles = []
|
|
|
|
def __enter__(self):
|
|
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:
|
|
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') 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')
|
|
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 --')
|
|
cwd = Path(args.cwd).resolve() if args.cwd else Path(task['worktree'])
|
|
known = {Path(r['worktree']).resolve() for r in repo.worktrees()}
|
|
if cwd not in known or cwd not in (Path(task['worktree']), repo.main):
|
|
deny('WORKTREE_MISMATCH', 'Command cwd must be this task worktree or authorized main operation')
|
|
if cwd == repo.main and not any(r.startswith('env:') or r in ('git:dev','git:test','git:main','git:worktrees') for r in args.resource):
|
|
deny('MAIN_WORKTREE_REQUIRED', 'Main checkout is reserved for registered integration/environment operations')
|
|
if cwd == repo.main and 'git:dev' in args.resource:
|
|
deny('USE_INTEGRATE', 'Protected dev updates use agentctl integrate, not an arbitrary main-worktree command')
|
|
fingerprint = digest(json.dumps([command, str(cwd), sorted(args.resource), args.shared]))
|
|
with Lease(repo, task, args.resource, args.request_id, fingerprint, args.wait,
|
|
'S' if args.shared else 'X') as lease:
|
|
if lease.replay:
|
|
emit({'status': 'replayed', 'run_id': lease.request_id, 'exit_code': lease.row['exit_code']}, sys.stderr)
|
|
return lease.row['exit_code'] or 0
|
|
env = os.environ.copy()
|
|
env.update(AGENTCTL_RUN_ID=lease.request_id, AGENTCTL_TASK_ID=args.task,
|
|
AGENTCTL_GENERATION=str(lease.row['generation']), AGENTCTL_EPOCH=repo.identity()['epoch'])
|
|
emit({'status': 'running', 'run_id': lease.request_id, 'generation': lease.row['generation']}, sys.stderr)
|
|
code, remaining = supervise(repo, lease, command, cwd, env)
|
|
recovery = remaining or bool(code and any(r.startswith(('env:', 'git:dev')) for r in args.resource))
|
|
lease.finish(code, recovery)
|
|
if recovery:
|
|
emit({'status': 'recovery_required', 'run_id': lease.request_id, 'child_group_alive': remaining}, sys.stderr)
|
|
return code or (1 if remaining else 0)
|
|
|
|
|
|
def doc_rows(content, record_id):
|
|
matches = []
|
|
for number, line in enumerate(content.splitlines()):
|
|
if line.lstrip(' ').startswith('|'):
|
|
cells = [cell.strip() for cell in line.strip().strip('|').split('|')]
|
|
if record_id in cells[:2]:
|
|
matches.append((number, line))
|
|
if len(matches) > 1:
|
|
deny('AMBIGUOUS_RECORD', 'Record id is not unique in this document')
|
|
return matches
|
|
|
|
|
|
def table_cells(line):
|
|
return [cell.strip() for cell in re.split(r'(?<!\\)\|', line.strip().strip('|'))]
|
|
|
|
|
|
def matrix_states(content):
|
|
states, status_index, width = {}, None, None
|
|
for line in content.splitlines():
|
|
if not line.lstrip(' ').startswith('|'):
|
|
if status_index is not None:
|
|
break
|
|
continue
|
|
cells = table_cells(line)
|
|
if status_index is None:
|
|
if cells[0].casefold() not in ('模块', 'module', 'feature'):
|
|
continue
|
|
labels = [c.strip('*').casefold() for c in cells]
|
|
columns = [i for i,c in enumerate(labels) if c in ('状态','state','status')]
|
|
if len(columns) != 1:
|
|
deny('INVALID_DOCUMENT', 'Matrix needs one explicit status column')
|
|
status_index, width = columns[0], len(cells)
|
|
continue
|
|
if all(re.fullmatch(r':?-+:?', cell.replace(' ', '')) for cell in cells):
|
|
continue
|
|
if len(cells) != width or cells[0] in states:
|
|
deny('INVALID_DOCUMENT', 'Matrix records must have unique ids and match its columns')
|
|
value = cells[status_index]
|
|
if value not in ('done','partial','planned','deferred','blocked','removed'):
|
|
deny('INVALID_STATE', 'Matrix status must use an unformatted canonical state')
|
|
states[cells[0]] = value
|
|
if status_index is None:
|
|
deny('INVALID_DOCUMENT', 'Matrix table header is missing')
|
|
return states
|
|
|
|
|
|
def docs_snapshot(repo, args):
|
|
if args.file not in SHARED_DOCS:
|
|
deny('PATH_OUT_OF_SCOPE', 'Only the two shared governance documents are managed')
|
|
content = safe_path(repo.current, args.file).read_text()
|
|
row = [(0, content)] if args.record == '@document' else doc_rows(content, args.record)
|
|
emit({'file': args.file, 'record_id': args.record, 'sha256': digest(row[0][1]) if row else None,
|
|
'row': row[0][1] if row else None, 'document_sha256': digest(content)})
|
|
|
|
|
|
def docs_apply(repo, args):
|
|
task = task_row(repo, args.task)
|
|
if repo.current != Path(task['worktree']) or repo.current == repo.main:
|
|
deny('WORKTREE_MISMATCH', 'Shared documents are edited only in the task worktree')
|
|
request = json.loads(Path(args.request).read_text())
|
|
if not isinstance(request, dict) or not isinstance(request.get('changes'), list) or not request['changes'] or any(not isinstance(c, dict) for c in request['changes']):
|
|
deny('INVALID_REQUEST', 'Expected a nonempty changes array')
|
|
resources = ['docs:publish']
|
|
for change in request['changes']:
|
|
if not isinstance(change.get('file'), str) or change['file'] not in SHARED_DOCS:
|
|
deny('PATH_OUT_OF_SCOPE', 'Unknown shared document')
|
|
resources.append('path:' + change['file'])
|
|
with Lease(repo, task, resources, request.get('request_id'), digest(json.dumps(request, sort_keys=True))) as lease:
|
|
if lease.replay:
|
|
emit({'status': 'replayed', 'run_id': lease.request_id, 'exit_code': lease.row['exit_code']})
|
|
return lease.row['exit_code'] or 0
|
|
before, after, seen = {}, {}, set()
|
|
for change in request['changes']:
|
|
filename, record_id = change['file'], change.get('record_id')
|
|
if not isinstance(record_id, str) or not record_id or (filename, record_id) in seen:
|
|
deny('INVALID_REQUEST', 'Each record must have a unique nonempty id')
|
|
seen.add((filename, record_id))
|
|
if record_id == '@document' and 'docs:admin' not in json.loads(task['capabilities']):
|
|
deny('CAPABILITY_REQUIRED', 'Document layout changes require docs:admin')
|
|
if filename.endswith('CURRENT.md') and record_id != args.task and 'docs:admin' not in json.loads(task['capabilities']):
|
|
deny('PATH_OUT_OF_SCOPE', 'A task can update only its own CURRENT record')
|
|
if record_id in ('holder', 'idle', 'Cursor', 'Codex', '-'):
|
|
deny('PATH_OUT_OF_SCOPE', 'Use task ids; slot ownership is not a document edit')
|
|
path = safe_path(repo.current, filename)
|
|
if filename not in before:
|
|
before[filename] = path.read_text()
|
|
base = git(repo.current, 'show', 'HEAD:' + filename, check=False)
|
|
with repo.db() as db:
|
|
prior = db.execute('SELECT hash FROM receipts WHERE task=? AND file=?', (args.task, filename)).fetchone()
|
|
if base.returncode != 0 or (before[filename] != base.stdout and (not prior or prior['hash'] != digest(before[filename]))):
|
|
deny('UNMANAGED_DOCUMENT_CHANGE', 'Shared document differs from HEAD and its last managed receipt; preserve and reconcile first')
|
|
after[filename] = before[filename]
|
|
existing = [(0, before[filename])] if record_id == '@document' else doc_rows(before[filename], record_id)
|
|
expected = digest(existing[0][1]) if existing else None
|
|
if 'expected_sha256' not in change or change['expected_sha256'] != expected:
|
|
deny('BASE_CHANGED', 'Record changed; read the current snapshot before editing', record_id=record_id)
|
|
replacement = change.get('replacement')
|
|
if record_id == '@document':
|
|
if not isinstance(replacement, str):
|
|
deny('INVALID_REQUEST', 'Expected full document text')
|
|
if any(other.get('file') == filename and other is not change for other in request['changes']):
|
|
deny('INVALID_REQUEST', 'Whole-document operation cannot mix row edits for the same file')
|
|
if filename.endswith('CURRENT.md'):
|
|
def slot(text):
|
|
match = re.search(r'(?m)^## 集成槽位\s*\n(.*?)(?=^## |\Z)', text, re.S | re.M)
|
|
return match.group(1) if match else None
|
|
if slot(replacement) != slot(before[filename]):
|
|
deny('PATH_OUT_OF_SCOPE', 'Document admin cannot overwrite integration slot ownership')
|
|
# A layout update cannot bypass the feature-completion evidence rule.
|
|
if filename.endswith('FEATURE-MATRIX.md'):
|
|
old_states = matrix_states(before[filename])
|
|
new_states = matrix_states(replacement)
|
|
if any(value == 'done' and old_states.get(key) != 'done' for key,value in new_states.items()):
|
|
deny('EVIDENCE_REQUIRED', 'Change capability completion using a reviewed row update')
|
|
after[filename] = replacement
|
|
continue
|
|
if not isinstance(replacement, str) or len(replacement.splitlines()) != 1 or not replacement.startswith('|'):
|
|
deny('INVALID_REQUEST', 'Replacement must be one Markdown table row')
|
|
cells = table_cells(replacement)
|
|
if record_id not in cells[:2]:
|
|
deny('INVALID_REQUEST', 'Replacement must preserve the record id')
|
|
if filename.endswith('FEATURE-MATRIX.md') and 'done' in cells and matrix_states(before[filename]).get(record_id) != 'done':
|
|
evidence = change.get('evidence', [])
|
|
if 'docs:approve' not in json.loads(task['capabilities']) or not isinstance(evidence, list) or not evidence or any(not isinstance(e, str) for e in evidence):
|
|
deny('EVIDENCE_REQUIRED', 'Capability completion requires approval capability and committed evidence')
|
|
for evidence_file in evidence:
|
|
safe_path(repo.current, evidence_file)
|
|
git(repo.current, 'cat-file', '-e', 'HEAD:' + relative(evidence_file))
|
|
lines = after[filename].splitlines()
|
|
live = doc_rows(after[filename], record_id)
|
|
if live:
|
|
lines[live[0][0]] = replacement
|
|
else:
|
|
header = '| owner | task |' if filename.endswith('CURRENT.md') else '| 模块 |'
|
|
positions = [i for i, line in enumerate(lines) if line.startswith(header)]
|
|
if len(positions) != 1:
|
|
deny('INVALID_DOCUMENT', 'Target table header is absent or ambiguous')
|
|
lines.insert(positions[0] + 2, replacement)
|
|
after[filename] = '\n'.join(lines) + '\n'
|
|
for name, content in after.items():
|
|
if name.endswith('FEATURE-MATRIX.md'):
|
|
matrix_states(content)
|
|
payload = {name: {'before': before[name], 'after': after[name]} for name in before}
|
|
with repo.db(True) as db:
|
|
db.execute("INSERT INTO documents VALUES(?,?,?,'prepared',?)",
|
|
(lease.request_id, args.task, str(repo.current), json.dumps(payload)))
|
|
for name, values in payload.items():
|
|
path = safe_path(repo.current, name)
|
|
if path.read_text() != values['before']:
|
|
deny('BASE_CHANGED', 'Document changed outside the managed entrypoint; recovery required')
|
|
atomic_write(path, values['after'])
|
|
with repo.db(True) as db:
|
|
for name, values in payload.items():
|
|
db.execute('INSERT OR REPLACE INTO receipts VALUES(?,?,?)', (args.task, name, digest(values['after'])))
|
|
db.execute("UPDATE documents SET state='applied' WHERE run=?", (lease.request_id,))
|
|
lease.finish()
|
|
emit({'status': 'applied', 'run_id': lease.request_id,
|
|
'files': {name: digest(values['after']) for name, values in payload.items()}})
|
|
return 0
|
|
|
|
|
|
def guard(repo, args):
|
|
if args.kind == 'build':
|
|
sha = git(repo.current, 'rev-parse', '--verify', args.commit + '^{commit}')
|
|
if git(repo.current, 'rev-parse', 'HEAD') != sha:
|
|
deny('BASE_CHANGED', 'Build source HEAD does not match the requested commit')
|
|
clean_source(repo.current, build=True)
|
|
elif args.kind == 'resource':
|
|
inherited(repo, args.resource)
|
|
elif args.kind == 'env':
|
|
if repo.current != repo.main or git(repo.main, 'branch', '--show-current') != 'dev':
|
|
deny('MAIN_WORKTREE_REQUIRED', 'Shared services run only from the main dev worktree')
|
|
targets = ['dev', 'test', 'main', 'shared'] if args.target == 'all' else [args.target]
|
|
for target in targets:
|
|
inherited(repo, 'env:' + target)
|
|
clean_source(repo.main)
|
|
emit({'status': 'allowed'})
|
|
|
|
|
|
def worktree_check(repo, path):
|
|
target = Path(path).resolve()
|
|
with repo.db() as db:
|
|
tasks = db.execute('SELECT id,state FROM tasks WHERE worktree=?', (str(target),)).fetchall()
|
|
for task in tasks:
|
|
row = db.execute("SELECT id FROM runs WHERE task=? AND state IN ('running','recovery_required','queued')", (task['id'],)).fetchone()
|
|
if row:
|
|
deny('LOCK_BUSY', 'Worktree has a live or unresolved operation', run_id=row['id'])
|
|
if task['state'] == 'active':
|
|
deny('TASK_ACTIVE', 'Close the registered task before removing its worktree')
|
|
emit({'status': 'allowed', 'path': str(target)})
|
|
|
|
|
|
def recover(repo, args):
|
|
with repo.db() as db:
|
|
row = db.execute('SELECT * FROM runs WHERE id=?', (args.run,)).fetchone()
|
|
document = db.execute('SELECT * FROM documents WHERE run=?', (args.run,)).fetchone()
|
|
if not row:
|
|
deny('UNKNOWN_OPERATION', 'Unknown run id')
|
|
if row['state'] in TERMINAL:
|
|
emit({'status': row['state'], 'run_id': args.run})
|
|
return
|
|
# 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')
|
|
handles = []
|
|
try:
|
|
for name in physical_names(json.loads(row['resources'])):
|
|
handle = repo.os_file(name).open('a+')
|
|
handles.append(handle)
|
|
try:
|
|
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
deny('LOCK_BUSY', 'A physical operation is still in progress')
|
|
if document and document['state'] == 'prepared':
|
|
root = Path(document['worktree'])
|
|
payload = json.loads(document['payload'])
|
|
for name, values in payload.items():
|
|
if safe_path(root, name).read_text() not in (values['before'], values['after']):
|
|
deny('BASE_CHANGED', 'Recovery would overwrite third-party content; preserve the worktree')
|
|
for name, values in payload.items():
|
|
if safe_path(root, name).read_text() != values['after']:
|
|
atomic_write(safe_path(root, name), values['after'])
|
|
with repo.db(True) as db:
|
|
for name, values in payload.items():
|
|
db.execute('INSERT OR REPLACE INTO receipts VALUES(?,?,?)', (row['task'], name, digest(values['after'])))
|
|
db.execute("UPDATE documents SET state='applied' WHERE run=?", (args.run,))
|
|
with repo.db(True) as db:
|
|
db.execute("UPDATE runs SET state='recovered',reason=?,exit_code=? WHERE id=?", (args.reason, 0 if document else 1, args.run))
|
|
event(db, args.run, {'event': 'operator_recovery', 'reason': args.reason})
|
|
emit({'status': 'recovered', 'run_id': args.run})
|
|
finally:
|
|
for handle in reversed(handles):
|
|
handle.close()
|
|
|
|
|
|
def hook_check(repo, merge=False):
|
|
if repo.current == repo.main or git(repo.current, 'branch', '--show-current') in ('dev', 'test', 'main', ''):
|
|
deny('PROTECTED_BRANCH', 'Direct development commits require a registered feature/fix worktree')
|
|
run, task = inherited(repo, 'git:index')
|
|
if str(repo.current) != task['worktree']:
|
|
deny('WORKTREE_MISMATCH', 'Commit worktree differs from registered task')
|
|
names = git(repo.current, 'diff', '--cached', '--name-only', '-z', index=True).split('\0')
|
|
scopes = json.loads(task['scopes'])
|
|
run_resources = json.loads(run['resources'])
|
|
merge_has_dev_lease = merge and 'git:dev' in run_resources
|
|
for name in filter(None, names):
|
|
if secret_path(name):
|
|
deny('PROTECTED_PATH', 'Sensitive/runtime paths cannot enter a task commit', path=name)
|
|
if not any(covers(scope, name) for scope in scopes):
|
|
deny('PATH_OUT_OF_SCOPE', 'Staged file is outside task scope', path=name)
|
|
safe_path(repo.current, name)
|
|
if not merge_has_dev_lease and not any(r.startswith('path:') and covers(r[5:], name) for r in run_resources):
|
|
deny('LEASE_REQUIRED', 'Staged path is not covered by this operation lease', path=name)
|
|
# Integration merge is separately fenced by git:dev and exact candidate validation.
|
|
if name in SHARED_DOCS and not merge_has_dev_lease:
|
|
value = git(repo.current, 'show', ':' + name, index=True, check=False)
|
|
staged = value.stdout if value.returncode == 0 else ''
|
|
with repo.db() as db:
|
|
receipt = db.execute('SELECT hash FROM receipts WHERE task=? AND file=?', (task['id'], name)).fetchone()
|
|
if not receipt or receipt['hash'] != digest(staged):
|
|
deny('DOCUMENT_RECEIPT_REQUIRED', 'Use docs apply for the staged shared document', path=name)
|
|
|
|
|
|
def admit_candidate(repo, task, candidate, expected):
|
|
parents = git(repo.main, 'rev-list', '--parents', '-n', '1', candidate).split()[1:]
|
|
if len(parents) != 2 or parents[0] != expected:
|
|
deny('INVALID_CANDIDATE', 'Candidate must have two parents and the frozen target as its first parent')
|
|
if git(task['worktree'], 'rev-parse', 'HEAD') != candidate:
|
|
deny('BASE_CHANGED', 'Candidate is not the owning task HEAD')
|
|
with repo.db() as db:
|
|
found = db.execute('SELECT sha FROM validations WHERE task=? AND sha=?', (task['id'], candidate)).fetchone()
|
|
if not found:
|
|
deny('VALIDATION_REQUIRED', 'Validate this exact candidate with its owning task first')
|
|
|
|
|
|
def hook_ref(repo, state):
|
|
lines = sys.stdin.read().splitlines()
|
|
if state != 'prepared':
|
|
return
|
|
for line in lines:
|
|
old, new, ref = line.split()
|
|
if ref not in ('refs/heads/dev', 'refs/heads/test', 'refs/heads/main'):
|
|
continue
|
|
target = ref.rsplit('/',1)[1]
|
|
run, task = inherited(repo, 'git:' + target)
|
|
if set(new) == {'0'} or set(old) == {'0'}:
|
|
deny('PROTECTED_BRANCH', 'Protected branch creation/deletion is not a task operation')
|
|
admit_candidate(repo, task, new, old)
|
|
if target == 'dev' and (run['kind'] != 'integrate' or run['fingerprint'] != digest(json.dumps([new, old]))):
|
|
deny('USE_INTEGRATE', 'dev requires the controller operation bound to this old/new pair')
|
|
for name in git(repo.main, 'diff', '--name-only', old, new).splitlines():
|
|
if secret_path(name):
|
|
deny('PROTECTED_PATH', 'Candidate contains a sensitive/runtime path')
|
|
|
|
|
|
def install_hooks(repo):
|
|
if repo.current != repo.main:
|
|
deny('MAIN_WORKTREE_REQUIRED', 'Install hooks from the main worktree after integration')
|
|
repo.identity()
|
|
configured = git(repo.main, 'config', '--get', 'core.hooksPath', check=False)
|
|
if configured.returncode == 0:
|
|
deny('EXISTING_HOOKS', 'Custom hooksPath detected; integrate hooks explicitly instead of overriding it')
|
|
hooks = repo.common / 'hooks'
|
|
hooks.mkdir(exist_ok=True)
|
|
commands = {'pre-commit':'hooks check', 'pre-merge-commit':'hooks check --merge', 'reference-transaction':'hooks ref "$1"'}
|
|
for name in commands:
|
|
p = hooks / name
|
|
if p.exists() and '# agentctl-managed-v1' not in p.read_text():
|
|
deny('EXISTING_HOOKS', 'Existing hook will not be overwritten', hook=name)
|
|
import shlex
|
|
entry = shlex.quote(str(repo.main/'scripts/agentctl.py'))
|
|
for name, command in commands.items():
|
|
p = hooks/name
|
|
atomic_write(p, '#!/bin/sh\n# agentctl-managed-v1\nexec python3 '+entry+' '+command+'\n')
|
|
p.chmod(0o755)
|
|
emit({'status':'installed','hooks':list(commands)})
|
|
|
|
|
|
def validate(repo, args):
|
|
task = task_row(repo, args.task)
|
|
if repo.current != Path(task['worktree']):
|
|
deny('WORKTREE_MISMATCH', 'Validate from the registered task worktree')
|
|
command = args.command[1:] if args.command[:1] == ['--'] else args.command
|
|
if not command:
|
|
deny('INVALID_REQUEST','Validation requires an explicit command')
|
|
# Read all source for exact-HEAD evidence, not just the task's changed paths.
|
|
resources = ['path:' + scope for scope in json.loads(task['scopes'])] + ['git:index']
|
|
with Lease(repo, task, resources, None, digest(json.dumps(['validate', command])), mode='S') as lease:
|
|
clean_source(repo.current)
|
|
sha = git(repo.current, 'rev-parse', 'HEAD')
|
|
with repo.db(True) as db:
|
|
db.execute('DELETE FROM validations WHERE task=? AND sha=?', (args.task, sha))
|
|
code, orphan = supervise(repo, lease, command, repo.current)
|
|
if code or orphan:
|
|
lease.finish(code or 1, orphan)
|
|
return code or 1
|
|
clean_source(repo.current)
|
|
if git(repo.current, 'rev-parse', 'HEAD') != sha:
|
|
deny('BASE_CHANGED','Validation changed HEAD')
|
|
with repo.db(True) as db:
|
|
db.execute('INSERT OR REPLACE INTO validations VALUES(?,?,?,?)',
|
|
(args.task,sha,digest(json.dumps(command)),time.time()))
|
|
lease.finish()
|
|
emit({'status':'validated','sha':sha,'task':args.task,'run_id':lease.request_id})
|
|
return 0
|
|
|
|
|
|
def integrate(repo, args):
|
|
task = task_row(repo, args.task)
|
|
candidate = git(task['worktree'], 'rev-parse', 'HEAD')
|
|
expected = git(repo.main, 'rev-parse', args.expected_dev+'^{commit}')
|
|
admit_candidate(repo, task, candidate, expected)
|
|
with Lease(repo,task,['git:dev','git:index'],args.request_id,digest(json.dumps([candidate,expected])),kind='integrate') as lease:
|
|
if lease.replay:
|
|
emit({'status':'replayed','run_id':lease.request_id,'exit_code':lease.row['exit_code']})
|
|
return lease.row['exit_code'] or 0
|
|
admit_candidate(repo, task, candidate, expected)
|
|
clean_source(repo.main)
|
|
clean_source(Path(task['worktree']))
|
|
if git(repo.main,'branch','--show-current') != 'dev' or git(repo.main,'rev-parse','dev') != expected:
|
|
deny('BASE_CHANGED','Main dev moved; rebuild and validate the candidate')
|
|
env=os.environ.copy();env.update(AGENTCTL_RUN_ID=lease.request_id,AGENTCTL_TASK_ID=args.task,
|
|
AGENTCTL_GENERATION=str(lease.row['generation']),AGENTCTL_EPOCH=repo.identity()['epoch'])
|
|
code, orphan = supervise(repo, lease, ['git','-C',str(repo.main),'merge','--ff-only',candidate], repo.main, env)
|
|
lease.finish(code,bool(code) or orphan)
|
|
emit({'status':'merged' if code==0 else 'recovery_required','sha':candidate,'run_id':lease.request_id})
|
|
return code
|
|
|
|
|
|
def queue_row(row):
|
|
return dict(row) if row is not None else None
|
|
|
|
|
|
def queue_enqueue(repo, args):
|
|
task = task_row(repo, args.task)
|
|
if repo.current != Path(task['worktree']):
|
|
deny('WORKTREE_MISMATCH', 'Enqueue from the registered task worktree')
|
|
head = git(task['worktree'], 'rev-parse', 'HEAD')
|
|
if args.head and git(task['worktree'], 'rev-parse', args.head + '^{commit}') != head:
|
|
deny('BASE_CHANGED', 'Requested head is not the owning task HEAD')
|
|
expected = git(repo.main, 'rev-parse', args.expected_dev + '^{commit}')
|
|
admit_candidate(repo, task, head, expected)
|
|
item_id = args.request_id or str(uuid.uuid4())
|
|
if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', item_id):
|
|
deny('INVALID_REQUEST', 'Invalid queue request id')
|
|
now = time.time()
|
|
with repo.db(True) as db:
|
|
existing = db.execute(
|
|
"SELECT * FROM queue_items WHERE task=? AND head_sha=? AND state NOT IN ('cancelled')",
|
|
(args.task, head)).fetchone()
|
|
if existing:
|
|
if existing['state'] in ('pending', 'claimed') and existing['expected_dev'] == expected:
|
|
emit({'status': 'duplicate', 'item': queue_row(existing)})
|
|
return 0
|
|
deny('QUEUE_CONFLICT', 'A non-cancelled queue item already covers this task head',
|
|
item=queue_row(existing))
|
|
if db.execute('SELECT id FROM queue_items WHERE id=?', (item_id,)).fetchone():
|
|
deny('REQUEST_ID_REUSED', 'Queue request id already exists')
|
|
db.execute(
|
|
'''INSERT INTO queue_items(id,task,head_sha,expected_dev,priority,state,created,updated,claim_run,reason)
|
|
VALUES(?,?,?,?,?,'pending',?,?,NULL,NULL)''',
|
|
(item_id, args.task, head, expected, max(0, int(args.priority)), now, now))
|
|
event(db, item_id, {'event': 'queue_enqueued', 'task': args.task, 'head_sha': head})
|
|
row = db.execute('SELECT * FROM queue_items WHERE id=?', (item_id,)).fetchone()
|
|
emit({'status': 'enqueued', 'item': queue_row(row)})
|
|
return 0
|
|
|
|
|
|
def queue_list(repo, args):
|
|
states = None if args.all else ('pending', 'claimed')
|
|
with repo.db(True) as db:
|
|
if states:
|
|
rows = db.execute(
|
|
"SELECT * FROM queue_items WHERE state IN (?,?) ORDER BY priority DESC, created ASC",
|
|
states).fetchall()
|
|
else:
|
|
rows = db.execute('SELECT * FROM queue_items ORDER BY created ASC').fetchall()
|
|
emit({'status': 'ok', 'items': [queue_row(r) for r in rows]})
|
|
return 0
|
|
|
|
|
|
def queue_cancel(repo, args):
|
|
task_row(repo, args.task)
|
|
with repo.db(True) as db:
|
|
row = db.execute('SELECT * FROM queue_items WHERE id=?', (args.id,)).fetchone()
|
|
if not row:
|
|
deny('UNKNOWN_QUEUE_ITEM', 'Unknown queue item id')
|
|
if row['task'] != args.task:
|
|
deny('PATH_OUT_OF_SCOPE', 'Queue item belongs to another task')
|
|
if row['state'] in QUEUE_TERMINAL:
|
|
emit({'status': 'replayed', 'item': queue_row(row)})
|
|
return 0
|
|
if row['state'] == 'claimed':
|
|
deny('LOCK_BUSY', 'Queue item is claimed by an in-flight schedule; recover or wait')
|
|
db.execute("UPDATE queue_items SET state='cancelled',reason=?,updated=? WHERE id=?",
|
|
(args.reason, time.time(), args.id))
|
|
event(db, args.id, {'event': 'queue_cancelled', 'reason': args.reason})
|
|
row = db.execute('SELECT * FROM queue_items WHERE id=?', (args.id,)).fetchone()
|
|
emit({'status': 'cancelled', 'item': queue_row(row)})
|
|
return 0
|
|
|
|
|
|
def queue_schedule(repo, args):
|
|
"""Claim the next durable queue item and run integrate once (or until drain/stop)."""
|
|
if repo.current != repo.main or git(repo.main, 'branch', '--show-current') != 'dev':
|
|
deny('MAIN_WORKTREE_REQUIRED', 'Schedule integrations only from the main dev worktree')
|
|
processed, results = 0, []
|
|
while True:
|
|
with repo.db(True) as db:
|
|
item = db.execute(
|
|
"SELECT * FROM queue_items WHERE state='pending' ORDER BY priority DESC, created ASC LIMIT 1"
|
|
).fetchone()
|
|
if not item:
|
|
break
|
|
db.execute("UPDATE queue_items SET state='claimed',updated=?,claim_run=? WHERE id=? AND state='pending'",
|
|
(time.time(), None, item['id']))
|
|
claimed = db.execute('SELECT * FROM queue_items WHERE id=?', (item['id'],)).fetchone()
|
|
item = queue_row(claimed)
|
|
try:
|
|
admit_candidate(repo, task_row(repo, item['task']), item['head_sha'], item['expected_dev'])
|
|
except Denied as exc:
|
|
with repo.db(True) as db:
|
|
db.execute("UPDATE queue_items SET state='failed',reason=?,updated=? WHERE id=?",
|
|
(exc.code + ': ' + exc.message, time.time(), item['id']))
|
|
event(db, item['id'], {'event': 'queue_failed', 'code': exc.code})
|
|
results.append({'id': item['id'], 'status': 'failed', 'code': exc.code, 'message': exc.message})
|
|
processed += 1
|
|
if not args.drain:
|
|
break
|
|
continue
|
|
request_id = 'queue-' + item['id']
|
|
integrate_args = argparse.Namespace(task=item['task'], expected_dev=item['expected_dev'],
|
|
request_id=request_id)
|
|
try:
|
|
code = integrate(repo, integrate_args)
|
|
except Denied as exc:
|
|
with repo.db(True) as db:
|
|
state = 'pending' if exc.code == 'LOCK_BUSY' else 'failed'
|
|
db.execute("UPDATE queue_items SET state=?,reason=?,updated=?,claim_run=NULL WHERE id=?",
|
|
(state, exc.code + ': ' + exc.message, time.time(), item['id']))
|
|
event(db, item['id'], {'event': 'queue_rescheduled' if state == 'pending' else 'queue_failed',
|
|
'code': exc.code})
|
|
results.append({'id': item['id'], 'status': state, 'code': exc.code, 'message': exc.message})
|
|
processed += 1
|
|
if exc.code == 'LOCK_BUSY' or not args.drain:
|
|
break
|
|
continue
|
|
with repo.db(True) as db:
|
|
if code == 0:
|
|
db.execute("UPDATE queue_items SET state='completed',updated=?,claim_run=?,reason=NULL WHERE id=?",
|
|
(time.time(), request_id, item['id']))
|
|
event(db, item['id'], {'event': 'queue_completed', 'run_id': request_id})
|
|
results.append({'id': item['id'], 'status': 'completed', 'run_id': request_id})
|
|
else:
|
|
db.execute("UPDATE queue_items SET state='failed',updated=?,claim_run=?,reason=? WHERE id=?",
|
|
(time.time(), request_id, 'integrate exit ' + str(code), item['id']))
|
|
event(db, item['id'], {'event': 'queue_failed', 'exit_code': code})
|
|
results.append({'id': item['id'], 'status': 'failed', 'exit_code': code, 'run_id': request_id})
|
|
processed += 1
|
|
if not args.drain:
|
|
break
|
|
emit({'status': 'ok', 'processed': processed, 'results': results})
|
|
return 0 if all(r.get('status') == 'completed' for r in results) or not results else 1
|
|
|
|
|
|
def docs_project(repo, args):
|
|
"""Project 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')",
|
|
(args.task,)).fetchall()]
|
|
queue_state = pending[0]['state'] if pending else 'idle'
|
|
status = args.status or ('queued' if pending else 'in_progress')
|
|
worktree_path = Path(task['worktree'])
|
|
try:
|
|
worktree_cell = str(worktree_path.relative_to(repo.main))
|
|
except ValueError:
|
|
worktree_cell = str(worktree_path)
|
|
|
|
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 = (
|
|
f"| {task['owner']} | {task['id']} | {task['branch']} | {worktree_cell} "
|
|
f"| {args.mutex or '-'} | {status} |"
|
|
)
|
|
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,
|
|
'snapshot_time': time.time(),
|
|
'dev_sha': git(repo.main, 'rev-parse', 'dev'),
|
|
'mode': mode,
|
|
}
|
|
request = {
|
|
'request_id': args.request_id or (args.task + '-project-1'),
|
|
'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 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')",
|
|
(task['id'],)).fetchone()
|
|
if row:
|
|
busy.append(dict(row))
|
|
if any(t['state'] == 'active' for t in tasks):
|
|
gate('registry_task', 'fail', 'active registered task owns this worktree', tasks=tasks)
|
|
else:
|
|
gate('registry_task', 'pass', tasks=tasks)
|
|
if busy:
|
|
gate('registry_runs', 'fail', 'live or unresolved operation', runs=busy)
|
|
else:
|
|
gate('registry_runs', 'pass')
|
|
|
|
key = digest(str(target))
|
|
lock_path = repo.root / 'os' / f'worktree-{key}.lock'
|
|
lifecycle = 'unknown'
|
|
if lock_path.exists():
|
|
try:
|
|
with lock_path.open('a+') as handle:
|
|
try:
|
|
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
fcntl.flock(handle, fcntl.LOCK_UN)
|
|
lifecycle = 'free'
|
|
except BlockingIOError:
|
|
lifecycle = 'held'
|
|
except OSError as exc:
|
|
lifecycle = f'error:{exc}'
|
|
else:
|
|
lifecycle = 'absent'
|
|
gate('lifecycle_lock', 'fail' if lifecycle == 'held' else ('pass' if lifecycle in ('free', 'absent') else 'unknown'),
|
|
lifecycle)
|
|
|
|
process = cleanup_process_gate(target)
|
|
gate('process_cwd', 'pass' if process['status'] == 'clear' else (
|
|
'fail' if process['status'] == 'blocked' else 'unknown'), process.get('reason'), **process)
|
|
|
|
allowed = all(g['status'] == 'pass' for g in gates)
|
|
emit({
|
|
'status': 'allowed' if allowed else 'blocked',
|
|
'path': str(target),
|
|
'cleanup_allowed': allowed,
|
|
'gates': gates,
|
|
})
|
|
return 0
|
|
|
|
|
|
def parser():
|
|
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--repo',default='.')
|
|
sub=p.add_subparsers(dest='action',required=True)
|
|
for name in ('init','status','doctor'):sub.add_parser(name)
|
|
task=sub.add_parser('task').add_subparsers(dest='task_action',required=True)
|
|
reg=task.add_parser('register');reg.add_argument('--id',required=True);reg.add_argument('--owner',required=True);reg.add_argument('--scope',action='append',required=True);reg.add_argument('--capability',action='append',default=[])
|
|
close=task.add_parser('close');close.add_argument('--id',required=True);close.add_argument('--status',choices=['done','cancelled'],default='done')
|
|
w=sub.add_parser('with');w.add_argument('--task',required=True);w.add_argument('--resource',action='append',required=True);w.add_argument('--shared',action='store_true');w.add_argument('--request-id');w.add_argument('--wait',type=float,default=0);w.add_argument('--cwd');w.add_argument('command',nargs=argparse.REMAINDER)
|
|
g=sub.add_parser('guard').add_subparsers(dest='kind',required=True)
|
|
r=g.add_parser('resource');r.add_argument('resource')
|
|
e=g.add_parser('env');e.add_argument('--target',choices=['dev','test','main','shared','all'],required=True)
|
|
b=g.add_parser('build');b.add_argument('--commit',required=True)
|
|
check=sub.add_parser('worktree-check');check.add_argument('--path',required=True)
|
|
d=sub.add_parser('docs').add_subparsers(dest='doc_action',required=True)
|
|
snap=d.add_parser('snapshot');snap.add_argument('--file',required=True);snap.add_argument('--record',required=True)
|
|
apply=d.add_parser('apply');apply.add_argument('--task',required=True);apply.add_argument('--request',required=True)
|
|
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')
|
|
q=sub.add_parser('queue').add_subparsers(dest='queue_action',required=True)
|
|
qe=q.add_parser('enqueue');qe.add_argument('--task',required=True);qe.add_argument('--expected-dev',required=True)
|
|
qe.add_argument('--head');qe.add_argument('--priority',type=int,default=0);qe.add_argument('--request-id')
|
|
ql=q.add_parser('list');ql.add_argument('--all',action='store_true')
|
|
qc=q.add_parser('cancel');qc.add_argument('--task',required=True);qc.add_argument('--id',required=True);qc.add_argument('--reason',required=True)
|
|
qs=q.add_parser('schedule');qs.add_argument('--drain',action='store_true')
|
|
cu=sub.add_parser('cleanup').add_subparsers(dest='cleanup_action',required=True)
|
|
cs=cu.add_parser('status');cs.add_argument('--path');cs.add_argument('--branch')
|
|
rec=sub.add_parser('recover');rec.add_argument('--run',required=True);rec.add_argument('--reason',required=True)
|
|
op=sub.add_parser('operation');op.add_argument('run')
|
|
for action in ('freeze','unfreeze'):
|
|
fr=sub.add_parser(action);fr.add_argument('--task',required=True);fr.add_argument('--resource',required=True);fr.add_argument('--reason',required=True)
|
|
hooks=sub.add_parser('hooks').add_subparsers(dest='hook_action',required=True)
|
|
hooks.add_parser('install');hc=hooks.add_parser('check');hc.add_argument('--merge',action='store_true');hr=hooks.add_parser('ref');hr.add_argument('state')
|
|
v=sub.add_parser('validate');v.add_argument('--task',required=True);v.add_argument('command',nargs=argparse.REMAINDER)
|
|
i=sub.add_parser('integrate');i.add_argument('--task',required=True);i.add_argument('--expected-dev',required=True);i.add_argument('--request-id')
|
|
q=sub.add_parser('queue').add_subparsers(dest='queue_action',required=True)
|
|
qe=q.add_parser('enqueue');qe.add_argument('--task',required=True);qe.add_argument('--expected-dev',default='dev');qe.add_argument('--head-sha');qe.add_argument('--priority',type=int,default=0);qe.add_argument('--request-id')
|
|
ql=q.add_parser('list');ql.add_argument('--all',action='store_true')
|
|
qc=q.add_parser('cancel');qc.add_argument('--id',required=True);qc.add_argument('--reason',required=True)
|
|
q.add_parser('schedule')
|
|
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)
|
|
return p
|
|
|
|
|
|
def main():
|
|
args=parser().parse_args();repo=Repo(args.repo)
|
|
if args.action=='init':initialize(repo)
|
|
elif args.action in ('status','doctor'):
|
|
identity=repo.identity()
|
|
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')")]
|
|
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'])
|
|
emit({'status':'ok','lock_root':str(repo.root),'main':str(repo.main),'version':VERSION,'epoch':identity['epoch'],
|
|
'tasks':tasks,'runs':runs,'queue':queue})
|
|
elif args.action=='task':
|
|
if args.task_action=='register':
|
|
if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}',args.id):deny('INVALID_TASK','Invalid task id')
|
|
branch=git(repo.current,'branch','--show-current')
|
|
if repo.current==repo.main or repo.current.parent!=repo.main/'.worktrees' or not branch.startswith(('feature/','fix/')):
|
|
deny('WORKTREE_REQUIRED','Register from a direct .worktrees feature/fix checkout')
|
|
scopes=sorted(set(relative(v) for v in args.scope))
|
|
for scope in scopes:
|
|
if scope != '.' and secret_path(scope):deny('PATH_OUT_OF_SCOPE','Git and runtime internals cannot be registered as source scopes')
|
|
safe_path(repo.current,scope)
|
|
for cap in args.capability:
|
|
if cap not in ('docs:admin','docs:approve'):resource(cap)
|
|
values=(args.id,args.owner,str(repo.current),branch,json.dumps(scopes),json.dumps(sorted(set(args.capability))),'active',git(repo.current,'rev-parse','HEAD'))
|
|
with repo.os_file('worktree-' + digest(str(repo.current)) + '.lock').open('a+') as lifecycle, repo.db(True) as db:
|
|
fcntl.flock(lifecycle,fcntl.LOCK_SH|fcntl.LOCK_NB)
|
|
if not repo.current.exists():deny('WORKTREE_MISMATCH','Worktree disappeared during registration')
|
|
old=db.execute('SELECT * FROM tasks WHERE id=?',(args.id,)).fetchone()
|
|
if old:
|
|
if tuple(old)[:6]!=values[:6] or old['state']!='active':deny('TASK_EXISTS','Task registration differs; no implicit authority expansion')
|
|
else:
|
|
if db.execute("SELECT id FROM tasks WHERE worktree=? AND state='active'", (str(repo.current),)).fetchone():
|
|
deny('WORKTREE_OCCUPIED', 'Another active task already owns this checkout')
|
|
db.execute('INSERT INTO tasks VALUES(?,?,?,?,?,?,?,?)',values)
|
|
event(db,args.id,{'event':'registered','worktree':str(repo.current)})
|
|
emit({'status':'registered','task':args.id,'worktree':str(repo.current)})
|
|
else:
|
|
task_row(repo,args.id)
|
|
with repo.db(True) as db:
|
|
if db.execute("SELECT id FROM runs WHERE task=? AND state IN ('running','queued','recovery_required')",(args.id,)).fetchone():deny('LOCK_BUSY','Task still has an active or unresolved operation')
|
|
db.execute('UPDATE tasks SET state=? WHERE id=?',(args.status,args.id))
|
|
emit({'status':args.status,'task':args.id})
|
|
elif args.action=='with':return execute(repo,args)
|
|
elif args.action=='guard':guard(repo,args)
|
|
elif args.action=='worktree-check':worktree_check(repo,args.path)
|
|
elif args.action=='docs':
|
|
if args.doc_action=='snapshot':docs_snapshot(repo,args)
|
|
elif args.doc_action=='project':return docs_project(repo,args)
|
|
else:return docs_apply(repo,args)
|
|
elif args.action=='queue':
|
|
if args.queue_action=='enqueue':return queue_enqueue(repo,args)
|
|
if args.queue_action=='list':return queue_list(repo,args)
|
|
if args.queue_action=='cancel':return queue_cancel(repo,args)
|
|
return queue_schedule(repo,args)
|
|
elif args.action=='cleanup':
|
|
if not args.path and not args.branch:
|
|
deny('INVALID_REQUEST', 'cleanup status requires --path or --branch')
|
|
if args.path and args.branch:
|
|
deny('INVALID_REQUEST', 'Provide only one of --path or --branch')
|
|
return cleanup_status(repo,args)
|
|
elif args.action=='recover':recover(repo,args)
|
|
elif args.action=='operation':
|
|
with repo.db() as db:row=db.execute('SELECT * FROM runs WHERE id=?',(args.run,)).fetchone()
|
|
if not row:deny('UNKNOWN_OPERATION','Unknown run id')
|
|
emit(dict(row))
|
|
elif args.action in ('freeze','unfreeze'):
|
|
task=task_row(repo,args.task);r=resource(args.resource);check_resources(repo,task,[r])
|
|
with repo.db(True) as db:
|
|
if args.action=='freeze':
|
|
for row in db.execute("SELECT resources FROM runs WHERE state IN ('running','recovery_required')"):
|
|
if any(overlaps(r,v) for v in json.loads(row['resources'])):deny('LOCK_BUSY','Cannot freeze a resource with an active writer')
|
|
db.execute('INSERT INTO freezes VALUES(?,?,?)',(r,args.task,args.reason))
|
|
else:
|
|
row=db.execute('SELECT owner FROM freezes WHERE resource=?',(r,)).fetchone()
|
|
if row and row['owner']!=args.task:deny('PATH_OUT_OF_SCOPE','Only the freeze owner can unfreeze')
|
|
db.execute('DELETE FROM freezes WHERE resource=?',(r,))
|
|
event(db,args.task,{'event':args.action,'resource':r,'reason':args.reason})
|
|
emit({'status':args.action,'resource':r})
|
|
elif args.action=='hooks':
|
|
if args.hook_action=='install':install_hooks(repo)
|
|
elif args.hook_action=='check':hook_check(repo,args.merge)
|
|
else:hook_ref(repo,args.state)
|
|
elif args.action=='validate':return validate(repo,args)
|
|
elif args.action=='integrate':return integrate(repo,args)
|
|
elif args.action=='deploy':return deploy_cmd(repo,args)
|
|
elif args.action=='queue':
|
|
if args.queue_action=='enqueue':return queue_enqueue(repo,args)
|
|
if args.queue_action=='list':return queue_list(repo,args)
|
|
if args.queue_action=='cancel':return queue_cancel(repo,args)
|
|
if args.queue_action=='schedule':return queue_schedule(repo,args)
|
|
return 0
|
|
|
|
|
|
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)
|