281 lines
9.2 KiB
Python
Executable File
281 lines
9.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
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': '🧭',
|
|
'triager': '🧩',
|
|
'architect': '🏗️',
|
|
'implementer': '🛠️',
|
|
'reviewer': '🔍',
|
|
'security': '🔒',
|
|
'qa': '🧪',
|
|
'documenter': '📚',
|
|
}
|
|
|
|
GATE_FILES = {
|
|
'reviewer': 'reviewer.json',
|
|
'security': 'security.json',
|
|
'qa': 'qa.json',
|
|
'documenter': 'documenter.md',
|
|
'publish': 'publish.json',
|
|
'leader': 'leader-close.json',
|
|
}
|
|
|
|
|
|
def now_iso():
|
|
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
|
|
|
|
|
|
def load_json(path: Path, default=None):
|
|
if not path.exists():
|
|
return default
|
|
return json.loads(path.read_text(encoding='utf-8'))
|
|
|
|
|
|
def save_json(path: Path, payload):
|
|
path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + '\n', encoding='utf-8')
|
|
|
|
|
|
def load_role_emojis():
|
|
emojis = dict(DEFAULT_EMOJIS)
|
|
if not MATRIX_PATH.exists():
|
|
return emojis
|
|
current_role = None
|
|
for line in MATRIX_PATH.read_text(encoding='utf-8').splitlines():
|
|
match_role = re.match(r'^ ([a-z_]+):\s*$', line)
|
|
if match_role:
|
|
current_role = match_role.group(1)
|
|
continue
|
|
match_emoji = re.match(r'^\s{4}emoji:\s*["\']?(.*?)["\']?\s*$', line)
|
|
if match_emoji and current_role:
|
|
emojis[current_role] = match_emoji.group(1)
|
|
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,
|
|
'stage': 'idle',
|
|
'agent': 'leader',
|
|
'action': 'Sin ejecución activa',
|
|
'state': 'waiting',
|
|
'next_agent': 'leader',
|
|
'waiting_for': 'Seleccionar una feature pending y actualizar este estado',
|
|
'updated_at': now_iso(),
|
|
'timeline': [],
|
|
}
|
|
|
|
|
|
def load_status():
|
|
status = load_json(STATUS_PATH, default_status())
|
|
base = default_status()
|
|
for key, value in base.items():
|
|
status.setdefault(key, value)
|
|
if not isinstance(status.get('timeline'), list):
|
|
status['timeline'] = []
|
|
return status
|
|
|
|
|
|
def gate_status(feature_id):
|
|
gates = {}
|
|
if not feature_id:
|
|
return gates
|
|
feature_dir = ARTIFACTS_DIR / feature_id
|
|
for gate, filename in GATE_FILES.items():
|
|
path = feature_dir / filename
|
|
if not path.exists():
|
|
gates[gate] = 'pending'
|
|
continue
|
|
if gate == 'documenter':
|
|
gates[gate] = 'approved'
|
|
continue
|
|
try:
|
|
payload = json.loads(path.read_text(encoding='utf-8'))
|
|
wanted = 'PUBLISHED' if gate == 'publish' else 'APPROVED'
|
|
gates[gate] = 'approved' if payload.get('verdict') == wanted else 'present'
|
|
except Exception:
|
|
gates[gate] = 'invalid'
|
|
return gates
|
|
|
|
|
|
def render_gate(gate, state, emojis):
|
|
icon = {
|
|
'approved': '✅',
|
|
'pending': '⏳',
|
|
'present': '⚠️',
|
|
'invalid': '❌',
|
|
}.get(state, '•')
|
|
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()
|
|
feature_id = status.get('feature_id') or '—'
|
|
current_agent = status.get('agent', 'leader')
|
|
next_agent = status.get('next_agent') or '—'
|
|
gates = gate_status(status.get('feature_id'))
|
|
|
|
print('╔══════════════════════════════════════════════════════════════╗')
|
|
print('║ ARNES · Runtime Status ║')
|
|
print('╚══════════════════════════════════════════════════════════════╝')
|
|
print(f"Feature activa : {feature_id}")
|
|
print(f"Stage actual : {status.get('stage', '—')}")
|
|
print(f"Agente actual : {emojis.get(current_agent, '•')} {current_agent}")
|
|
print(f"Acción : {status.get('action', '—')}")
|
|
print(f"Estado : {status.get('state', '—')}")
|
|
print(f"Siguiente : {emojis.get(next_agent, '•')} {next_agent}")
|
|
print(f"Esperando : {status.get('waiting_for', '—')}")
|
|
print(f"Actualizado : {status.get('updated_at', '—')}")
|
|
print()
|
|
print('Gates')
|
|
if gates:
|
|
for gate in ['reviewer', 'security', 'qa', 'documenter', 'publish', 'leader']:
|
|
print(f" {render_gate(gate, gates.get(gate, 'pending'), emojis)}")
|
|
else:
|
|
print(' — Sin feature activa —')
|
|
print()
|
|
print('Timeline')
|
|
timeline = status.get('timeline', [])[-8:]
|
|
if not timeline:
|
|
print(' — Sin eventos —')
|
|
return
|
|
for item in timeline:
|
|
agent = item.get('agent', 'leader')
|
|
emoji = emojis.get(agent, '•')
|
|
ts = item.get('ts', '—')
|
|
stage = item.get('stage', '—')
|
|
state = item.get('state', '—')
|
|
message = item.get('message', '')
|
|
print(f" - {ts} · {emoji} {agent} · {stage} · {state} · {message}")
|
|
|
|
|
|
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
|
|
if args.stage is not None:
|
|
status['stage'] = args.stage
|
|
if args.agent is not None:
|
|
status['agent'] = args.agent
|
|
if args.action is not None:
|
|
status['action'] = args.action
|
|
if args.state is not None:
|
|
status['state'] = args.state
|
|
if args.next_agent is not None:
|
|
status['next_agent'] = args.next_agent
|
|
if args.waiting_for is not None:
|
|
status['waiting_for'] = args.waiting_for
|
|
|
|
status['updated_at'] = now_iso()
|
|
event_message = args.note or status.get('action') or 'Estado actualizado'
|
|
status['timeline'].append({
|
|
'ts': status['updated_at'],
|
|
'agent': status.get('agent', 'leader'),
|
|
'stage': status.get('stage', '—'),
|
|
'state': status.get('state', '—'),
|
|
'message': event_message,
|
|
})
|
|
status['timeline'] = status['timeline'][-20:]
|
|
save_json(STATUS_PATH, status)
|
|
show_status()
|
|
|
|
|
|
def reset_status(_args):
|
|
status = default_status()
|
|
status['updated_at'] = now_iso()
|
|
save_json(STATUS_PATH, status)
|
|
show_status()
|
|
|
|
|
|
def build_parser():
|
|
parser = argparse.ArgumentParser(description='Renderiza y actualiza el estado visible de ARNES.')
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
|
|
sub.add_parser('show', help='Muestra el panel visible de estado')
|
|
|
|
set_parser = sub.add_parser('set', help='Actualiza el estado runtime y añade evento a timeline')
|
|
set_parser.add_argument('--feature-id')
|
|
set_parser.add_argument('--stage')
|
|
set_parser.add_argument('--agent')
|
|
set_parser.add_argument('--action')
|
|
set_parser.add_argument('--state')
|
|
set_parser.add_argument('--next-agent')
|
|
set_parser.add_argument('--waiting-for')
|
|
set_parser.add_argument('--note')
|
|
|
|
sub.add_parser('reset', help='Resetea el estado runtime a idle')
|
|
return parser
|
|
|
|
|
|
def main():
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
if args.command == 'show':
|
|
show_status()
|
|
elif args.command == 'set':
|
|
set_status(args)
|
|
elif args.command == 'reset':
|
|
reset_status(args)
|
|
else:
|
|
parser.print_help()
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|