Compare commits

..
Author SHA1 Message Date
WeiCHEN fd6b642bec merge: feature/agent-governance-config-2026-09-23 (governance JSON policy) 2026-09-23 14:51:05 +08:00
WeiCHEN 33c7b9a260 feat(agentctl): externalize governance policy to JSON
Load configs/agent-governance.json from the main worktree with fail-closed
POLICY_* codes, and keep test fixtures writing the same contract file.
2026-09-23 14:50:56 +08:00
WeiCHEN 1cb8c6f4b3 merge: fix/agentctl-automation-converge-2026-09-23 (automation converge into w4) 2026-09-23 14:39:57 +08:00
WeiCHEN dd75f42c98 fix(agentctl): converge automation into w4 and drop fork
Port deploy phase transition guards and cleanup markdown projection into
agentctl_w4, delete the unused automation module, and pin illegal jumps
with a regression test.
2026-09-23 14:39:49 +08:00
WeiCHEN 21c1278c04 merge: coordination-db-daemon-impl-plan closeout (CURRENT+short-term) 2026-09-23 14:17:52 +08:00
11 changed files with 414 additions and 427 deletions
+1
View File
@@ -18,6 +18,7 @@
| owner | task | branch | worktree | mutex | status |
|---|---|---|---|---|---|
| Cursor | agent-governance-config-2026-09-23 | feature/agent-governance-config-2026-09-23 | .worktrees/agent-governance-config-2026-09-23 | agent-governance-config | in_progress |
| Cursor | coordination-db-daemon-impl-plan-2026-09-23 | feature/coordination-db-daemon-impl-plan-2026-09-23 | .worktrees/coordination-db-daemon-impl-plan-2026-09-23 | coordination-db-daemon-impl-plan | done |
| Cursor | coordination-db-daemon-2026-09-23 | feature/coordination-db-daemon-2026-09-23 | .worktrees/coordination-db-daemon-2026-09-23 | coordination-db-daemon | done |
| Cursor | integrate-pi-runner-response-size-r2-2026-09-23 | feature/integrate-pi-runner-response-size-2026-09-23 | .worktrees/integrate-pi-runner-response-size-2026-09-23 | git:dev | done |
@@ -0,0 +1,45 @@
---
title: "W1-U2:治理策略外部化为 agent-governance.json"
task_id: agent-governance-config-2026-09-23
status: in_progress
created: 2026-09-23
updated: 2026-09-23
owner: Cursor
task_type: implementation
task_tier: single
policy_revision: "agent-governance-2026-09-21-r1 @ 1cb8c6f4b3379ef1136972ab36c5e120829d11f8"
branch: feature/agent-governance-config-2026-09-23
worktree: .worktrees/agent-governance-config-2026-09-23
base_dev_sha: "1cb8c6f4b3379ef1136972ab36c5e120829d11f8"
write_scope: ["scripts", "configs", ".doc"]
allowed_environments: []
mutex: agent-governance-config
review_state: pending
integration_state: pending
deployment: not_required
cleanup: pending
cleanup_owner: "独立维护者"
cleanup_trigger: "合入 local_dev 且无进程占用后复查"
remote_sync: not_requested
---
# W1-U2:治理策略外部化
## 用户目标与授权
- 用户目标:`configs/agent-governance.json`(stdlib JSON)外部化 SHARED_DOCS / JOB_KINDS / doc_rules 等;`load_policy` fail-closed。
- 授权:W1 波次 U2;不 push。
- 交付终点:合入 local_dev;负例三类拒绝码可测。
## 步骤
1. [x] worktree + register + CURRENT 认领
2. [ ] 新建 `configs/agent-governance.json`
3. [ ] `load_policy` + 写路径前置校验;只读不创建
4. [ ] 测试 fixture 写入策略文件
5. [ ] 负例测试 + unittest 全绿 + integrate
## 相关链接
- [实施承接计划](../../remediation/2026-09-23-coordination-db-daemon-implementation-undertaking-plan.md)
- 前序 U1r tip:`1cb8c6f4`
@@ -0,0 +1,70 @@
---
title: "W1-U1r:agentctl automation 归一"
task_id: agentctl-automation-converge-2026-09-23
status: in_progress
created: 2026-09-23
updated: 2026-09-23
owner: Cursor
task_type: implementation
task_tier: single
policy_revision: "agent-governance-2026-09-21-r1 @ 21c1278c04566f90230c6e354121041f2c020214"
branch: fix/agentctl-automation-converge-2026-09-23
worktree: .worktrees/agentctl-automation-converge-2026-09-23
base_dev_sha: "21c1278c04566f90230c6e354121041f2c020214"
write_scope: ["scripts", ".doc"]
allowed_environments: []
mutex: agentctl-automation-converge
review_state: pending
integration_state: pending
deployment: not_required
cleanup: pending
cleanup_owner: "独立维护者"
cleanup_trigger: "合入 local_dev 且无进程占用后复查"
remote_sync: not_requested
---
# W1-U1r:agentctl automation 归一
## 用户目标与授权
- 用户目标:删除分叉 `scripts/agentctl_automation.py`,移植部署阶段转移校验与 cleanup Markdown 投影到 `agentctl_w4.py`。
- 授权:W1 波次实施计划 U1r;write_scope=`scripts`(+ short-term/CURRENT);不 push。
- 交付终点:代码合入 local_dev;五态可审。
## 移植与丢弃
### 已移植
- `deployment_set_phase` 合法转移边(`prepared→deploying→accepting→passed`;非法 `passed→deploying` 拒绝)。
- `project_cleanup_markdown` → `agentctl_w4.py`(供 W2-U5)。
### 已丢弃(检索证据)
| 符号 | 取代方 / 原因 |
|---|---|
| `assess_cleanup_entry` | `agentctl.py` `cleanup_status` 8 gate |
| `process_platform_ready` | W1-B macOS 进程层三值 `alive_status` |
| `build_command_for_job` | 无调用方的 `*-proxy` 占位 |
| automation `jobs` schema | W4 `jobs` 已 landed |
| `operations` / `doc_registry` 表 | 活库 `.tables` 无此二者 |
`DEFAULT_DOC_REGISTRY` per-doc 规则不在本包移植,交 U2 策略 JSON。
## 步骤与验收
1. [x] worktree + register + CURRENT 认领
2. [x] 移植 phase 校验与 `project_cleanup_markdown`
3. [x] 删除 `agentctl_automation.py`;`rg agentctl_automation` 为空;清理 0 字节 `.locks/state.db`
4. [x] 补 `test_deploy_phase_transition_rejects_illegal_jump`;`unittest discover` 104 ran / 2 skipped / OK
5. [ ] 原子提交 + 候选 validate/integrate
## 验证与证据
| 范围 | 命令 | 结果 |
|---|---|---|
| 全套治理脚本 | `python3 -B -m unittest discover -s scripts/tests -p 'test_*.py' -v` | Ran 104 / OK (skipped=2) |
## 相关链接
- [实施承接计划](../../remediation/2026-09-23-coordination-db-daemon-implementation-undertaking-plan.md)
- [W1 实施计划](file:///Users/weichen/.cursor/plans/w1_三包实施_43ec5550.plan.md)
+34
View File
@@ -0,0 +1,34 @@
{
"protocol_version": 1,
"min_protocol_version": 1,
"policy_revision": "agent-governance-2026-09-21-r1",
"shared_docs": [".doc/ops/CURRENT.md", ".doc/status/FEATURE-MATRIX.md"],
"projectable_docs": [".doc/ops/CURRENT.md", ".doc/status/FEATURE-MATRIX.md"],
"projectable_prefixes": [".doc/ops/", ".doc/status/", ".doc/remediation/"],
"job_kinds": [
"validate",
"integrate",
"with",
"deploy_prepare",
"deploy_run",
"deploy_accept",
"docs_project",
"cleanup_inspect",
"noop"
],
"doc_rules": {
".doc/ops/CURRENT.md": {
"own_row_only": true,
"slot_protected": true,
"header": "| owner | task |"
},
".doc/status/FEATURE-MATRIX.md": {
"evidence_for_done": true,
"header": "| 模块 |"
}
},
"reserved_record_ids": ["holder", "idle", "Cursor", "Codex", "-"],
"protected_sections": {
".doc/ops/CURRENT.md": ["## 集成槽位"]
}
}
+148 -13
View File
@@ -24,7 +24,36 @@ 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'}
POLICY_REL = 'configs/agent-governance.json'
DEFAULT_POLICY = {
'protocol_version': 1,
'min_protocol_version': 1,
'policy_revision': 'agent-governance-2026-09-21-r1',
'shared_docs': ['.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'],
'projectable_docs': ['.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'],
'projectable_prefixes': ['.doc/ops/', '.doc/status/', '.doc/remediation/'],
'job_kinds': [
'validate', 'integrate', 'with', 'deploy_prepare', 'deploy_run',
'deploy_accept', 'docs_project', 'cleanup_inspect', 'noop',
],
'doc_rules': {
'.doc/ops/CURRENT.md': {
'own_row_only': True,
'slot_protected': True,
'header': '| owner | task |',
},
'.doc/status/FEATURE-MATRIX.md': {
'evidence_for_done': True,
'header': '| 模块 |',
},
},
'reserved_record_ids': ['holder', 'idle', 'Cursor', 'Codex', '-'],
'protected_sections': {'.doc/ops/CURRENT.md': ['## 集成槽位']},
}
SHARED_DOCS = set(DEFAULT_POLICY['shared_docs'])
RESERVED_RECORD_IDS = set(DEFAULT_POLICY['reserved_record_ids'])
DOC_RULES = dict(DEFAULT_POLICY['doc_rules'])
PROTECTED_SECTIONS = {k: list(v) for k, v in DEFAULT_POLICY['protected_sections'].items()}
ACTIVE = ('running', 'recovery_required')
TERMINAL = ('completed', 'failed', 'recovered')
QUEUE_TERMINAL = ('completed', 'failed', 'cancelled')
@@ -69,6 +98,100 @@ def digest(value):
return hashlib.sha256(value.encode() if isinstance(value, str) else value).hexdigest()
def _policy_path(main: Path) -> Path:
return Path(main) / POLICY_REL
def _validate_policy_shape(data) -> dict:
if not isinstance(data, dict):
deny('POLICY_INVALID', 'Governance policy must be a JSON object')
required = (
'protocol_version', 'min_protocol_version', 'policy_revision',
'shared_docs', 'projectable_docs', 'projectable_prefixes', 'job_kinds',
'doc_rules', 'reserved_record_ids', 'protected_sections',
)
missing = [key for key in required if key not in data]
if missing:
deny('POLICY_INVALID', 'Governance policy missing required keys', keys=missing)
if not isinstance(data['protocol_version'], int) or not isinstance(data['min_protocol_version'], int):
deny('POLICY_INVALID', 'protocol_version and min_protocol_version must be integers')
if data['protocol_version'] < data['min_protocol_version']:
deny('POLICY_MISMATCH', 'protocol_version below min_protocol_version',
protocol_version=data['protocol_version'],
min_protocol_version=data['min_protocol_version'])
for key in ('shared_docs', 'projectable_docs', 'projectable_prefixes',
'job_kinds', 'reserved_record_ids'):
if not isinstance(data[key], list) or not all(isinstance(item, str) for item in data[key]):
deny('POLICY_INVALID', f'{key} must be a list of strings')
if not isinstance(data['doc_rules'], dict) or not isinstance(data['protected_sections'], dict):
deny('POLICY_INVALID', 'doc_rules and protected_sections must be objects')
if not isinstance(data['policy_revision'], str) or not data['policy_revision']:
deny('POLICY_INVALID', 'policy_revision must be a nonempty string')
# Structure must stay compatible with the builtin DEFAULT_POLICY contract.
if set(data['shared_docs']) != set(DEFAULT_POLICY['shared_docs']):
deny('POLICY_MISMATCH', 'shared_docs diverges from DEFAULT_POLICY contract')
if set(data['job_kinds']) != set(DEFAULT_POLICY['job_kinds']):
deny('POLICY_MISMATCH', 'job_kinds diverges from DEFAULT_POLICY contract')
if set(data['projectable_docs']) != set(DEFAULT_POLICY['projectable_docs']):
deny('POLICY_MISMATCH', 'projectable_docs diverges from DEFAULT_POLICY contract')
if list(data['projectable_prefixes']) != list(DEFAULT_POLICY['projectable_prefixes']):
deny('POLICY_MISMATCH', 'projectable_prefixes diverges from DEFAULT_POLICY contract')
return data
def apply_policy(policy: dict) -> dict:
"""Install a validated policy into module globals used by write paths."""
global SHARED_DOCS, RESERVED_RECORD_IDS, DOC_RULES, PROTECTED_SECTIONS
SHARED_DOCS = set(policy['shared_docs'])
RESERVED_RECORD_IDS = set(policy['reserved_record_ids'])
DOC_RULES = {key: dict(value) for key, value in policy['doc_rules'].items()}
PROTECTED_SECTIONS = {key: list(value) for key, value in policy['protected_sections'].items()}
w4.apply_policy(policy)
return policy
def load_policy(main: Path) -> dict:
"""Load governance policy from the main worktree. Fail-closed; never creates the file."""
path = _policy_path(main)
if not path.is_file():
deny('POLICY_MISSING', f'Governance policy missing at {POLICY_REL}', path=str(path))
try:
raw = path.read_text(encoding='utf-8')
data = json.loads(raw)
except (OSError, UnicodeError) as exc:
deny('POLICY_INVALID', f'Cannot read governance policy: {exc}', path=str(path))
except json.JSONDecodeError as exc:
deny('POLICY_INVALID', f'Governance policy is not valid JSON: {exc}', path=str(path))
policy = _validate_policy_shape(data)
return apply_policy(policy)
def policy_status(main: Path) -> dict:
"""Read-only policy probe for status/doctor. Never creates files."""
path = _policy_path(main)
if not path.is_file():
return {'status': 'missing', 'code': 'POLICY_MISSING', 'path': str(path)}
try:
data = json.loads(path.read_text(encoding='utf-8'))
policy = _validate_policy_shape(data)
apply_policy(policy)
return {
'status': 'ok',
'path': str(path),
'policy_revision': policy['policy_revision'],
'protocol_version': policy['protocol_version'],
}
except Denied as exc:
return {'status': 'denied', 'code': exc.code, 'message': exc.message, 'path': str(path),
**exc.details}
except (OSError, UnicodeError) as exc:
return {'status': 'denied', 'code': 'POLICY_INVALID', 'message': str(exc), 'path': str(path)}
def require_policy(repo) -> dict:
return load_policy(repo.main)
def git(path, *args, index=False, check=True, input=None):
env = os.environ.copy()
for key in GIT_CONTEXT:
@@ -879,7 +1002,7 @@ def docs_apply(repo, args):
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', '-'):
if record_id in RESERVED_RECORD_IDS:
deny('PATH_OUT_OF_SCOPE', 'Use task ids; slot ownership is not a document edit')
path = safe_path(repo.current, filename)
if filename not in before:
@@ -900,14 +1023,14 @@ def docs_apply(repo, args):
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)
for section in PROTECTED_SECTIONS.get(filename, []):
def slot(text, marker=section):
match = re.search(r'(?m)^' + re.escape(marker) + r'\s*\n(.*?)(?=^## |\Z)', text, re.S | re.M)
return match.group(1) if match else None
if slot(replacement) != slot(before[filename]):
deny('PATH_OUT_OF_SCOPE', 'Document admin cannot overwrite integration slot ownership')
# A layout update cannot bypass the feature-completion evidence rule.
if filename.endswith('FEATURE-MATRIX.md'):
if DOC_RULES.get(filename, {}).get('evidence_for_done') or 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()):
@@ -931,7 +1054,9 @@ def docs_apply(repo, args):
if live:
lines[live[0][0]] = replacement
else:
header = '| owner | task |' if filename.endswith('CURRENT.md') else '| 模块 |'
header = DOC_RULES.get(filename, {}).get('header')
if not header:
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')
@@ -1640,9 +1765,24 @@ def parser():
def main():
args=parser().parse_args();repo=Repo(args.repo)
write_actions = {
'init', 'task', 'with', 'docs', 'validate', 'integrate', 'queue',
'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup',
}
# status/doctor remain readable without a policy file and never create one.
if args.action in write_actions:
if args.action == 'docs' and getattr(args, 'doc_action', None) == 'snapshot':
pass # snapshot is read-only CAS helper
elif args.action == 'queue' and getattr(args, 'queue_action', None) == 'list':
pass
elif args.action == 'cleanup':
require_policy(repo) # status report still needs contract constants
else:
require_policy(repo)
if args.action=='init':initialize(repo)
elif args.action in ('status','doctor'):
identity=repo.identity()
policy=policy_status(repo.main)
try:
with repo.db(True) as db:
ensure_schema(db)
@@ -1658,7 +1798,7 @@ def main():
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})
'tasks':tasks,'runs':runs,'queue':queue,'policy':policy})
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')
@@ -1733,11 +1873,6 @@ def main():
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
-413
View File
@@ -1,413 +0,0 @@
#!/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}')
+50 -1
View File
@@ -75,11 +75,18 @@ DEPLOY_PHASES = (
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 apply_policy(policy: dict) -> None:
"""Apply governance policy fields that W4 owns (job kinds / projectable docs)."""
global JOB_KINDS, PROJECTABLE_DOCS, PROJECTABLE_PREFIXES
JOB_KINDS = frozenset(policy['job_kinds'])
PROJECTABLE_DOCS = frozenset(policy['projectable_docs'])
PROJECTABLE_PREFIXES = tuple(policy['projectable_prefixes'])
def ensure_schema(db: sqlite3.Connection) -> None:
db.executescript(SCHEMA_SQL)
@@ -312,6 +319,21 @@ def deployment_set_phase(db, deploy_id: str, phase: str, *, receipt: Optional[di
error: Optional[str] = None) -> dict:
if phase not in DEPLOY_PHASES:
raise ValueError(f'invalid deploy phase: {phase}')
current = deployment_get(db, deploy_id)
if current is None:
raise RuntimeError(f'unknown deploy: {deploy_id}')
from_phase = current['phase']
# Terminal passed has empty outbound set; recover may jump failed/recovery_required → passed.
allowed = {
'prepared': {'deploying', 'failed', 'recovery_required'},
'deploying': {'accepting', 'failed', 'recovery_required'},
'accepting': {'passed', 'failed', 'recovery_required'},
'failed': {'recovery_required', 'passed'},
'recovery_required': {'deploying', 'passed'},
'passed': set(),
}
if phase != from_phase and phase not in allowed.get(from_phase, set()):
raise RuntimeError(f'illegal transition {from_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,
@@ -532,6 +554,33 @@ def cleanup_visibility_report(
}
def project_cleanup_markdown(entries):
"""Render cleanup visibility projection from structured entries.
Ported for W2-U5 projection reuse; structured report remains authoritative.
"""
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 _darwin_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
"""Fail-closed cwd scan via lsof on Darwin (agentctl-macos process layer)."""
uid = os.getuid() if uid is None else uid
+3
View File
@@ -26,6 +26,9 @@ class WorktreeTest(unittest.TestCase):
(self.root / "scripts").mkdir()
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 / "configs").mkdir(parents=True, exist_ok=True)
shutil.copy2(SOURCE.parent / "configs" / "agent-governance.json",
self.root / "configs" / "agent-governance.json")
(self.root / ".gitignore").write_text(".worktrees/\n.locks/\nignored-evidence/\n")
(self.root / "source.txt").write_text("base\n")
self.env = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}
+43
View File
@@ -16,6 +16,13 @@ import unittest
SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl.py"
POLICY_SOURCE = SOURCE.parents[1] / "configs" / "agent-governance.json"
def install_governance_policy(root: pathlib.Path) -> None:
dest = root / "configs"
dest.mkdir(parents=True, exist_ok=True)
shutil.copy2(POLICY_SOURCE, dest / "agent-governance.json")
class AgentctlTests(unittest.TestCase):
@@ -30,6 +37,7 @@ class AgentctlTests(unittest.TestCase):
(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")
install_governance_policy(self.root)
(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)
@@ -747,6 +755,40 @@ class AgentctlTests(unittest.TestCase):
item = next(i for i in listed["items"] if i["id"] == "q-sched-1")
self.assertEqual(item["state"], "completed")
def test_policy_missing_rejects_write_but_status_stays_readonly(self):
self.cli("init")
policy = self.root / "configs" / "agent-governance.json"
policy.unlink()
self.assertFalse(policy.exists())
denied = self.cli("hooks", "install", check=False)
self.assertNotEqual(denied.returncode, 0)
self.assertIn("POLICY_MISSING", denied.stderr)
self.assertFalse(policy.exists())
status = json.loads(self.cli("status").stdout)
self.assertEqual(status["status"], "ok")
self.assertEqual(status["policy"]["code"], "POLICY_MISSING")
self.assertFalse(policy.exists())
doctor = json.loads(self.cli("doctor").stdout)
self.assertEqual(doctor["status"], "ok")
self.assertEqual(doctor["policy"]["code"], "POLICY_MISSING")
self.assertFalse(policy.exists())
def test_policy_invalid_json_rejects_write(self):
policy = self.root / "configs" / "agent-governance.json"
policy.write_text("{not-json", encoding="utf-8")
denied = self.cli("init", check=False)
self.assertNotEqual(denied.returncode, 0)
self.assertIn("POLICY_INVALID", denied.stderr)
def test_policy_protocol_too_low_rejects_write(self):
policy = self.root / "configs" / "agent-governance.json"
data = json.loads(policy.read_text(encoding="utf-8"))
data["protocol_version"] = 0
policy.write_text(json.dumps(data), encoding="utf-8")
denied = self.cli("init", check=False)
self.assertNotEqual(denied.returncode, 0)
self.assertIn("POLICY_MISMATCH", denied.stderr)
class ProcessPlatformTests(unittest.TestCase):
@classmethod
@@ -818,6 +860,7 @@ class ProcessPlatformTests(unittest.TestCase):
(root / "scripts").mkdir()
shutil.copy2(SOURCE, root / "scripts" / "agentctl.py")
shutil.copy2(SOURCE.parent / "agentctl_w4.py", root / "scripts" / "agentctl_w4.py")
install_governance_policy(root)
(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")
+18
View File
@@ -32,6 +32,9 @@ class AgentctlW4Tests(unittest.TestCase):
self.root.mkdir()
(self.root / "scripts").mkdir()
shutil.copy2(W4_SOURCE, self.root / "scripts" / "agentctl_w4.py")
(self.root / "configs").mkdir(parents=True, exist_ok=True)
shutil.copy2(SOURCE.parent / "configs" / "agent-governance.json",
self.root / "configs" / "agent-governance.json")
self.w4 = load_w4(self.root / "scripts" / "agentctl_w4.py")
self.db_path = Path(self.tmp.name) / "w4.sqlite"
self.db = sqlite3.connect(self.db_path)
@@ -151,6 +154,21 @@ class AgentctlW4Tests(unittest.TestCase):
self.assertEqual(receipt.get("recover_reason"),
"operator cleared after fake failure")
def test_deploy_phase_transition_rejects_illegal_jump(self):
deploy = self.w4.deployment_create(
self.db, deploy_id="dep-phase", task="phase-task",
env="dev", expected_commit="abcdef4", manifest=None, dry_run=True)
self.assertEqual(deploy["phase"], "prepared")
self.w4.deployment_set_phase(self.db, "dep-phase", "deploying")
self.w4.deployment_set_phase(self.db, "dep-phase", "accepting")
passed = self.w4.deployment_set_phase(self.db, "dep-phase", "passed")
self.assertEqual(passed["phase"], "passed")
with self.assertRaises(RuntimeError) as ctx:
self.w4.deployment_set_phase(self.db, "dep-phase", "deploying")
self.assertIn("illegal transition passed -> deploying", str(ctx.exception))
still = self.w4.deployment_get(self.db, "dep-phase")
self.assertEqual(still["phase"], "passed")
class DarwinCwdVisibilityTests(unittest.TestCase):
def setUp(self):
+2
View File
@@ -24,6 +24,8 @@ class OpsGuardsTest(unittest.TestCase):
(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")
(self.root / "configs").mkdir()
policy_src = SOURCE.parent / "configs" / "agent-governance.json"
shutil.copy2(policy_src, self.root / "configs" / "agent-governance.json")
for target in ("shared", "dev", "test", "main"):
(self.root / f"docker-compose.{target}.yml").write_text("services: {}\n")
(self.root / "configs" / f".env.{target}.example").write_text("EXAMPLE=1\n")