|
|
|
@@ -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}')
|
WeiCHEN