#!/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_POST(self): parsed_url = urllib.parse.urlparse(self.path) if parsed_url.path == '/api/issue': content_length = int(self.headers.get('Content-Length', 0)) post_data = self.rfile.read(content_length).decode('utf-8') try: payload = json.loads(post_data) project = payload.get('project', 'camper-knowledge-graph') title = payload.get('title', '').strip() body = payload.get('body', 'Erstellt über Agent RPG Terminal').strip() if not title: self.send_response(400) self.end_headers() self.wfile.write(json.dumps({"error": "Title is required"}).encode('utf-8')) return req = urllib.request.Request( f"{GITEA_BASE_URL}/repos/openclaw/{project}/issues", data=json.dumps({"title": title, "body": body}).encode('utf-8'), headers={'Authorization': f'token {GITEA_TOKEN}', 'Content-Type': 'application/json'} ) with urllib.request.urlopen(req, timeout=3) as resp: resp_data = json.loads(resp.read().decode()) self.send_response(201) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps({"success": True, "issue": resp_data}).encode('utf-8')) except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode('utf-8')) else: self.send_response(404) self.end_headers() 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 }, "metrics": { "active_agents": 0, "token_velocity": 0, "system_health": 100 }, "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 and recent failure active_runs_count = 0 latest_run_failed = False 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()) runs = runs_data.get("workflow_runs", []) for r in runs: if r.get("status") in ["running", "waiting", "queued"]: active_runs_count += 1 if runs and runs[0].get("status") == "completed" and runs[0].get("conclusion") in ["failure", "cancelled"]: latest_run_failed = True except Exception: pass status["gitea"]["active_ci_jobs"] = active_runs_count if active_runs_count > 0: status["gitea"]["ci_status"] = "BUILDING" elif latest_run_failed: status["gitea"]["ci_status"] = "FAILURE" else: status["gitea"]["ci_status"] = "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 Subagent Runs & Session Activity via SQLite state & files lead_heartbeat_active = False lead_task_desc = "" lead_last_pulse_seconds = 999 lead_recent_logs = [] coder_sessions = [] reviewer_sessions = [] tester_sessions = [] active_coder_count = 0 active_reviewer_count = 0 active_tester_count = 0 proj_slugs = [target_project.lower(), target_project.lower().replace('-', '_')] if 'cinnamon' in target_project or 'cinamon' in target_project: proj_slugs.extend(['cinnamon-compass', 'cinamon-compass', 'cinnamon_compass', 'cinamon_compass', 'zimtkompass']) elif 'mainscreen' in target_project: proj_slugs.extend(['mainscreen-party-framework', 'mainscreen-party', 'mainscreen_party_framework', 'mainscreen_party', 'appsoluteweb']) elif 'appsoluteweb' in target_project: proj_slugs.extend(['appsoluteweb-relaunch', 'appsoluteweb_relaunch', 'appsoluteweb']) db_path = '/home/christoph/openclaw/data/openclaw/state/openclaw.sqlite' now_ms = time.time() * 1000 # A. Query OpenClaw SQLite state database if os.path.exists(db_path): try: import sqlite3 conn = sqlite3.connect(f'file:{db_path}?mode=ro', uri=True, timeout=2.0) conn.row_factory = sqlite3.Row # Lead Agent Activity from audit_events & session_state_heads try: lead_cur = conn.execute(''' SELECT occurred_at, session_key, action, status, tool_name, run_id FROM audit_events WHERE session_key LIKE '%:main' OR session_key LIKE '%-lead:%' ORDER BY sequence DESC LIMIT 40 ''') for row in lead_cur: skey = (row['session_key'] or '').lower() if any(p in skey for p in proj_slugs) or (target_project == 'camper-knowledge-graph' and 'lead' in skey): age_s = round((now_ms - row['occurred_at']) / 1000.0, 1) if lead_last_pulse_seconds == 999: lead_last_pulse_seconds = age_s lead_heartbeat_active = (age_s < 90) if len(lead_recent_logs) < 5: tname = row['tool_name'] or row['action'] or 'action' st = row['status'] or '' lead_recent_logs.append(f'⚙️ {tname} ({st}) [{age_s:.0f}s ago]') except Exception: pass # Subagent runs from subagent_runs & audit_events try: cur = conn.execute(''' SELECT run_id, child_session_key, controller_session_key, created_at, payload_json FROM subagent_runs ORDER BY rowid DESC LIMIT 60 ''') for r in cur: try: p = json.loads(r['payload_json']) except Exception: continue exec_info = p.get('execution', {}) exec_status = exec_info.get('status') ended_at = exec_info.get('endedAt', 0) started_at = exec_info.get('startedAt', r['created_at']) req_agent = str(p.get('requesterAgentId') or r['controller_session_key'] or '').lower() task = p.get('task', '') label = p.get('label', '') matches_proj = any(s in req_agent or s in task.lower() or s in label.lower() for s in proj_slugs) if not matches_proj: continue # Audit events for this subagent run audit_cur = conn.execute(''' SELECT occurred_at, action, status, tool_name FROM audit_events WHERE run_id = ? ORDER BY sequence DESC LIMIT 5 ''', (r['run_id'],)) audits = [dict(a) for a in audit_cur] last_audit_at = audits[0]['occurred_at'] if audits else started_at age_audit_sec = (now_ms - last_audit_at) / 1000.0 if last_audit_at else 9999 age_ended_sec = (now_ms - ended_at) / 1000.0 if ended_at else 9999 is_active = (exec_status == 'running') or (age_audit_sec < 60 and exec_status != 'terminal') or (age_ended_sec < 10) logs = [] for a in audits: tname = a.get('tool_name') or a.get('action') or 'step' st = a.get('status') or '' t_age = (now_ms - a['occurred_at']) / 1000.0 logs.append(f'⚙️ {tname} ({st}) [{t_age:.0f}s ago]') if not logs: logs = [f'🤖 {label or "Subagent Active"}'] lbl = label.lower() tsk = task.lower()[:120] if any(tsk.startswith(k) or k in tsk for k in ['du bist der coder', 'act as coder', 'act as the isolated coder', 'arbeite als coder']): role = 'coder' elif any(tsk.startswith(k) or k in tsk for k in ['act as an independent reviewer', 'unabhängige review', 'du bist der reviewer', 'du bist unabhängiger reviewer', 'review pr #']): role = 'reviewer' elif any(tsk.startswith(k) or k in tsk for k in ['act as tester', 'du bist der tester', 'staging test']): role = 'tester' elif any(k in lbl for k in ['re-review', 'qa reviewer', 'audit']) or ('review' in lbl and 'korrektur' not in lbl and 'findings' not in lbl and 'fix' not in lbl): role = 'reviewer' elif any(k in lbl for k in ['tester', 'staging', 'e2e pipeline']): role = 'tester' else: role = 'coder' sub_info = { 'run_id': r['run_id'], 'label': label, 'task': label or task[:120], 'is_active': is_active, 'age_audit_sec': round(age_audit_sec, 1), 'exec_status': exec_status, 'logs': logs, 'role': role } if role == 'reviewer': reviewer_sessions.append(sub_info) if is_active: active_reviewer_count += 1 elif role == 'tester': tester_sessions.append(sub_info) if is_active: active_tester_count += 1 else: coder_sessions.append(sub_info) if is_active: active_coder_count += 1 except Exception: pass conn.close() except Exception: pass # B. Fallback / supplementary check from rollout jsonl files if not coder_sessions: 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/{target_project}-lead/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[:15]: 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']) recent_logs = [] session_prompt = "" 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 Exception: pass user_prompts = [] 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("") and not txt.startswith("OpenClaw assembled context"): user_prompts.append(txt) except Exception: pass first_prompt = user_prompts[0] if user_prompts else "" if first_prompt.startswith("Du bist der Coder") or "Du bist der Coder für" in first_prompt: session_role = "coder" elif "review" in first_prompt.lower() and ("du bist der reviewer" in first_prompt.lower() or "qa" in first_prompt.lower()): session_role = "reviewer" else: session_role = "lead" if session_role == "coder": if not is_completed and age_seconds < 60: active_coder_count += 1 coder_sessions.append({ "label": session_prompt[:60] or "Coding Session", "task": session_prompt[:120], "logs": recent_logs, "age_audit_sec": age_seconds, "is_active": True, "exec_status": "running" }) 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 except Exception: pass status["lead_heartbeat"] = { "last_pulse_seconds_ago": lead_last_pulse_seconds, "project": target_project, "is_beating": lead_heartbeat_active } # 4. Derive Agent RPG States STRICTLY for target_project # Coder 1 (supports multi-instance count x2, x3) active_working_coders = [c for c in coder_sessions if c.get('is_active')] 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'"] status["agents"]["coder1"]["parallelCount"] = max(1, active_coder_count) elif active_working_coders: primary_coder = active_working_coders[0] status["agents"]["coder1"]["state"] = "WORKING" status["agents"]["coder1"]["currentTask"] = primary_coder.get('task') or primary_coder.get('label') or 'Coding Session' status["agents"]["coder1"]["lastAction"] = f"OpenClaw Subagent Coder ({len(active_working_coders)}x active)" status["agents"]["coder1"]["liveLog"] = primary_coder.get('logs') or ["Implementing feature / bugfix..."] status["agents"]["coder1"]["parallelCount"] = len(active_working_coders) elif coder_sessions: last_coder = coder_sessions[0] status["agents"]["coder1"]["state"] = "IDLE" status["agents"]["coder1"]["currentTask"] = f"Letzter Task: {last_coder.get('label') or last_coder.get('task')}" status["agents"]["coder1"]["lastAction"] = f"Abgeschlossen ({last_coder.get('age_audit_sec', 0)}s her)" status["agents"]["coder1"]["liveLog"] = last_coder.get('logs') or ["Bereit für neue Aufgaben"] status["agents"]["coder1"]["parallelCount"] = 0 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..."] status["agents"]["coder1"]["parallelCount"] = 0 # Coder 2 (Tester / Staging) active_working_testers = [t for t in tester_sessions if t.get('is_active')] 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"] elif active_working_testers: primary_tester = active_working_testers[0] status["agents"]["coder2"]["state"] = "WORKING" status["agents"]["coder2"]["currentTask"] = primary_tester.get('task') or primary_tester.get('label') or 'Testing Pipeline' status["agents"]["coder2"]["lastAction"] = "Active Tester Subagent" status["agents"]["coder2"]["liveLog"] = primary_tester.get('logs') or ["Running test suites..."] elif len(active_working_coders) >= 2: second_coder = active_working_coders[1] status["agents"]["coder2"]["state"] = "WORKING" status["agents"]["coder2"]["currentTask"] = second_coder.get('task') or second_coder.get('label') or 'Parallele Code-Session' status["agents"]["coder2"]["lastAction"] = "Active Subagent Coder #2" status["agents"]["coder2"]["liveLog"] = second_coder.get('logs') or ["Coding..."] 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 & Dispatching'}" status["agents"]["lead"]["lastAction"] = f"Lead Active ({lead_last_pulse_seconds}s pulse)" 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 ({lead_last_pulse_seconds}s her)." status["agents"]["lead"]["lastAction"] = "Standby" status["agents"]["lead"]["liveLog"] = lead_recent_logs or ["Letzter Heartbeat abgeschlossen"] # Reviewer Agent active_working_reviewers = [r for r in reviewer_sessions if r.get('is_active')] 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"] elif active_working_reviewers: primary_reviewer = active_working_reviewers[0] status["agents"]["reviewer"]["state"] = "WORKING" status["agents"]["reviewer"]["currentTask"] = primary_reviewer.get('task') or primary_reviewer.get('label') or 'PR Review' status["agents"]["reviewer"]["lastAction"] = "Active Reviewer Subagent" status["agents"]["reviewer"]["liveLog"] = primary_reviewer.get('logs') or ["Reviewing PR..."] 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"] status["agents"]["coder_instances"] = coder_sessions # Calculate live system metrics active_count = sum(1 for a in status["agents"].values() if isinstance(a, dict) and a.get("state") == "WORKING") status["metrics"]["active_agents"] = active_count status["metrics"]["token_velocity"] = (active_count * 185) + (total_active * 240) status["metrics"]["system_health"] = 90 if status["gitea"]["ci_status"] == "FAILURE" else 100 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()