weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(agentctl): converge automation into w4 and drop fork

Browse Source 
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.
This commit is contained in:
WeiCHEN
2026-09-23 14:39:49 +08:00
parent 21c1278c04
commit dd75f42c98
4 changed files with 127 additions and 413 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/agentctl_automation.py
-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}')
scripts/agentctl_w4.py
+42
View File
@@ -312,6 +312,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 +547,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
scripts/tests/test_agentctl_w4.py
+15
View File
@@ -151,6 +151,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):