diff --git a/server.py b/server.py index 534dbc4..8842085 100644 --- a/server.py +++ b/server.py @@ -278,104 +278,242 @@ class RealtimeRPGHandler(http.server.SimpleHTTPRequestHandler): status["gitea"]["connected"] = False status["gitea"]["error"] = str(e) - # 3. Check OpenClaw Active Session Activity for target_project (Lead, Coder, Reviewer) + # 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 - 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)) + 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']) - 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']) - - # Extract logs - recent_logs = [] - session_prompt = "" - session_role = "lead" + db_path = '/home/christoph/openclaw/data/openclaw/state/openclaw.sqlite' + now_ms = time.time() * 1000 - for l in lines[-8:]: + # 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: - 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 + p = json.loads(r['payload_json']) + except Exception: + continue - # Determine true session role by inspecting the actual initial user prompt - 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: - pass + 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', '') - 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" - session_prompt = first_prompt - elif "review" in first_prompt.lower() and ("du bist der reviewer" in first_prompt.lower() or "qa" in first_prompt.lower()): - session_role = "reviewer" - session_prompt = first_prompt - else: - session_role = "lead" - session_prompt = first_prompt or "Heartbeat / Backlog Monitoring" + 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 - # Map to specific agent - if session_role == "coder": - if not is_completed and age_seconds < 60: - active_coder_count += 1 - coder_sessions.append({ - "task": session_prompt[:120], - "logs": recent_logs, - "age": age_seconds - }) - 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 + # 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 - 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} + 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" @@ -383,13 +521,20 @@ class RealtimeRPGHandler(http.server.SimpleHTTPRequestHandler): 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_coder_count > 0: + elif active_working_coders: + primary_coder = active_working_coders[0] status["agents"]["coder1"]["state"] = "WORKING" - primary_coder = coder_sessions[0] - status["agents"]["coder1"]["currentTask"] = f"Aktiv ({active_coder_count}x): {primary_coder['task'] or 'Coding Session'}" - status["agents"]["coder1"]["lastAction"] = f"OpenClaw Coder Turn ({active_coder_count} parallel)" - status["agents"]["coder1"]["liveLog"] = primary_coder["logs"] or ["Implementing feature / bugfix..."] - status["agents"]["coder1"]["parallelCount"] = active_coder_count + 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}" @@ -398,12 +543,25 @@ class RealtimeRPGHandler(http.server.SimpleHTTPRequestHandler): 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" @@ -419,22 +577,29 @@ class RealtimeRPGHandler(http.server.SimpleHTTPRequestHandler): 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"]["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." + 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"] = ["Letzter Heartbeat abgeschlossen (task_complete)"] + 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)"