merge: feature/governance-automation-2026-09-22
Keep macOS process platform; land W4 queue/automation and cleanup-report.
This commit is contained in:
WeiCHEN
@@ -0,0 +1,69 @@
|
||||
---
|
||||
title: "W4 治理自动化剩余：持久队列与闭环"
|
||||
task_id: governance-automation-2026-09-22
|
||||
status: in_progress
|
||||
created: 2026-09-22
|
||||
updated: 2026-09-22
|
||||
owner: Cursor
|
||||
task_type: implementation
|
||||
task_tier: multi_stage
|
||||
policy_revision: "agent-governance-2026-09-21-r1 @ 364ed7f1a3f0393ef77ec4623a9518116ed2cb87"
|
||||
branch: feature/governance-automation-2026-09-22
|
||||
worktree: .worktrees/governance-automation-2026-09-22
|
||||
base_dev_sha: 364ed7f1a3f0393ef77ec4623a9518116ed2cb87
|
||||
write_scope: [".doc", "scripts", "docker-compose.dev.yml", "docker-compose.dev.pi-image.yml"]
|
||||
allowed_environments: []
|
||||
head_sha: "ae001a85c15a2b9213125f9d1d8c3e006221d532"
|
||||
mutex: agent-workflow-governance-automation
|
||||
review_state: pending
|
||||
integration_state: pending
|
||||
deployment: not_required
|
||||
cleanup: pending
|
||||
cleanup_owner: "Cursor 之外的集成/维护执行者"
|
||||
cleanup_trigger: "合入 dev、验收通过且 owner 退出后核对清理门禁"
|
||||
remote_sync: not_requested
|
||||
---
|
||||
|
||||
# W4 治理自动化剩余：持久队列与闭环
|
||||
|
||||
## 用户目标与授权
|
||||
|
||||
- 用户目标：推进 agent-workflow-governance 的 W1–W4 遗留——后台持久队列与自动调度、通用文档投影、业务任务部署/恢复闭环、产品 PI 源码镜像交付、清理可见性。
|
||||
- 交付终点：可测的 durable queue 设计与部分实现、相关 CLI、单测；短文记录 landed vs deferred；增量提交；不 push。
|
||||
- 禁止：push；强删 worktree；操作未授权环境；宣称完整 W3/W4 已关闭。
|
||||
|
||||
## 步骤与验收
|
||||
|
||||
1. [x] Durable queue schema + enqueue/list/cancel/tick（`scripts/agentctl_w4.py` / automation 增量）
|
||||
2. [~] Generic doc projection（registry 骨架；完整投影仍待）
|
||||
3. [~] Deploy/recover 状态机 + 收据（表与 helper 落地；真实 env 闭环未跑）
|
||||
4. [ ] PI 源码镜像交付 compose profile + 检查入口（deferred）
|
||||
5. [~] Cleanup visibility report（fail-closed 文案；Darwin 需 agentctl-macos 进程层）
|
||||
6. [x] short-term 写明 landed vs deferred；测试夹具复制 `agentctl_w4.py`
|
||||
|
||||
## 验证与证据
|
||||
|
||||
| 范围 | 命令 | 实际 |
|
||||
|---|---|---|
|
||||
| ops_guards 抽样 | `python3 -B -m unittest scripts.tests.test_ops_guards -v` | 14 跑；12 PASS；2 FAIL（`Lease`/`boot` 在 Darwin 仍依赖 `/proc`，本分支未含已合入的 agentctl-macos 进程层） |
|
||||
| 夹具回归 | 临时仓缺 `agentctl_w4.py` → `ModuleNotFoundError` | 已修：tests 复制 sibling 模块 |
|
||||
|
||||
## Landed vs deferred
|
||||
|
||||
**Landed（本分支 tip）**
|
||||
|
||||
- `scripts/agentctl_w4.py`：jobs / deployments / projections schema + enqueue/list/cancel/tick helpers
|
||||
- `scripts/agentctl_automation.py`：并行自动化增量（待与 w4 收敛）
|
||||
- `scripts/agentctl.py`：接入 w4 schema/`ensure_schema`
|
||||
- 测试夹具复制 `agentctl_w4.py`，避免黑盒仓 import 崩
|
||||
|
||||
**Deferred**
|
||||
|
||||
- Rebase/merge 本地 `dev` 已合入的 `agentctl-macos` 进程层后，Darwin 上 `agentctl with`/`boot()` 才能非崩溃跑全套脚本测
|
||||
- 完整 docs projection apply、真实 deploy/recover 收据、PI 镜像 compose profile
|
||||
- 清理 `agentctl_automation.py` 与 `agentctl_w4.py` 重复实现
|
||||
|
||||
## 收口
|
||||
|
||||
- 状态保持 `in_progress`（禁止宣称 W3/W4 全关）。
|
||||
- `integration_state=pending`；不 push。
|
||||
@@ -199,6 +199,33 @@ def assert_no_process_cwd(path):
|
||||
raise Refused(f"process {pid} still uses this worktree as cwd")
|
||||
|
||||
|
||||
def inspect_process_cwd(path):
|
||||
"""Structured process-cwd visibility for cleanup reports (never force-clears)."""
|
||||
path = Path(path).resolve()
|
||||
try:
|
||||
for pid, cwd in iter_same_user_cwds():
|
||||
try:
|
||||
resolved = cwd.resolve()
|
||||
except OSError:
|
||||
return {'state': 'unreadable', 'unreadable': [str(pid)]}
|
||||
if resolved == path or path in resolved.parents:
|
||||
return {'state': 'occupied', 'pid': str(pid)}
|
||||
return {'state': 'clear'}
|
||||
except Refused as exc:
|
||||
msg = str(exc)
|
||||
if 'unsupported' in msg:
|
||||
return {
|
||||
'state': 'unsupported_platform',
|
||||
'platform': sys.platform,
|
||||
'message': msg,
|
||||
}
|
||||
if 'cannot verify cwd of same-user process' in msg:
|
||||
# Extract pid list when present after the word process
|
||||
return {'state': 'unreadable', 'unreadable': [msg], 'message': msg}
|
||||
return {'state': 'unreadable', 'unreadable': [], 'message': msg}
|
||||
|
||||
|
||||
|
||||
@contextmanager
|
||||
def removal_lock(main, path):
|
||||
# Every agentctl run/registration holds the shared side of this same lock.
|
||||
@@ -259,10 +286,134 @@ def remove_checked(repo, main, branch, delete_branch, path):
|
||||
print(json.dumps({"removed": str(path), "branch": branch, "branch_deleted": delete_branch}))
|
||||
|
||||
|
||||
def inspect_worktree(repo, main, entry):
|
||||
"""Read-only cleanup readiness for one porcelain worktree entry."""
|
||||
path = Path(entry.get("worktree", ""))
|
||||
branch_ref = entry.get("branch", "")
|
||||
report = {
|
||||
"path": str(path),
|
||||
"branch": branch_ref[len("refs/heads/"):] if branch_ref.startswith("refs/heads/") else branch_ref or None,
|
||||
"verdict": "blocked",
|
||||
"blockers": [],
|
||||
"notes": [],
|
||||
}
|
||||
if "bare" in entry:
|
||||
report["blockers"].append("bare_worktree")
|
||||
return report
|
||||
try:
|
||||
safe_directory(main, path.resolve())
|
||||
except Refused as exc:
|
||||
report["blockers"].append(f"path:{exc}")
|
||||
return report
|
||||
if "locked" in entry:
|
||||
report["blockers"].append("git_locked")
|
||||
if "prunable" in entry:
|
||||
report["blockers"].append("git_prunable")
|
||||
if not path.is_dir():
|
||||
report["blockers"].append("missing_directory")
|
||||
return report
|
||||
if not report["branch"]:
|
||||
report["blockers"].append("detached_or_unmapped_branch")
|
||||
return report
|
||||
try:
|
||||
actual = git(path, "symbolic-ref", "--quiet", "HEAD").stdout.strip()
|
||||
if actual != f"refs/heads/{report['branch']}":
|
||||
report["blockers"].append("branch_identity_mismatch")
|
||||
head = git(path, "rev-parse", "HEAD").stdout.strip()
|
||||
report["head"] = head
|
||||
if git(main, "merge-base", "--is-ancestor", head, "refs/heads/dev", check=False).returncode:
|
||||
report["blockers"].append("not_ancestor_of_dev")
|
||||
else:
|
||||
report["notes"].append("ancestor_of_dev")
|
||||
for marker in ("MERGE_HEAD", "CHERRY_PICK_HEAD", "REVERT_HEAD", "rebase-merge", "rebase-apply", "BISECT_START"):
|
||||
marker_path = Path(git(path, "rev-parse", "--path-format=absolute", "--git-path", marker).stdout.strip())
|
||||
if marker_path.exists():
|
||||
report["blockers"].append(f"unfinished_git:{marker}")
|
||||
dirty = git(path, "status", "--porcelain=v1", "--untracked-files=all", "--ignored=matching").stdout
|
||||
if dirty:
|
||||
report["blockers"].append("dirty_or_ignored_files")
|
||||
report["dirty_sample"] = dirty.splitlines()[:8]
|
||||
except Refused as exc:
|
||||
report["blockers"].append(f"git:{exc}")
|
||||
check = control(repo, "worktree-check", "--path", str(path), check=False)
|
||||
if check.returncode:
|
||||
try:
|
||||
payload = json.loads(check.stderr.strip().splitlines()[-1])
|
||||
report["blockers"].append(f"registry:{payload.get('code', 'DENIED')}")
|
||||
if payload.get("message"):
|
||||
report["notes"].append(payload["message"])
|
||||
except (ValueError, IndexError, json.JSONDecodeError):
|
||||
report["blockers"].append("registry:worktree_check_failed")
|
||||
else:
|
||||
report["notes"].append("registry_clear")
|
||||
process = inspect_process_cwd(path)
|
||||
report["process"] = process
|
||||
if process["state"] == "occupied":
|
||||
report["blockers"].append(f"process_cwd:{process['pid']}")
|
||||
elif process["state"] == "unreadable":
|
||||
report["blockers"].append("process_cwd_unreadable")
|
||||
elif process["state"] == "unsupported_platform":
|
||||
report["blockers"].append("process_cwd_platform_pending")
|
||||
report["notes"].append(process.get("message") or "platform process check pending")
|
||||
key = hashlib.sha256(str(path.resolve()).encode()).hexdigest()
|
||||
lock_path = main / ".locks" / "os" / f"worktree-{key}.lock"
|
||||
if lock_path.exists():
|
||||
fd = None
|
||||
try:
|
||||
fd = os.open(lock_path, os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
||||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||||
report["notes"].append("lifecycle_lock_idle")
|
||||
except BlockingIOError:
|
||||
report["blockers"].append("lifecycle_lock_busy")
|
||||
except OSError as exc:
|
||||
report["blockers"].append(f"lifecycle_lock:{exc}")
|
||||
finally:
|
||||
if fd is not None:
|
||||
os.close(fd)
|
||||
if not report["blockers"]:
|
||||
report["verdict"] = "ready"
|
||||
return report
|
||||
|
||||
|
||||
def cleanup_report(repo, main):
|
||||
"""Surface cleanup readiness without deleting anything."""
|
||||
try:
|
||||
control(repo, "status")
|
||||
registry = "initialized"
|
||||
except Exception:
|
||||
registry = "unavailable"
|
||||
items = []
|
||||
for entry in worktrees(main):
|
||||
path = entry.get("worktree")
|
||||
if not path:
|
||||
continue
|
||||
resolved = Path(path).resolve()
|
||||
if resolved == main.resolve():
|
||||
continue
|
||||
if resolved.parent != (main / ".worktrees").resolve():
|
||||
items.append({
|
||||
"path": str(resolved),
|
||||
"branch": entry.get("branch"),
|
||||
"verdict": "blocked",
|
||||
"blockers": ["outside_dot_worktrees"],
|
||||
"notes": ["not managed by agent-worktree cleanup"],
|
||||
})
|
||||
continue
|
||||
items.append(inspect_worktree(repo, main, entry))
|
||||
ready = sum(1 for item in items if item.get("verdict") == "ready")
|
||||
print(json.dumps({
|
||||
"status": "ok",
|
||||
"registry": registry,
|
||||
"worktrees": items,
|
||||
"summary": {"total": len(items), "ready": ready, "blocked": len(items) - ready},
|
||||
}, ensure_ascii=False, sort_keys=True))
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
commands = parser.add_subparsers(dest="command", required=True)
|
||||
commands.add_parser("list", help="read-only Git worktree listing")
|
||||
commands.add_parser("cleanup-report", help="read-only cleanup readiness for .worktrees entries")
|
||||
for name in ("create", "remove"):
|
||||
child = commands.add_parser(name)
|
||||
child.add_argument("branch")
|
||||
@@ -275,6 +426,9 @@ def main():
|
||||
if args.command == "list":
|
||||
print(git(root, "worktree", "list").stdout, end="")
|
||||
return 0
|
||||
if args.command == "cleanup-report":
|
||||
cleanup_report(repo, root)
|
||||
return 0
|
||||
safe_branch(args.branch)
|
||||
with write_lock(repo, root):
|
||||
if args.command == "create":
|
||||
|
||||
+418
-3
@@ -19,11 +19,32 @@ 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']
|
||||
@@ -396,6 +417,7 @@ class Repo:
|
||||
if write:
|
||||
db.execute('PRAGMA synchronous=FULL')
|
||||
db.execute('BEGIN IMMEDIATE')
|
||||
ensure_schema(db)
|
||||
yield db
|
||||
if write:
|
||||
db.commit()
|
||||
@@ -444,8 +466,9 @@ def initialize(repo):
|
||||
fcntl.flock(initialization, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
if (repo.root / 'identity.json').exists():
|
||||
identity = repo.identity()
|
||||
with repo.db() as db:
|
||||
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'
|
||||
@@ -497,7 +520,7 @@ def initialize(repo):
|
||||
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()
|
||||
@@ -508,6 +531,13 @@ def initialize(repo):
|
||||
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()
|
||||
@@ -1139,6 +1169,348 @@ def integrate(repo, args):
|
||||
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 machine registry / queue into a docs-apply request for CURRENT (generic snapshot JSON)."""
|
||||
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 filename not in SHARED_DOCS:
|
||||
deny('PATH_OUT_OF_SCOPE', 'Only shared governance documents are projected')
|
||||
if not filename.endswith('CURRENT.md'):
|
||||
deny('INVALID_REQUEST', 'Generic projection currently emits CURRENT claim rows only')
|
||||
content = safe_path(repo.current, filename).read_text()
|
||||
existing = doc_rows(content, args.task)
|
||||
expected = digest(existing[0][1]) if existing else None
|
||||
with repo.db(True) as 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)
|
||||
replacement = (
|
||||
f"| {task['owner']} | {task['id']} | {task['branch']} | {worktree_cell} "
|
||||
f"| {args.mutex or '-'} | {status} |"
|
||||
)
|
||||
change = {
|
||||
'file': filename,
|
||||
'record_id': args.task,
|
||||
'expected_sha256': expected,
|
||||
'replacement': replacement,
|
||||
}
|
||||
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'),
|
||||
}
|
||||
emit({
|
||||
'status': 'projected',
|
||||
'request': {'request_id': args.request_id or (args.task + '-project-1'), 'changes': [change]},
|
||||
'machine': machine,
|
||||
})
|
||||
return 0
|
||||
|
||||
|
||||
def cleanup_process_gate(path):
|
||||
"""Report process-cwd visibility without deleting. Fail-closed unknowns are explicit."""
|
||||
path = Path(path).resolve()
|
||||
if not sys.platform.startswith('linux'):
|
||||
return {
|
||||
'status': 'unknown',
|
||||
'platform': sys.platform,
|
||||
'reason': 'process cwd enumeration requires platform support; cleanup_pending until verified',
|
||||
'holders': [],
|
||||
'unreadable': [],
|
||||
}
|
||||
holders, unreadable = [], []
|
||||
try:
|
||||
entries = list(Path('/proc').iterdir())
|
||||
except OSError as exc:
|
||||
return {
|
||||
'status': 'unknown', 'platform': sys.platform,
|
||||
'reason': f'cannot list /proc: {exc}', 'holders': [], 'unreadable': [],
|
||||
}
|
||||
for proc in entries:
|
||||
if not proc.name.isdigit():
|
||||
continue
|
||||
try:
|
||||
if proc.stat().st_uid != os.getuid():
|
||||
continue
|
||||
cwd = Path(os.readlink(proc / 'cwd'))
|
||||
if cwd == path or path in cwd.parents:
|
||||
holders.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 holders:
|
||||
return {'status': 'blocked', 'platform': sys.platform, 'holders': holders, 'unreadable': unreadable,
|
||||
'reason': 'live process cwd uses worktree'}
|
||||
if unreadable:
|
||||
return {'status': 'unknown', 'platform': sys.platform, 'holders': [], 'unreadable': unreadable,
|
||||
'reason': 'cannot verify cwd of same-user processes; cleanup_pending'}
|
||||
return {'status': 'clear', 'platform': sys.platform, 'holders': [], 'unreadable': [], 'reason': None}
|
||||
|
||||
|
||||
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)
|
||||
@@ -1155,6 +1527,16 @@ def parser():
|
||||
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')
|
||||
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'):
|
||||
@@ -1163,6 +1545,11 @@ def parser():
|
||||
hooks.add_parser('install');hc=hooks.add_parser('check');hc.add_argument('--merge',action='store_true');hr=hooks.add_parser('ref');hr.add_argument('state')
|
||||
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')
|
||||
return p
|
||||
|
||||
|
||||
@@ -1171,11 +1558,22 @@ def main():
|
||||
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})
|
||||
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')
|
||||
@@ -1212,7 +1610,19 @@ def main():
|
||||
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()
|
||||
@@ -1237,6 +1647,11 @@ def main():
|
||||
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=='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
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,413 @@
|
||||
#!/usr/bin/env python3
|
||||
"""W4 governance automation: durable queue, doc projection, deploy ops.
|
||||
|
||||
Process liveness remains in agentctl.py (Linux /proc today; macOS sibling).
|
||||
This module is intentionally importable for unit tests without supervising children.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
# Job / operation terminal and active states.
|
||||
JOB_PENDING = 'pending'
|
||||
JOB_CLAIMED = 'claimed'
|
||||
JOB_RUNNING = 'running'
|
||||
JOB_COMPLETED = 'completed'
|
||||
JOB_FAILED = 'failed'
|
||||
JOB_CANCELLED = 'cancelled'
|
||||
JOB_ACTIVE = (JOB_PENDING, JOB_CLAIMED, JOB_RUNNING)
|
||||
JOB_TERMINAL = (JOB_COMPLETED, JOB_FAILED, JOB_CANCELLED)
|
||||
|
||||
DEPLOY_PHASES = (
|
||||
'planned', 'deploying', 'accepting', 'passed', 'failed',
|
||||
'recovery_required', 'recovered',
|
||||
)
|
||||
|
||||
# Managed shared docs plus projection-capable documents.
|
||||
DEFAULT_DOC_REGISTRY = {
|
||||
'.doc/ops/CURRENT.md': {
|
||||
'kind': 'shared_cas',
|
||||
'header': '| owner | task |',
|
||||
'rules': {'own_row_only': True, 'slot_protected': True},
|
||||
},
|
||||
'.doc/status/FEATURE-MATRIX.md': {
|
||||
'kind': 'shared_cas',
|
||||
'header': '| 模块 |',
|
||||
'rules': {'evidence_for_done': True},
|
||||
},
|
||||
'.doc/ops/projections/cleanup-visibility.md': {
|
||||
'kind': 'projection',
|
||||
'header': '| path | branch | reason | cleanup_owner |',
|
||||
'rules': {'generated': True},
|
||||
},
|
||||
}
|
||||
|
||||
SCHEMA_SQL = '''
|
||||
CREATE TABLE IF NOT EXISTS jobs(
|
||||
generation INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT UNIQUE NOT NULL,
|
||||
task TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
resources TEXT NOT NULL,
|
||||
state TEXT NOT NULL,
|
||||
priority INTEGER NOT NULL DEFAULT 100,
|
||||
created REAL NOT NULL,
|
||||
claimed_at REAL,
|
||||
claimed_by TEXT,
|
||||
run_id TEXT,
|
||||
finished_at REAL,
|
||||
result TEXT,
|
||||
fingerprint TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, priority, generation);
|
||||
CREATE TABLE IF NOT EXISTS operations(
|
||||
id TEXT PRIMARY KEY,
|
||||
task TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
phase TEXT NOT NULL,
|
||||
env TEXT NOT NULL,
|
||||
expected_sha TEXT,
|
||||
runtime_sha TEXT,
|
||||
receipt TEXT NOT NULL,
|
||||
created REAL NOT NULL,
|
||||
updated REAL NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS operations_task ON operations(task, updated);
|
||||
CREATE TABLE IF NOT EXISTS doc_registry(
|
||||
path TEXT PRIMARY KEY,
|
||||
kind TEXT NOT NULL,
|
||||
header TEXT,
|
||||
rules TEXT NOT NULL DEFAULT '{}'
|
||||
);
|
||||
'''
|
||||
|
||||
|
||||
def ensure_schema(db):
|
||||
"""Idempotent schema evolution; keeps protocol VERSION stable."""
|
||||
db.executescript(SCHEMA_SQL)
|
||||
for path, meta in DEFAULT_DOC_REGISTRY.items():
|
||||
db.execute(
|
||||
'INSERT OR IGNORE INTO doc_registry(path,kind,header,rules) VALUES(?,?,?,?)',
|
||||
(path, meta['kind'], meta.get('header'), json.dumps(meta.get('rules') or {})),
|
||||
)
|
||||
|
||||
|
||||
def process_platform_ready():
|
||||
"""True when the host can supply process start tokens for leases.
|
||||
|
||||
Linux uses /proc. Darwin needs the agentctl-macos process layer; until that
|
||||
lands, lease-backed dispatch stays deferred rather than TypeError.
|
||||
"""
|
||||
if os.environ.get('AGENTCTL_FORCE_PROC') == '1':
|
||||
return True
|
||||
if os.environ.get('AGENTCTL_FORCE_PROC') == '0':
|
||||
return False
|
||||
return Path('/proc/self/stat').is_file()
|
||||
|
||||
|
||||
def job_fingerprint(kind, payload, resources):
|
||||
blob = json.dumps({'kind': kind, 'payload': payload, 'resources': sorted(resources)},
|
||||
sort_keys=True, ensure_ascii=False)
|
||||
return blob
|
||||
|
||||
|
||||
def enqueue_job(db, task_id, kind, payload, resources, job_id=None, priority=100):
|
||||
if kind not in ('command', 'validate', 'integrate', 'deploy', 'docs_apply', 'cleanup_report'):
|
||||
raise ValueError(f'unsupported job kind: {kind}')
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError('payload must be an object')
|
||||
if not isinstance(resources, (list, tuple)) or not resources:
|
||||
raise ValueError('resources must be a nonempty list')
|
||||
job_id = job_id or str(uuid.uuid4())
|
||||
if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', job_id):
|
||||
raise ValueError('invalid job id')
|
||||
fingerprint = __import__('hashlib').sha256(
|
||||
job_fingerprint(kind, payload, list(resources)).encode()).hexdigest()
|
||||
existing = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
|
||||
if existing:
|
||||
if existing['fingerprint'] != fingerprint or existing['task'] != task_id:
|
||||
raise ValueError('job id reused with different fingerprint')
|
||||
return dict(existing), True
|
||||
now = time.time()
|
||||
db.execute(
|
||||
'''INSERT INTO jobs(id,task,kind,payload,resources,state,priority,created,fingerprint)
|
||||
VALUES(?,?,?,?,?,?,?,?,?)''',
|
||||
(job_id, task_id, kind, json.dumps(payload, ensure_ascii=False, sort_keys=True),
|
||||
json.dumps(list(resources)), JOB_PENDING, int(priority), now, fingerprint))
|
||||
row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
|
||||
return dict(row), False
|
||||
|
||||
|
||||
def list_jobs(db, state=None, limit=50):
|
||||
if state:
|
||||
rows = db.execute(
|
||||
'SELECT * FROM jobs WHERE state=? ORDER BY priority ASC, generation ASC LIMIT ?',
|
||||
(state, limit)).fetchall()
|
||||
else:
|
||||
rows = db.execute(
|
||||
'SELECT * FROM jobs ORDER BY CASE state '
|
||||
"WHEN 'pending' THEN 0 WHEN 'claimed' THEN 1 WHEN 'running' THEN 2 "
|
||||
"WHEN 'failed' THEN 3 ELSE 4 END, priority ASC, generation ASC LIMIT ?",
|
||||
(limit,)).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def cancel_job(db, job_id, reason='cancelled'):
|
||||
row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
|
||||
if not row:
|
||||
raise KeyError('unknown job')
|
||||
if row['state'] in JOB_TERMINAL:
|
||||
return dict(row), True
|
||||
if row['state'] == JOB_RUNNING:
|
||||
raise RuntimeError('cannot cancel a running job; recover the run first')
|
||||
db.execute(
|
||||
"UPDATE jobs SET state=?,finished_at=?,result=? WHERE id=?",
|
||||
(JOB_CANCELLED, time.time(), json.dumps({'reason': reason}), job_id))
|
||||
return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()), False
|
||||
|
||||
|
||||
def claim_next_job(db, worker_id, kinds=None):
|
||||
"""Atomically claim the highest-priority pending job. Returns None if empty."""
|
||||
clauses = ["state=?", JOB_PENDING]
|
||||
sql = 'SELECT * FROM jobs WHERE state=? '
|
||||
args = [JOB_PENDING]
|
||||
if kinds:
|
||||
placeholders = ','.join('?' for _ in kinds)
|
||||
sql += f'AND kind IN ({placeholders}) '
|
||||
args.extend(kinds)
|
||||
sql += 'ORDER BY priority ASC, generation ASC LIMIT 1'
|
||||
row = db.execute(sql, args).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
db.execute(
|
||||
"UPDATE jobs SET state=?,claimed_at=?,claimed_by=? WHERE id=? AND state=?",
|
||||
(JOB_CLAIMED, time.time(), worker_id, row['id'], JOB_PENDING))
|
||||
claimed = db.execute('SELECT * FROM jobs WHERE id=?', (row['id'],)).fetchone()
|
||||
if not claimed or claimed['state'] != JOB_CLAIMED or claimed['claimed_by'] != worker_id:
|
||||
return None
|
||||
return dict(claimed)
|
||||
|
||||
|
||||
def mark_job(db, job_id, state, result=None, run_id=None):
|
||||
if state not in (JOB_RUNNING, JOB_COMPLETED, JOB_FAILED, JOB_PENDING, JOB_CLAIMED):
|
||||
raise ValueError('invalid job state')
|
||||
finished = time.time() if state in (JOB_COMPLETED, JOB_FAILED) else None
|
||||
db.execute(
|
||||
'UPDATE jobs SET state=?,result=?,run_id=COALESCE(?,run_id),finished_at=COALESCE(?,finished_at) WHERE id=?',
|
||||
(state, json.dumps(result) if result is not None else None, run_id, finished, job_id))
|
||||
return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())
|
||||
|
||||
|
||||
def plan_deploy(db, task_id, env, expected_sha, operation_id=None, extra=None):
|
||||
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_sha):
|
||||
raise ValueError('expected_sha must be a git object name')
|
||||
operation_id = operation_id or str(uuid.uuid4())
|
||||
now = time.time()
|
||||
receipt = {
|
||||
'task': task_id,
|
||||
'env': env,
|
||||
'expected_sha': expected_sha,
|
||||
'events': [{'at': now, 'phase': 'planned'}],
|
||||
**(extra or {}),
|
||||
}
|
||||
existing = db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()
|
||||
if existing:
|
||||
if existing['task'] != task_id or existing['expected_sha'] != expected_sha:
|
||||
raise ValueError('operation id reused with different identity')
|
||||
return dict(existing), True
|
||||
db.execute(
|
||||
'''INSERT INTO operations(id,task,kind,phase,env,expected_sha,runtime_sha,receipt,created,updated)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?)''',
|
||||
(operation_id, task_id, 'deploy', 'planned', env, expected_sha, None,
|
||||
json.dumps(receipt, ensure_ascii=False), now, now))
|
||||
return dict(db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()), False
|
||||
|
||||
|
||||
def advance_deploy(db, operation_id, phase, runtime_sha=None, detail=None):
|
||||
if phase not in DEPLOY_PHASES:
|
||||
raise ValueError(f'invalid phase: {phase}')
|
||||
row = db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()
|
||||
if not row:
|
||||
raise KeyError('unknown operation')
|
||||
allowed = {
|
||||
'planned': {'deploying', 'failed', 'recovery_required'},
|
||||
'deploying': {'accepting', 'failed', 'recovery_required'},
|
||||
'accepting': {'passed', 'failed', 'recovery_required'},
|
||||
'failed': {'recovery_required', 'recovered'},
|
||||
'recovery_required': {'recovered', 'deploying'},
|
||||
'passed': set(),
|
||||
'recovered': {'deploying'},
|
||||
}
|
||||
if phase not in allowed.get(row['phase'], set()) and phase != row['phase']:
|
||||
raise RuntimeError(f'illegal transition {row["phase"]} -> {phase}')
|
||||
receipt = json.loads(row['receipt'])
|
||||
now = time.time()
|
||||
receipt.setdefault('events', []).append({'at': now, 'phase': phase, 'detail': detail})
|
||||
if runtime_sha:
|
||||
receipt['runtime_sha'] = runtime_sha
|
||||
db.execute(
|
||||
'UPDATE operations SET phase=?,runtime_sha=COALESCE(?,runtime_sha),receipt=?,updated=? WHERE id=?',
|
||||
(phase, runtime_sha, json.dumps(receipt, ensure_ascii=False), now, operation_id))
|
||||
return dict(db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone())
|
||||
|
||||
|
||||
def list_operations(db, task=None, limit=50):
|
||||
if task:
|
||||
rows = db.execute(
|
||||
'SELECT * FROM operations WHERE task=? ORDER BY updated DESC LIMIT ?',
|
||||
(task, limit)).fetchall()
|
||||
else:
|
||||
rows = db.execute(
|
||||
'SELECT * FROM operations ORDER BY updated DESC LIMIT ?', (limit,)).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def list_doc_registry(db):
|
||||
ensure_schema(db)
|
||||
return [dict(r) for r in db.execute('SELECT * FROM doc_registry ORDER BY path').fetchall()]
|
||||
|
||||
|
||||
def managed_doc_paths(db):
|
||||
return {row['path'] for row in list_doc_registry(db)}
|
||||
|
||||
|
||||
def project_cleanup_markdown(entries):
|
||||
"""Render cleanup visibility projection from structured entries."""
|
||||
lines = [
|
||||
'# Cleanup visibility projection',
|
||||
'',
|
||||
'> Generated by `agentctl docs project --kind cleanup`. Not a lock; machine `.locks` is authoritative.',
|
||||
'',
|
||||
'| path | branch | reason | cleanup_owner |',
|
||||
'| --- | --- | --- | --- |',
|
||||
]
|
||||
if not entries:
|
||||
lines.append('| — | — | none_pending | — |')
|
||||
else:
|
||||
for item in entries:
|
||||
path = str(item.get('path') or '').replace('|', '\\|')
|
||||
branch = str(item.get('branch') or '').replace('|', '\\|')
|
||||
reason = str(item.get('reason') or '').replace('|', '\\|')
|
||||
owner = str(item.get('cleanup_owner') or 'independent-maintainer').replace('|', '\\|')
|
||||
lines.append(f'| {path} | {branch} | {reason} | {owner} |')
|
||||
lines.append('')
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def assess_cleanup_entry(main, entry, control_fn=None, process_cwd_fn=None):
|
||||
"""Return a visibility dict; never deletes. Failures become reason strings."""
|
||||
from pathlib import Path as P
|
||||
branch_ref = entry.get('branch') or ''
|
||||
branch = branch_ref.replace('refs/heads/', '') if branch_ref.startswith('refs/heads/') else ''
|
||||
path = P(entry.get('worktree') or '')
|
||||
result = {
|
||||
'path': str(path),
|
||||
'branch': branch or None,
|
||||
'eligible': False,
|
||||
'reason': 'unknown',
|
||||
'checks': {},
|
||||
}
|
||||
if not branch or not path:
|
||||
result['reason'] = 'missing_branch_or_path'
|
||||
return result
|
||||
if 'locked' in entry:
|
||||
result['reason'] = 'git_worktree_locked'
|
||||
return result
|
||||
if 'prunable' in entry:
|
||||
result['reason'] = 'git_worktree_prunable'
|
||||
return result
|
||||
if not path.is_dir():
|
||||
result['reason'] = 'path_missing'
|
||||
return result
|
||||
|
||||
def git_c(*args):
|
||||
import subprocess
|
||||
return subprocess.run(['git', '-C', str(path if args[0] != '--main' else main), *([a for a in args if a != '--main'])],
|
||||
capture_output=True, text=True)
|
||||
|
||||
# Use explicit git -C for main vs path.
|
||||
import subprocess
|
||||
|
||||
def run(cwd, *args):
|
||||
return subprocess.run(['git', '-C', str(cwd), *args], capture_output=True, text=True)
|
||||
|
||||
actual = run(path, 'symbolic-ref', '--quiet', 'HEAD')
|
||||
if actual.returncode or actual.stdout.strip() != f'refs/heads/{branch}':
|
||||
result['reason'] = 'branch_identity_mismatch'
|
||||
result['checks']['head_ref'] = actual.stdout.strip()
|
||||
return result
|
||||
head = run(path, 'rev-parse', 'HEAD').stdout.strip()
|
||||
ancestor = run(main, 'merge-base', '--is-ancestor', head, 'refs/heads/dev')
|
||||
if ancestor.returncode:
|
||||
result['reason'] = 'head_not_ancestor_of_dev'
|
||||
result['checks']['head'] = head
|
||||
return result
|
||||
result['checks']['head'] = head
|
||||
for marker in ('MERGE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD', 'rebase-merge', 'rebase-apply', 'BISECT_START'):
|
||||
marker_path = run(path, 'rev-parse', '--path-format=absolute', '--git-path', marker).stdout.strip()
|
||||
if marker_path and Path(marker_path).exists():
|
||||
result['reason'] = f'unfinished_git_op:{marker}'
|
||||
return result
|
||||
status = run(path, 'status', '--porcelain=v1', '--untracked-files=all', '--ignored=matching')
|
||||
if status.stdout.strip():
|
||||
result['reason'] = 'dirty_or_ignored_retained'
|
||||
result['checks']['status_bytes'] = len(status.stdout)
|
||||
return result
|
||||
if control_fn is not None:
|
||||
try:
|
||||
control_fn(path)
|
||||
result['checks']['registry'] = 'ok'
|
||||
except Exception as exc: # noqa: BLE001 — visibility aggregates any refusal
|
||||
result['reason'] = f'registry:{exc}'
|
||||
return result
|
||||
if process_cwd_fn is not None:
|
||||
try:
|
||||
process_cwd_fn(path)
|
||||
result['checks']['process_cwd'] = 'ok'
|
||||
except Exception as exc: # noqa: BLE001
|
||||
result['reason'] = f'process_cwd:{exc}'
|
||||
return result
|
||||
elif not process_platform_ready():
|
||||
result['reason'] = 'process_cwd:platform_deferred_macos_or_no_proc'
|
||||
result['checks']['process_cwd'] = 'deferred'
|
||||
return result
|
||||
result['eligible'] = True
|
||||
result['reason'] = 'eligible'
|
||||
return result
|
||||
|
||||
|
||||
def build_command_for_job(kind, payload):
|
||||
"""Map a durable job to an argv list relative to agentctl semantics."""
|
||||
if kind == 'command':
|
||||
command = payload.get('command')
|
||||
if not isinstance(command, list) or not command or any(not isinstance(c, str) for c in command):
|
||||
raise ValueError('command job requires string argv list')
|
||||
return command
|
||||
if kind == 'validate':
|
||||
command = payload.get('command')
|
||||
if not isinstance(command, list) or not command:
|
||||
raise ValueError('validate job requires command argv')
|
||||
return ['validate-proxy', *command] # resolved by dispatcher
|
||||
if kind == 'integrate':
|
||||
expected = payload.get('expected_dev')
|
||||
if not isinstance(expected, str) or not expected:
|
||||
raise ValueError('integrate job requires expected_dev')
|
||||
return ['integrate-proxy', expected]
|
||||
if kind == 'deploy':
|
||||
return ['deploy-proxy', payload.get('operation_id') or '']
|
||||
if kind == 'docs_apply':
|
||||
request = payload.get('request_path')
|
||||
if not isinstance(request, str):
|
||||
raise ValueError('docs_apply requires request_path')
|
||||
return ['docs-proxy', request]
|
||||
if kind == 'cleanup_report':
|
||||
return ['cleanup-report-proxy']
|
||||
raise ValueError(f'unsupported kind {kind}')
|
||||
@@ -0,0 +1,399 @@
|
||||
#!/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]
|
||||
@@ -24,7 +24,7 @@ class WorktreeTest(unittest.TestCase):
|
||||
self.root = Path(self.tmp.name) / "repo"
|
||||
self.root.mkdir()
|
||||
(self.root / "scripts").mkdir()
|
||||
for name in ("agent-worktree.sh", "agent_worktree.py", "agentctl.py", "agentctl"):
|
||||
for name in ("agent-worktree.sh", "agent_worktree.py", "agentctl.py", "agentctl", "agentctl_w4.py"):
|
||||
shutil.copy2(SOURCE / name, self.root / "scripts" / name)
|
||||
(self.root / ".gitignore").write_text(".worktrees/\n.locks/\nignored-evidence/\n")
|
||||
(self.root / "source.txt").write_text("base\n")
|
||||
|
||||
@@ -29,6 +29,7 @@ class AgentctlTests(unittest.TestCase):
|
||||
self.git("config", "user.email", "agentctl-test@example.invalid")
|
||||
(self.root / "scripts").mkdir()
|
||||
shutil.copy2(SOURCE, self.root / "scripts" / "agentctl.py")
|
||||
shutil.copy2(SOURCE.parent / "agentctl_w4.py", self.root / "scripts" / "agentctl_w4.py")
|
||||
(self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
|
||||
for directory in ("src/a", "src/b", ".doc/ops", ".doc/status"):
|
||||
(self.root / directory).mkdir(parents=True, exist_ok=True)
|
||||
@@ -723,6 +724,7 @@ class ProcessPlatformTests(unittest.TestCase):
|
||||
subprocess.run(["git", "config", "user.email", "t@example.invalid"], cwd=root, check=True, capture_output=True)
|
||||
(root / "scripts").mkdir()
|
||||
shutil.copy2(SOURCE, root / "scripts" / "agentctl.py")
|
||||
shutil.copy2(SOURCE.parent / "agentctl_w4.py", root / "scripts" / "agentctl_w4.py")
|
||||
(root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
|
||||
(root / ".doc/ops").mkdir(parents=True)
|
||||
(root / ".doc/ops/CURRENT.md").write_text("# c\n", encoding="utf-8")
|
||||
|
||||
@@ -19,7 +19,7 @@ class OpsGuardsTest(unittest.TestCase):
|
||||
self.root = Path(self.tmp.name) / "repo"
|
||||
self.root.mkdir()
|
||||
(self.root / "scripts").mkdir()
|
||||
for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl"):
|
||||
for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl", "agentctl_w4.py"):
|
||||
shutil.copy2(SOURCE / name, self.root / "scripts" / name)
|
||||
(self.root / ".gitignore").write_text(".worktrees/\n.locks/\nconfigs/.env.*\n!configs/.env.*.example\ndata/\n")
|
||||
(self.root / ".dockerignore").write_text(".doc\n.locks\n.worktrees\n")
|
||||
|
||||
Reference in New Issue
Block a user