624 lines
33 KiB
Python
624 lines
33 KiB
Python
#!/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("<recommended_plugins>") 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()
|