#!/usr/bin/env python3 import http.server import socketserver import json import subprocess import urllib.request import urllib.parse import os import time import glob PORT = 4420 GITEA_TOKEN = "03b6dad42ad8faa3e9e3e1f4ea1ee688ffd92591" GITEA_BASE_URL = "http://127.0.0.1:3000/api/v1" DIRECTORY = "/root/agent-rpg-poc" AVAILABLE_PROJECTS = [ {"slug": "camper-knowledge-graph", "name": "Camper Knowledge Graph", "domain": "ckg.cpfoertner.de"}, {"slug": "appsoluteweb-relaunch", "name": "Appsoluteweb Relaunch", "domain": "appsoluteweb.de"}, {"slug": "cinnamon-compass", "name": "Cinnamon Compass", "domain": "cinnamon.cpfoertner.de"}, {"slug": "mainscreen-party-framework", "name": "Mainscreen Party Framework", "domain": "mainscreen.cpfoertner.de"} ] class RealtimeRPGHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=DIRECTORY, **kwargs) def do_GET(self): parsed_url = urllib.parse.urlparse(self.path) if parsed_url.path == '/api/status': query = urllib.parse.parse_qs(parsed_url.query) target_project = query.get('project', ['camper-knowledge-graph'])[0] self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() data = self.collect_live_status(target_project) self.wfile.write(json.dumps(data).encode('utf-8')) else: super().do_GET() def collect_live_status(self, target_project): status = { "timestamp": time.time(), "target_project": target_project, "available_projects": AVAILABLE_PROJECTS, "bullmq": { "queues": {}, "total_waiting": 0, "total_active": 0, "status": "IDLE" }, "gitea": { "project": target_project, "open_prs_count": 0, "open_prs": [], "kanban_columns": { "planned": 0, "ready": 0, "in_progress": 0, "review": 0, "done": 0 }, "active_ci_jobs": 0, "ci_status": "PASSING", "connected": True }, "agents": { "coder1": { "name": f"Coder #{target_project[:4].upper()}", "state": "IDLE", "currentTask": f"Keine offenen Tasks in {target_project}", "lastAction": "Standby", "liveLog": [] }, "coder2": { "name": f"Tester #{target_project[:4].upper()}", "state": "IDLE", "currentTask": "Wartet auf Staging Pipeline", "lastAction": "Standby", "liveLog": [] }, "lead": { "name": f"Lead #{target_project[:4].upper()}", "state": "IDLE", "currentTask": f"Überwacht {target_project}", "lastAction": "Monitoring", "liveLog": [] }, "reviewer": { "name": f"QA Reviewer", "state": "IDLE", "currentTask": "Wartet auf PR Reviews", "lastAction": "Standby", "liveLog": [] } } } # 1. Collect BullMQ Live Stats filtered for target_project queues = ['coder-tasks', 'lead-tasks', 'pr-reviews', 'tester-tasks', 'deployments'] total_waiting = 0 total_active = 0 latest_active_tasks = {} try: for q in queues: try: cmd = ['docker', 'exec', 'openclaw-event-broker-valkey', 'valkey-cli'] waiting = subprocess.check_output(cmd + ['llen', f'bull:{q}:wait'], timeout=1).decode().strip() if not waiting or waiting == '0': waiting = subprocess.check_output(cmd + ['zcard', f'bull:{q}:waiting'], timeout=1).decode().strip() active = subprocess.check_output(cmd + ['llen', f'bull:{q}:active'], timeout=1).decode().strip() w_count = int(waiting or 0) a_count = int(active or 0) # Inspect active tasks to verify they belong to target_project if a_count > 0: active_id = subprocess.check_output(cmd + ['lindex', f'bull:{q}:active', '0'], timeout=1).decode().strip() if active_id: job_raw = subprocess.check_output(cmd + ['hget', f'bull:{q}:{active_id}', 'data'], timeout=1).decode().strip() try: parsed = json.loads(job_raw) if parsed.get("projectSlug") == target_project or target_project in str(parsed): latest_active_tasks[q] = parsed total_active += 1 except: if target_project in job_raw: latest_active_tasks[q] = {"title": job_raw[:100]} total_active += 1 status["bullmq"]["queues"][q] = {"waiting": w_count, "active": a_count} total_waiting += w_count except Exception: pass status["bullmq"]["total_waiting"] = total_waiting status["bullmq"]["total_active"] = total_active status["bullmq"]["status"] = "PROCESSING" if (total_active > 0 or total_waiting > 0) else "IDLE" except Exception as e: status["bullmq"]["error"] = str(e) # 2. Collect Gitea Live PRs & Issues for Whiteboard Kanban strictly for target_project try: # Check Gitea Actions API for actively running workflows active_runs_count = 0 try: req_runs = urllib.request.Request( f"{GITEA_BASE_URL}/repos/openclaw/{target_project}/actions/runs?limit=10", headers={'Authorization': f'token {GITEA_TOKEN}'} ) with urllib.request.urlopen(req_runs, timeout=2) as resp: runs_data = json.loads(resp.read().decode()) for r in runs_data.get("workflow_runs", []): if r.get("status") in ["running", "waiting", "queued"]: active_runs_count += 1 except Exception: pass status["gitea"]["active_ci_jobs"] = active_runs_count status["gitea"]["ci_status"] = "BUILDING" if active_runs_count > 0 else "PASSING" # Fetch Open Pull Requests req_prs = urllib.request.Request( f"{GITEA_BASE_URL}/repos/openclaw/{target_project}/pulls?state=open", headers={'Authorization': f'token {GITEA_TOKEN}'} ) with urllib.request.urlopen(req_prs, timeout=2) as resp: prs = json.loads(resp.read().decode()) status["gitea"]["open_prs_count"] = len(prs) for p in prs: status["gitea"]["open_prs"].append({ "number": p.get("number"), "title": p.get("title"), "user": p.get("user", {}).get("username"), "created_at": p.get("created_at") }) # Fetch All Issues for Real Kanban Board Sync req_issues = urllib.request.Request( f"{GITEA_BASE_URL}/repos/openclaw/{target_project}/issues?state=all&limit=50", headers={'Authorization': f'token {GITEA_TOKEN}'} ) with urllib.request.urlopen(req_issues, timeout=2) as resp: issues = json.loads(resp.read().decode()) kanban = {"planned": 0, "ready": 0, "in_progress": 0, "review": 0, "done": 0} kanban_items = {"planned": [], "ready": [], "in_progress": [], "review": [], "done": []} for iss in issues: labels = [l.get("name", "").lower() for l in iss.get("labels", [])] is_closed = iss.get("state") == "closed" is_pr = bool(iss.get("pull_request")) item_meta = { "number": iss.get("number"), "title": iss.get("title"), "is_pr": is_pr, "state": iss.get("state") } if is_closed: kanban["done"] += 1 if len(kanban_items["done"]) < 5: kanban_items["done"].append(item_meta) elif any("review" in l for l in labels) or is_pr: kanban["review"] += 1 if len(kanban_items["review"]) < 5: kanban_items["review"].append(item_meta) elif any("in-progress" in l or "in_progress" in l for l in labels) or ("coder-tasks" in latest_active_tasks): kanban["in_progress"] += 1 if len(kanban_items["in_progress"]) < 5: kanban_items["in_progress"].append(item_meta) elif any("ready" in l or "board/active" in l for l in labels): kanban["ready"] += 1 if len(kanban_items["ready"]) < 5: kanban_items["ready"].append(item_meta) else: kanban["planned"] += 1 if len(kanban_items["planned"]) < 5: kanban_items["planned"].append(item_meta) # Ensure active BullMQ coder task reflects in Kanban WIP if not already if status["bullmq"]["total_active"] > 0 and kanban["in_progress"] == 0: kanban["in_progress"] = status["bullmq"]["total_active"] status["gitea"]["kanban_columns"] = kanban status["gitea"]["kanban_items"] = kanban_items except Exception as e: status["gitea"]["connected"] = False status["gitea"]["error"] = str(e) # 3. Check OpenClaw Active Session Activity for target_project (Lead, Coder, Reviewer) lead_heartbeat_active = False lead_task_desc = "" lead_last_pulse_seconds = 999 lead_recent_logs = [] coder_session_active = False coder_task_desc = "" coder_recent_logs = [] try: search_patterns = [ f'/home/christoph/openclaw/data/openclaw/agents/{target_project}-coder/agent/codex-home/sessions/**/rollout-*.jsonl', f'/home/christoph/openclaw/data/openclaw/agents/{target_project}-lead/agent/codex-home/sessions/**/rollout-*.jsonl', f'/home/christoph/openclaw/data/openclaw/agents/projekt-lead/agent/codex-home/sessions/**/rollout-*.jsonl' ] all_rollout_files = [] for p in search_patterns: all_rollout_files.extend(glob.glob(p, recursive=True)) rollout_files = sorted(all_rollout_files, key=os.path.getmtime, reverse=True) for rf in rollout_files[:10]: age_seconds = round(time.time() - os.path.getmtime(rf), 1) with open(rf, 'r') as fp: lines = fp.readlines() if not lines: continue last_line_data = json.loads(lines[-1]) last_payload_type = last_line_data.get('payload', {}).get('type') is_completed = (last_payload_type in ['task_complete', 'turn_complete', 'stop']) # Extract logs recent_logs = [] session_prompt = "" session_role = "lead" for l in lines[-8:]: try: item = json.loads(l) p_payload = item.get('payload', {}) if p_payload.get('type') == 'message' and p_payload.get('role') == 'assistant': t = p_payload.get('content', [{}])[0].get('text', '') if t: recent_logs.append(f"🤖 {t[:90]}...") elif p_payload.get('type') == 'item_completed': i_name = p_payload.get('item', {}).get('type') if i_name: recent_logs.append(f"⚙️ Step: {i_name}") except: pass for l in lines: try: p = json.loads(l) if p.get('payload', {}).get('type') == 'message' and p.get('payload', {}).get('role') == 'user': txt = p.get('payload', {}).get('content', [{}])[0].get('text', '') if not txt.startswith(""): session_prompt = txt if "coder für issue" in txt.lower() or "du bist der coder" in txt.lower() or "openclaw-coder" in txt.lower(): session_role = "coder" elif "review" in txt.lower() and ("du bist der" in txt.lower() or "qa" in txt.lower()): session_role = "reviewer" except: pass # Map to specific agent if session_role == "coder" and not coder_session_active: if not is_completed and age_seconds < 60: coder_session_active = True coder_task_desc = session_prompt[:120] coder_recent_logs = recent_logs elif session_role == "lead" and not lead_heartbeat_active: lead_last_pulse_seconds = age_seconds if not is_completed and age_seconds < 60: lead_heartbeat_active = True lead_task_desc = session_prompt[:120] lead_recent_logs = recent_logs status["lead_heartbeat"] = { "last_pulse_seconds_ago": lead_last_pulse_seconds, "project": target_project, "is_beating": lead_heartbeat_active } except Exception as e: status["lead_heartbeat"] = {"error": str(e), "is_beating": False, "last_pulse_seconds_ago": 999} # 4. Derive Agent RPG States STRICTLY for target_project # Coder 1 if "coder-tasks" in latest_active_tasks: task = latest_active_tasks["coder-tasks"] status["agents"]["coder1"]["state"] = "WORKING" status["agents"]["coder1"]["currentTask"] = f"Task #{task.get('issueNumber', '')}: {task.get('title', 'Coding')}" status["agents"]["coder1"]["lastAction"] = "BullMQ: Active Code Session" status["agents"]["coder1"]["liveLog"] = ["git checkout feature-branch", "npm run test", "git commit -m 'feat: implementation'"] elif coder_session_active: status["agents"]["coder1"]["state"] = "WORKING" status["agents"]["coder1"]["currentTask"] = f"Aktiv: {coder_task_desc or 'Coding Session'}" status["agents"]["coder1"]["lastAction"] = "OpenClaw Coder Turn Active" status["agents"]["coder1"]["liveLog"] = coder_recent_logs or ["Implementing feature / bugfix..."] else: status["agents"]["coder1"]["state"] = "IDLE" status["agents"]["coder1"]["currentTask"] = f"Keine offenen Tasks in {target_project}" status["agents"]["coder1"]["lastAction"] = "Idle / Bereit" status["agents"]["coder1"]["liveLog"] = ["Wartet auf neue Queue-Aufgaben..."] # Coder 2 (Tester / Staging) if "tester-tasks" in latest_active_tasks or "deployments" in latest_active_tasks: task = latest_active_tasks.get("tester-tasks") or latest_active_tasks.get("deployments") status["agents"]["coder2"]["state"] = "WORKING" status["agents"]["coder2"]["currentTask"] = f"Staging Pipeline: {task.get('title', 'Testing')}" status["agents"]["coder2"]["lastAction"] = "Testing / Deploying" status["agents"]["coder2"]["liveLog"] = ["curl -s https://staging/health", "e2e tests passing"] else: status["agents"]["coder2"]["state"] = "IDLE" status["agents"]["coder2"]["currentTask"] = f"Staging Environment ({target_project}) ist Healthy" status["agents"]["coder2"]["lastAction"] = "Healthy" status["agents"]["coder2"]["liveLog"] = ["Staging Health Check: 200 OK"] # Lead Agent if "lead-tasks" in latest_active_tasks: task = latest_active_tasks["lead-tasks"] status["agents"]["lead"]["state"] = "WORKING" status["agents"]["lead"]["currentTask"] = f"Lead Task: {task.get('title', 'Orchestration')}" status["agents"]["lead"]["lastAction"] = "Delegating & Dispatching" status["agents"]["lead"]["liveLog"] = lead_recent_logs or ["Dispatching event to subagents"] elif lead_heartbeat_active: status["agents"]["lead"]["state"] = "WORKING" status["agents"]["lead"]["currentTask"] = f"💓 Heartbeat in {target_project}: {lead_task_desc or 'Prüft Backlog'}" status["agents"]["lead"]["lastAction"] = "Lead Active" status["agents"]["lead"]["liveLog"] = lead_recent_logs or ["Heartbeat execution in progress..."] else: status["agents"]["lead"]["state"] = "IDLE" status["agents"]["lead"]["currentTask"] = f"{status['gitea']['open_prs_count']} offene PRs in {target_project}. Heartbeat wartet." status["agents"]["lead"]["lastAction"] = "Standby" status["agents"]["lead"]["liveLog"] = ["Letzter Heartbeat abgeschlossen (task_complete)"] # Reviewer Agent if "pr-reviews" in latest_active_tasks: task = latest_active_tasks["pr-reviews"] status["agents"]["reviewer"]["state"] = "WORKING" status["agents"]["reviewer"]["currentTask"] = f"PR Review #{task.get('prNumber', '')}: {task.get('title', '')}" status["agents"]["reviewer"]["lastAction"] = "Security & Code Review" status["agents"]["reviewer"]["liveLog"] = ["Auditing PR diff...", "Running automated security scan"] else: status["agents"]["reviewer"]["state"] = "IDLE" status["agents"]["reviewer"]["currentTask"] = f"Wartet auf PR Reviews in {target_project} ({status['gitea']['open_prs_count']} PRs gelistet)" status["agents"]["reviewer"]["lastAction"] = "Idle / Bereit" status["agents"]["reviewer"]["liveLog"] = ["PR Watcher aktiv"] return status if __name__ == '__main__': socketserver.TCPServer.allow_reuse_address = True with socketserver.TCPServer(("", PORT), RealtimeRPGHandler) as httpd: print(f"Server läuft auf Port {PORT} mit Multi-Project API") httpd.serve_forever()