refactor: make ARNES external-repo based with ticket publish flow

This commit is contained in:
rikrdo
2026-05-18 00:26:32 +02:00
parent 3ff9b70e4c
commit b396b6d3c9
101 changed files with 810 additions and 6140 deletions

View File

@@ -8,7 +8,9 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
STATUS_PATH = ROOT / 'work' / 'runtime-status.json'
MATRIX_PATH = ROOT / 'harness' / 'agents.matrix.yml'
WORKFLOW_PATH = ROOT / 'harness' / 'workflow.stages.yml'
ARTIFACTS_DIR = ROOT / 'work' / 'artifacts'
VALID_RUNTIME_STATES = {'idle', 'waiting', 'running', 'blocked', 'done'}
DEFAULT_EMOJIS = {
'leader': '🧭',
@@ -26,6 +28,7 @@ GATE_FILES = {
'security': 'security.json',
'qa': 'qa.json',
'documenter': 'documenter.md',
'publish': 'publish.json',
'leader': 'leader-close.json',
}
@@ -60,6 +63,28 @@ def load_role_emojis():
return emojis
def load_roles():
roles = []
if not MATRIX_PATH.exists():
return roles
for line in MATRIX_PATH.read_text(encoding='utf-8').splitlines():
match_role = re.match(r'^ ([a-z_]+):\s*$', line)
if match_role:
roles.append(match_role.group(1))
return roles
def load_stage_names():
stages = []
if not WORKFLOW_PATH.exists():
return stages
for line in WORKFLOW_PATH.read_text(encoding='utf-8').splitlines():
match_stage = re.match(r'^ - name:\s*([a-z_]+)\s*$', line)
if match_stage:
stages.append(match_stage.group(1))
return stages
def default_status():
return {
'feature_id': None,
@@ -99,7 +124,8 @@ def gate_status(feature_id):
continue
try:
payload = json.loads(path.read_text(encoding='utf-8'))
gates[gate] = 'approved' if payload.get('verdict') == 'APPROVED' else 'present'
wanted = 'PUBLISHED' if gate == 'publish' else 'APPROVED'
gates[gate] = 'approved' if payload.get('verdict') == wanted else 'present'
except Exception:
gates[gate] = 'invalid'
return gates
@@ -115,10 +141,25 @@ def render_gate(gate, state, emojis):
label = {
'leader': 'close',
'documenter': 'docs',
'publish': 'publish',
}.get(gate, gate)
return f"{icon} {emojis.get(gate, '')} {label}: {state.upper()}"
def validate_runtime_args(args):
roles = set(load_roles()) or set(DEFAULT_EMOJIS)
stages = set(load_stage_names()) | {'idle'}
if args.agent is not None and args.agent not in roles:
raise SystemExit(f"Invalid agent: {args.agent}. Allowed: {', '.join(sorted(roles))}")
if args.next_agent is not None and args.next_agent not in roles:
raise SystemExit(f"Invalid next-agent: {args.next_agent}. Allowed: {', '.join(sorted(roles))}")
if args.stage is not None and args.stage not in stages:
raise SystemExit(f"Invalid stage: {args.stage}. Allowed: {', '.join(sorted(stages))}")
if args.state is not None and args.state not in VALID_RUNTIME_STATES:
raise SystemExit(f"Invalid state: {args.state}. Allowed: {', '.join(sorted(VALID_RUNTIME_STATES))}")
def show_status():
status = load_status()
emojis = load_role_emojis()
@@ -141,7 +182,7 @@ def show_status():
print()
print('Gates')
if gates:
for gate in ['reviewer', 'security', 'qa', 'documenter', 'leader']:
for gate in ['reviewer', 'security', 'qa', 'documenter', 'publish', 'leader']:
print(f" {render_gate(gate, gates.get(gate, 'pending'), emojis)}")
else:
print(' — Sin feature activa —')
@@ -162,6 +203,7 @@ def show_status():
def set_status(args):
validate_runtime_args(args)
status = load_status()
if args.feature_id is not None:
status['feature_id'] = args.feature_id or None