400 lines
15 KiB
Python
400 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""W4 automation helpers for agentctl: durable jobs, doc projection, deploy loop.
|
|
|
|
Linux-first. Process liveness reuses the host agentctl proc/alive/boot helpers
|
|
passed in as callbacks so a future macOS process layer can plug in without
|
|
rewriting queue semantics.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional
|
|
|
|
SCHEMA_SQL = '''
|
|
CREATE TABLE IF NOT EXISTS jobs(
|
|
id TEXT PRIMARY KEY,
|
|
task TEXT NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
priority INTEGER NOT NULL DEFAULT 100,
|
|
not_before REAL NOT NULL,
|
|
created REAL NOT NULL,
|
|
updated REAL NOT NULL,
|
|
lease_run TEXT,
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
last_error TEXT,
|
|
result TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, not_before, priority, created);
|
|
CREATE TABLE IF NOT EXISTS deployments(
|
|
id TEXT PRIMARY KEY,
|
|
task TEXT NOT NULL,
|
|
env TEXT NOT NULL,
|
|
expected_commit TEXT NOT NULL,
|
|
manifest TEXT,
|
|
phase TEXT NOT NULL,
|
|
dry_run INTEGER NOT NULL DEFAULT 0,
|
|
created REAL NOT NULL,
|
|
updated REAL NOT NULL,
|
|
receipt TEXT,
|
|
last_error TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS deployments_task ON deployments(task, phase);
|
|
CREATE TABLE IF NOT EXISTS projections(
|
|
file TEXT NOT NULL,
|
|
record_id TEXT NOT NULL,
|
|
task TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
content_hash TEXT NOT NULL,
|
|
updated REAL NOT NULL,
|
|
PRIMARY KEY(file, record_id)
|
|
);
|
|
'''
|
|
|
|
JOB_KINDS = frozenset({
|
|
'validate', 'integrate', 'with', 'deploy_prepare', 'deploy_run',
|
|
'deploy_accept', 'docs_project', 'cleanup_inspect', 'noop',
|
|
})
|
|
JOB_ACTIVE = ('pending', 'leased', 'running')
|
|
JOB_TERMINAL = ('completed', 'failed', 'cancelled')
|
|
DEPLOY_PHASES = (
|
|
'prepared', 'deploying', 'accepting', 'passed', 'failed', 'recovery_required',
|
|
)
|
|
# Beyond CURRENT/MATRIX: any Markdown table under .doc/ may be projected when
|
|
# registered here. Layout changes still require docs:admin via docs apply.
|
|
PROJECTABLE_DOCS = frozenset({
|
|
'.doc/ops/CURRENT.md',
|
|
'.doc/status/FEATURE-MATRIX.md',
|
|
'.doc/ops/CURRENT.md', # duplicate ok in set
|
|
})
|
|
PROJECTABLE_PREFIXES = ('.doc/ops/', '.doc/status/', '.doc/remediation/')
|
|
|
|
|
|
def ensure_schema(db: sqlite3.Connection) -> None:
|
|
db.executescript(SCHEMA_SQL)
|
|
|
|
|
|
def _now() -> float:
|
|
return time.time()
|
|
|
|
|
|
def queue_enqueue(db, *, task: str, kind: str, payload: dict, priority: int = 100,
|
|
not_before: Optional[float] = None, job_id: Optional[str] = None) -> dict:
|
|
if kind not in JOB_KINDS:
|
|
raise ValueError(f'unsupported job kind: {kind}')
|
|
if not isinstance(payload, dict):
|
|
raise ValueError('payload must be an object')
|
|
jid = job_id or str(uuid.uuid4())
|
|
if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', jid):
|
|
raise ValueError('invalid job id')
|
|
ts = _now()
|
|
when = float(not_before) if not_before is not None else ts
|
|
existing = db.execute('SELECT * FROM jobs WHERE id=?', (jid,)).fetchone()
|
|
if existing:
|
|
if (existing['task'] != task or existing['kind'] != kind
|
|
or existing['payload'] != json.dumps(payload, sort_keys=True)):
|
|
raise ValueError('job id reused with different payload')
|
|
return dict(existing)
|
|
db.execute(
|
|
'''INSERT INTO jobs(id,task,kind,payload,state,priority,not_before,created,updated)
|
|
VALUES(?,?,?,?,?,?,?,?,?)''',
|
|
(jid, task, kind, json.dumps(payload, sort_keys=True), 'pending',
|
|
int(priority), when, ts, ts))
|
|
return dict(db.execute('SELECT * FROM jobs WHERE id=?', (jid,)).fetchone())
|
|
|
|
|
|
def queue_list(db, *, task: Optional[str] = None, state: Optional[str] = None,
|
|
limit: int = 50) -> list:
|
|
clauses, args = [], []
|
|
if task:
|
|
clauses.append('task=?')
|
|
args.append(task)
|
|
if state:
|
|
clauses.append('state=?')
|
|
args.append(state)
|
|
where = (' WHERE ' + ' AND '.join(clauses)) if clauses else ''
|
|
rows = db.execute(
|
|
f'SELECT * FROM jobs{where} ORDER BY priority ASC, created ASC LIMIT ?',
|
|
(*args, max(1, min(int(limit), 500)))).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def queue_cancel(db, job_id: str, reason: str) -> dict:
|
|
row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
|
|
if not row:
|
|
raise KeyError(job_id)
|
|
if row['state'] in JOB_TERMINAL:
|
|
return dict(row)
|
|
if row['state'] == 'running':
|
|
raise RuntimeError('cannot cancel a running job; recover the lease first')
|
|
db.execute(
|
|
"UPDATE jobs SET state='cancelled',last_error=?,updated=? WHERE id=?",
|
|
(reason, _now(), job_id))
|
|
return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())
|
|
|
|
|
|
def claim_next_job(db, *, worker: str) -> Optional[dict]:
|
|
"""Atomically lease the next due pending job. Caller must be in a write txn."""
|
|
ts = _now()
|
|
row = db.execute(
|
|
'''SELECT * FROM jobs WHERE state='pending' AND not_before<=?
|
|
ORDER BY priority ASC, created ASC LIMIT 1''', (ts,)).fetchone()
|
|
if not row:
|
|
return None
|
|
db.execute(
|
|
'''UPDATE jobs SET state='leased',lease_run=?,attempts=attempts+1,updated=?
|
|
WHERE id=? AND state='pending' ''',
|
|
(worker, ts, row['id']))
|
|
claimed = db.execute('SELECT * FROM jobs WHERE id=?', (row['id'],)).fetchone()
|
|
if not claimed or claimed['state'] != 'leased':
|
|
return None
|
|
return dict(claimed)
|
|
|
|
|
|
def finish_job(db, job_id: str, *, ok: bool, result: Optional[dict] = None,
|
|
error: Optional[str] = None) -> dict:
|
|
state = 'completed' if ok else 'failed'
|
|
db.execute(
|
|
'''UPDATE jobs SET state=?,result=?,last_error=?,updated=? WHERE id=?''',
|
|
(state, json.dumps(result or {}, sort_keys=True), error, _now(), job_id))
|
|
return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())
|
|
|
|
|
|
def mark_job_running(db, job_id: str) -> None:
|
|
db.execute("UPDATE jobs SET state='running',updated=? WHERE id=?", (_now(), job_id))
|
|
|
|
|
|
def projectable_file(path: str) -> bool:
|
|
if path in PROJECTABLE_DOCS or path in {'.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'}:
|
|
return True
|
|
return any(path.startswith(prefix) for prefix in PROJECTABLE_PREFIXES) and path.endswith('.md')
|
|
|
|
|
|
def build_projection_row(*, task_id: str, owner: str, branch: str, worktree: str,
|
|
mutex: str, status: str, extra: Optional[dict] = None) -> str:
|
|
"""Canonical CURRENT-style row used for machine→doc projection."""
|
|
cells = [owner, task_id, branch, worktree, mutex, status]
|
|
if extra:
|
|
for key in sorted(extra):
|
|
cells.append(str(extra[key]))
|
|
return '| ' + ' | '.join(cells) + ' |'
|
|
|
|
|
|
def record_projection(db, *, file: str, record_id: str, task: str, source: str,
|
|
content_hash: str) -> None:
|
|
db.execute(
|
|
'''INSERT INTO projections(file,record_id,task,source,content_hash,updated)
|
|
VALUES(?,?,?,?,?,?)
|
|
ON CONFLICT(file,record_id) DO UPDATE SET
|
|
task=excluded.task, source=excluded.source,
|
|
content_hash=excluded.content_hash, updated=excluded.updated''',
|
|
(file, record_id, task, source, content_hash, _now()))
|
|
|
|
|
|
def deployment_create(db, *, deploy_id: str, task: str, env: str, expected_commit: str,
|
|
manifest: Optional[str], dry_run: bool) -> dict:
|
|
if env not in ('dev', 'test', 'main'):
|
|
raise ValueError('env must be dev|test|main')
|
|
if not re.fullmatch(r'[0-9a-f]{7,40}', expected_commit):
|
|
raise ValueError('expected_commit must be a hex SHA')
|
|
ts = _now()
|
|
existing = db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone()
|
|
if existing:
|
|
if (existing['task'] != task or existing['env'] != env
|
|
or existing['expected_commit'] != expected_commit):
|
|
raise ValueError('deploy id reused with different identity')
|
|
return dict(existing)
|
|
db.execute(
|
|
'''INSERT INTO deployments(id,task,env,expected_commit,manifest,phase,dry_run,created,updated)
|
|
VALUES(?,?,?,?,?,?,?,?,?)''',
|
|
(deploy_id, task, env, expected_commit, manifest, 'prepared',
|
|
1 if dry_run else 0, ts, ts))
|
|
return dict(db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone())
|
|
|
|
|
|
def deployment_set_phase(db, deploy_id: str, phase: str, *, receipt: Optional[dict] = None,
|
|
error: Optional[str] = None) -> dict:
|
|
if phase not in DEPLOY_PHASES:
|
|
raise ValueError(f'invalid deploy phase: {phase}')
|
|
db.execute(
|
|
'''UPDATE deployments SET phase=?,receipt=?,last_error=?,updated=? WHERE id=?''',
|
|
(phase, json.dumps(receipt or {}, sort_keys=True) if receipt is not None else None,
|
|
error, _now(), deploy_id))
|
|
return dict(db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone())
|
|
|
|
|
|
def deployment_get(db, deploy_id: str) -> Optional[dict]:
|
|
row = db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def dry_run_deploy_steps(deploy: dict) -> dict:
|
|
"""Closed-loop steps without touching shared docker. Used by tests and design."""
|
|
return {
|
|
'env': deploy['env'],
|
|
'expected_commit': deploy['expected_commit'],
|
|
'manifest': deploy.get('manifest'),
|
|
'steps': [
|
|
{'name': 'guard_env_lease', 'status': 'ok'},
|
|
{'name': 'verify_main_clean', 'status': 'ok'},
|
|
{'name': 'apply_or_rebuild', 'status': 'skipped_dry_run'},
|
|
{'name': 'health_ready', 'status': 'skipped_dry_run'},
|
|
{'name': 'pi_image_identity', 'status': 'skipped_dry_run'},
|
|
],
|
|
'dry_run': True,
|
|
}
|
|
|
|
|
|
def cleanup_visibility_report(
|
|
*,
|
|
path: Path,
|
|
branch: str,
|
|
head: str,
|
|
ancestor_of_dev: bool,
|
|
dirty: bool,
|
|
unfinished_git: list,
|
|
registry: dict,
|
|
process: dict,
|
|
) -> dict:
|
|
"""Structured cleanup gate report. Never deletes; visibility only."""
|
|
blockers = []
|
|
if not ancestor_of_dev:
|
|
blockers.append({'code': 'NOT_MERGED', 'message': 'HEAD is not an ancestor of dev'})
|
|
if dirty:
|
|
blockers.append({'code': 'DIRTY_TREE', 'message': 'tracked/untracked/ignored files remain'})
|
|
for marker in unfinished_git:
|
|
blockers.append({'code': 'GIT_IN_PROGRESS', 'message': f'unfinished Git operation: {marker}'})
|
|
if registry.get('active_task'):
|
|
blockers.append({'code': 'TASK_ACTIVE', 'message': 'registered task still active'})
|
|
if registry.get('live_run'):
|
|
blockers.append({'code': 'LOCK_BUSY', 'message': 'live or unresolved operation',
|
|
'run_id': registry.get('live_run')})
|
|
status = process.get('status')
|
|
if status == 'busy':
|
|
blockers.append({'code': 'PROCESS_BUSY', 'message': 'process still uses worktree cwd',
|
|
'pids': process.get('pids', [])})
|
|
elif status == 'unreadable':
|
|
blockers.append({'code': 'PROCESS_UNREADABLE', 'message': 'cannot verify same-user cwd',
|
|
'pids': process.get('unreadable', []),
|
|
'platform_note': process.get('platform_note')})
|
|
elif status == 'unsupported':
|
|
blockers.append({'code': 'PROCESS_UNSUPPORTED',
|
|
'message': process.get('message', 'process inspection unsupported'),
|
|
'platform_note': process.get('platform_note')})
|
|
return {
|
|
'status': 'ready' if not blockers else 'cleanup_pending',
|
|
'path': str(path),
|
|
'branch': branch,
|
|
'head': head,
|
|
'gates': {
|
|
'ancestor_of_dev': ancestor_of_dev,
|
|
'dirty': dirty,
|
|
'unfinished_git': unfinished_git,
|
|
'registry': registry,
|
|
'process': process,
|
|
},
|
|
'blockers': blockers,
|
|
}
|
|
|
|
|
|
def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
|
|
"""Fail-closed cwd scan using /proc. macOS should use lsof via sibling worktree."""
|
|
import sys
|
|
if not sys.platform.startswith('linux'):
|
|
return {
|
|
'status': 'unsupported',
|
|
'message': f'/proc cwd scan unavailable on {sys.platform}',
|
|
'platform_note': 'Use agentctl-macos lsof enumerator; treat unknown as cleanup_pending',
|
|
'pids': [],
|
|
'unreadable': [],
|
|
}
|
|
uid = os.getuid() if uid is None else uid
|
|
busy, unreadable = [], []
|
|
try:
|
|
entries = list(Path('/proc').iterdir())
|
|
except OSError:
|
|
return {
|
|
'status': 'unreadable',
|
|
'message': 'cannot list /proc',
|
|
'platform_note': 'fail-closed',
|
|
'pids': [],
|
|
'unreadable': ['*'],
|
|
}
|
|
for proc in entries:
|
|
if not proc.name.isdigit():
|
|
continue
|
|
try:
|
|
if proc.stat().st_uid != uid:
|
|
continue
|
|
cwd = Path(os.readlink(proc / 'cwd'))
|
|
if cwd == path or path in cwd.parents:
|
|
busy.append(proc.name)
|
|
except (FileNotFoundError, ProcessLookupError):
|
|
continue
|
|
except PermissionError:
|
|
try:
|
|
if (proc / 'comm').read_text().strip() == 'systemd':
|
|
continue
|
|
except (OSError, UnicodeError):
|
|
pass
|
|
unreadable.append(proc.name)
|
|
if busy:
|
|
return {'status': 'busy', 'pids': busy, 'unreadable': unreadable}
|
|
if unreadable:
|
|
return {'status': 'unreadable', 'pids': [], 'unreadable': unreadable,
|
|
'platform_note': 'fail-closed; do not treat as exited'}
|
|
return {'status': 'clear', 'pids': [], 'unreadable': []}
|
|
|
|
|
|
def pi_source_bind_paths() -> tuple:
|
|
"""Host paths that must not be bind-mounted when PI image delivery is enforced."""
|
|
return (
|
|
'./deploy/pi-agent/runner.mjs',
|
|
'./deploy/pi-agent/pi-args.mjs',
|
|
'./deploy/pi-agent/result-v1.mjs',
|
|
'./deploy/pi-agent/extensions',
|
|
'deploy/pi-agent/runner.mjs',
|
|
'deploy/pi-agent/pi-args.mjs',
|
|
'deploy/pi-agent/result-v1.mjs',
|
|
'deploy/pi-agent/extensions',
|
|
)
|
|
|
|
|
|
def inspect_compose_for_pi_binds(compose_text: str) -> dict:
|
|
"""Static check: flag source bind mounts under pi-agent service."""
|
|
forbidden = pi_source_bind_paths()
|
|
hits = []
|
|
in_pi = False
|
|
for line in compose_text.splitlines():
|
|
stripped = line.strip()
|
|
if re.match(r'^pi-agent\s*:', stripped):
|
|
in_pi = True
|
|
continue
|
|
if in_pi and re.match(r'^[A-Za-z0-9_.-]+:\s*$', stripped):
|
|
in_pi = False
|
|
if in_pi:
|
|
for item in forbidden:
|
|
if item in stripped and (':/app/' in stripped or stripped.startswith('-')):
|
|
hits.append(stripped)
|
|
break
|
|
return {
|
|
'ok': not hits,
|
|
'source_binds': hits,
|
|
'recommendation': (
|
|
'Apply docker-compose.dev.pi-image.yml overlay so runner source ships in the image'
|
|
if hits else 'no pi-agent source binds detected'
|
|
),
|
|
}
|
|
|
|
|
|
# Type aliases for agentctl callbacks (avoid circular imports at runtime).
|
|
EmitFn = Callable[[Any], None]
|
|
DenyFn = Callable[..., None]
|