fix: integrate OpenClaw SQLite state & audit events for accurate live coder subagent tracking

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:
+        proj_slugs = [target_project.lower(), target_project.lower().replace('-', '_')]
-            search_patterns = [
+        if 'cinnamon' in target_project or 'cinamon' in target_project:
-                f'/home/christoph/openclaw/data/openclaw/agents/{target_project}-coder/agent/codex-home/sessions/**/rollout-*.jsonl',
+            proj_slugs.extend(['cinnamon-compass', 'cinamon-compass', 'cinnamon_compass', 'cinamon_compass', 'zimtkompass'])
-                f'/home/christoph/openclaw/data/openclaw/agents/{target_project}-lead/agent/codex-home/sessions/**/rollout-*.jsonl',
+        elif 'mainscreen' in target_project:
-                f'/home/christoph/openclaw/data/openclaw/agents/projekt-lead/agent/codex-home/sessions/**/rollout-*.jsonl'
+            proj_slugs.extend(['mainscreen-party-framework', 'mainscreen-party', 'mainscreen_party_framework', 'mainscreen_party', 'appsoluteweb'])
-            ]
+        elif 'appsoluteweb' in target_project:
-            all_rollout_files = []
+            proj_slugs.extend(['appsoluteweb-relaunch', 'appsoluteweb_relaunch', 'appsoluteweb'])
-            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)
+        db_path = '/home/christoph/openclaw/data/openclaw/state/openclaw.sqlite'
-            
+        now_ms = time.time() * 1000
-            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"
 
-                    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 = json.loads(r['payload_json'])
-                            p_payload = item.get('payload', {})
+                        except Exception:
-                            if p_payload.get('type') == 'message' and p_payload.get('role') == 'assistant':
+                            continue
-                                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
 
-                    # Determine true session role by inspecting the actual initial user prompt
+                        exec_info = p.get('execution', {})
-                    user_prompts = []
+                        exec_status = exec_info.get('status')
-                    for l in lines:
+                        ended_at = exec_info.get('endedAt', 0)
-                        try:
+                        started_at = exec_info.get('startedAt', r['created_at'])
-                            p = json.loads(l)
+                        req_agent = str(p.get('requesterAgentId') or r['controller_session_key'] or '').lower()
-                            if p.get('payload', {}).get('type') == 'message' and p.get('payload', {}).get('role') == 'user':
+                        task = p.get('task', '')
-                                txt = p.get('payload', {}).get('content', [{}])[0].get('text', '')
+                        label = p.get('label', '')
-                                if not txt.startswith("<recommended_plugins>") and not txt.startswith("OpenClaw assembled context"):
-                                    user_prompts.append(txt)
-                        except:
-                            pass
 
-                    first_prompt = user_prompts[0] if user_prompts else ""
+                        matches_proj = any(s in req_agent or s in task.lower() or s in label.lower() for s in proj_slugs)
-                    if first_prompt.startswith("Du bist der Coder") or "Du bist der Coder für" in first_prompt:
+                        if not matches_proj:
-                        session_role = "coder"
+                            continue
-                        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"
 
-                    # Map to specific agent
+                        # Audit events for this subagent run
-                    if session_role == "coder":
+                        audit_cur = conn.execute('''
-                        if not is_completed and age_seconds < 60:
+                            SELECT occurred_at, action, status, tool_name
-                            active_coder_count += 1
+                            FROM audit_events
-                            coder_sessions.append({
+                            WHERE run_id = ?
-                                "task": session_prompt[:120],
+                            ORDER BY sequence DESC
-                                "logs": recent_logs,
+                            LIMIT 5
-                                "age": age_seconds
+                        ''', (r['run_id'],))
-                            })
+                        audits = [dict(a) for a in audit_cur]
-                    elif session_role == "lead" and not lead_heartbeat_active:
+                        last_audit_at = audits[0]['occurred_at'] if audits else started_at
-                        lead_last_pulse_seconds = age_seconds
+                        age_audit_sec = (now_ms - last_audit_at) / 1000.0 if last_audit_at else 9999
-                        if not is_completed and age_seconds < 60:
+                        age_ended_sec = (now_ms - ended_at) / 1000.0 if ended_at else 9999
-                            lead_heartbeat_active = True
-                            lead_task_desc = session_prompt[:120]
-                            lead_recent_logs = recent_logs
 
-            status["lead_heartbeat"] = {
+                        is_active = (exec_status == 'running') or (age_audit_sec < 60 and exec_status != 'terminal') or (age_ended_sec < 10)
-                "last_pulse_seconds_ago": lead_last_pulse_seconds,
+
-                "project": target_project,
+                        logs = []
-                "is_beating": lead_heartbeat_active
+                        for a in audits:
-            }
+                            tname = a.get('tool_name') or a.get('action') or 'step'
-        except Exception as e:
+                            st = a.get('status') or ''
-            status["lead_heartbeat"] = {"error": str(e), "is_beating": False, "last_pulse_seconds_ago": 999}
+                            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"
@@ -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"] = primary_coder.get('task') or primary_coder.get('label') or 'Coding Session'
-            status["agents"]["coder1"]["currentTask"] = f"Aktiv ({active_coder_count}x): {primary_coder['task'] or 'Coding Session'}"
+            status["agents"]["coder1"]["lastAction"] = f"OpenClaw Subagent Coder ({len(active_working_coders)}x active)"
-            status["agents"]["coder1"]["lastAction"] = f"OpenClaw Coder Turn ({active_coder_count} parallel)"
+            status["agents"]["coder1"]["liveLog"] = primary_coder.get('logs') or ["Implementing feature / bugfix..."]
-            status["agents"]["coder1"]["liveLog"] = primary_coder["logs"] or ["Implementing feature / bugfix..."]
+            status["agents"]["coder1"]["parallelCount"] = len(active_working_coders)
-            status["agents"]["coder1"]["parallelCount"] = active_coder_count
+        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"]["currentTask"] = f"💓 Heartbeat in {target_project}: {lead_task_desc or 'Prüft Backlog & Dispatching'}"
-            status["agents"]["lead"]["lastAction"] = "Lead Active"
+            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)"