250 lines
9.1 KiB
Python
250 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
openclaw-triage.py: Rapid zero-dependency incident triage for OpenClaw multi-agent sessions.
|
|
Directly inspects ground-truth SQLite session state and transcript events in < 1 second.
|
|
"""
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
|
|
TOPIC_TO_AGENT = {
|
|
"20": "fitness",
|
|
"21": "koch",
|
|
"22": "hebamme",
|
|
"51": "travel",
|
|
"115": "shopping",
|
|
"116": "regensburg",
|
|
"117": "recherche",
|
|
"118": "doktor",
|
|
"119": "itguy",
|
|
"164": "grabengasse_media",
|
|
"2796": "mila_tracker",
|
|
"2797": "nutrition_tracker",
|
|
}
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="Rapid OpenClaw multi-agent session triage.")
|
|
parser.add_argument("--topic", "-t", help="Telegram forum topic ID (e.g. 21, 22, 117)")
|
|
parser.add_argument("--agent", "-a", help="Agent identifier (e.g. koch, hebamme, recherche)")
|
|
parser.add_argument("--limit", "-n", type=int, default=10, help="Number of recent events to display (default: 10)")
|
|
parser.add_argument("--db-root", default="/home/node/.openclaw/agents", help="Root path of agent SQLite databases")
|
|
return parser.parse_args()
|
|
|
|
def triage_agent_session(agent, topic_id=None, limit=10, db_root="/home/node/.openclaw/agents"):
|
|
db_path = os.path.join(db_root, agent, "agent", "openclaw-agent.sqlite")
|
|
if not os.path.exists(db_path):
|
|
print(f"[-] Database not found for agent '{agent}': {db_path}")
|
|
return False
|
|
|
|
try:
|
|
con = sqlite3.connect(db_path, timeout=5.0)
|
|
except Exception as e:
|
|
print(f"[-] Failed to connect to SQLite DB for agent '{agent}': {e}")
|
|
return False
|
|
|
|
cur = con.cursor()
|
|
# Find matching session
|
|
if topic_id:
|
|
target_pattern = f"%topic:{topic_id}%"
|
|
cur.execute(
|
|
"SELECT session_key, current_session_id, status, updated_at, entry_json FROM session_nodes WHERE session_key LIKE ? AND session_key NOT LIKE '%:heartbeat' ORDER BY updated_at DESC LIMIT 1",
|
|
(target_pattern,)
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"SELECT session_key, current_session_id, status, updated_at, entry_json FROM session_nodes WHERE session_key NOT LIKE '%:heartbeat' ORDER BY updated_at DESC LIMIT 1"
|
|
)
|
|
|
|
row = cur.fetchone()
|
|
if not row:
|
|
print(f"[-] No session found in {agent} DB (filter: topic={topic_id})")
|
|
con.close()
|
|
return False
|
|
|
|
skey, sid, status, updated_at_ms, entry_json = row
|
|
now_ms = time.time() * 1000
|
|
age_min = (now_ms - updated_at_ms) / 60000.0 if updated_at_ms else 0.0
|
|
|
|
entry = {}
|
|
if entry_json:
|
|
try:
|
|
entry = json.loads(entry_json)
|
|
except Exception:
|
|
pass
|
|
|
|
tokens = entry.get("totalTokens", "?")
|
|
model = entry.get("model", "default")
|
|
last_run_id = entry.get("activeWriterRunId") or entry.get("activeRunId", None)
|
|
|
|
# Detect anomalies
|
|
badges = []
|
|
if isinstance(tokens, int):
|
|
if tokens > 60000:
|
|
badges.append("🚨 [CRITICAL_CONTEXT_BLOAT: >60k tokens]")
|
|
elif tokens > 20000:
|
|
badges.append("⚠️ [HIGH_CONTEXT_TOKENS: >20k tokens]")
|
|
if status == "running" and age_min > 5.0:
|
|
badges.append("🔒 [POSSIBLY_WEDGED_RUNNING: updated >5m ago]")
|
|
elif status == "processing":
|
|
badges.append("⚡ [ACTIVE_PROCESSING]")
|
|
|
|
# Check ingress queue status
|
|
state_db = "/home/node/.openclaw/state/openclaw.sqlite"
|
|
if topic_id and os.path.exists(state_db):
|
|
try:
|
|
s_con = sqlite3.connect(state_db, timeout=2.0)
|
|
lane_pat = f"%topic:{topic_id}%"
|
|
pending_cnt = s_con.execute(
|
|
"SELECT count(*) FROM channel_ingress_events WHERE lane_key LIKE ? AND status IN ('pending', 'claimed')",
|
|
(lane_pat,)
|
|
).fetchone()[0]
|
|
s_con.close()
|
|
if pending_cnt > 0:
|
|
badges.append(f"📬 [INGRESS_STALLED: {pending_cnt} message(s) pending]")
|
|
except Exception:
|
|
pass
|
|
|
|
print("=" * 72)
|
|
print(f"🎯 AGENT: {agent.upper():<12} | TOPIC: {topic_id or 'N/A':<6} | STATUS: {status.upper()}")
|
|
print(f" SESSION KEY: {skey}")
|
|
print(f" SESSION ID: {sid}")
|
|
print(f" MODEL: {model}")
|
|
print(f" TOKENS: {tokens} | LAST UPDATED: {age_min:.1f}m ago")
|
|
if badges:
|
|
print(f" ANOMALIES: {' '.join(badges)}")
|
|
print("=" * 72)
|
|
|
|
# Query transcript events
|
|
cur.execute(
|
|
"SELECT seq, created_at, event_json FROM transcript_events WHERE session_id = ? ORDER BY seq DESC LIMIT ?",
|
|
(sid, limit)
|
|
)
|
|
events = cur.fetchall()
|
|
con.close()
|
|
|
|
if not events:
|
|
print(" (No transcript events found)")
|
|
return True
|
|
|
|
events.reverse()
|
|
print("\n📜 RECENT TRANSCRIPT EVENTS (Chronological):")
|
|
|
|
last_role = None
|
|
for seq, ca_ms, ej_str in events:
|
|
try:
|
|
ev = json.loads(ej_str)
|
|
except Exception:
|
|
print(f"[{seq}] <unparseable event json>")
|
|
continue
|
|
|
|
etype = ev.get("type")
|
|
if etype != "message":
|
|
continue
|
|
|
|
msg = ev.get("message", {})
|
|
role = msg.get("role")
|
|
last_role = role
|
|
content = msg.get("content", [])
|
|
|
|
if role == "user":
|
|
txt = content if isinstance(content, str) else json.dumps(content)
|
|
# Check for audio transcript
|
|
if isinstance(content, str) and "[Audio]" in content:
|
|
print(f"\n👉 [SEQ {seq}] 🎤 USER AUDIO PROMPT:")
|
|
for line in content.splitlines()[:6]:
|
|
print(f" {line}")
|
|
if len(content.splitlines()) > 6:
|
|
print(" ...")
|
|
else:
|
|
txt_preview = txt.replace("\n", " ")[:240]
|
|
print(f"\n👉 [SEQ {seq}] 👤 USER: {txt_preview}")
|
|
|
|
elif role == "assistant":
|
|
print(f"\n🤖 [SEQ {seq}] ASSISTANT TURN:")
|
|
if isinstance(content, list):
|
|
for item in content:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
itype = item.get("type")
|
|
if itype == "thinking":
|
|
th = item.get("thinking", "").replace("\n", " ")
|
|
print(f" 🧠 THINKING: {th[:280]}...")
|
|
elif itype == "toolCall":
|
|
tname = item.get("name", "unknown")
|
|
targs = json.dumps(item.get("arguments", {}))
|
|
print(f" ⚙️ TOOL CALL: {tname}({targs[:160]})")
|
|
elif itype == "text":
|
|
tx = item.get("text", "")
|
|
lines = tx.splitlines()
|
|
prev = lines[0][:150] if lines else ""
|
|
print(f" 💬 TEXT OUTPUT: {prev}...")
|
|
elif isinstance(content, str):
|
|
print(f" 💬 TEXT OUTPUT: {content[:200]}...")
|
|
|
|
elif role == "toolResult":
|
|
is_err = ev.get("isError", False)
|
|
tname = ev.get("details", {}).get("tool") or msg.get("toolName", "tool")
|
|
res_txt = json.dumps(content) if not isinstance(content, str) else content
|
|
if is_err or "error" in res_txt.lower() or "must have required property" in res_txt:
|
|
print(f" ❌ [SEQ {seq}] TOOL ERROR ({tname}): {res_txt[:260]}")
|
|
else:
|
|
print(f" ✅ [SEQ {seq}] TOOL RESULT ({tname}): {res_txt[:180]}")
|
|
|
|
if last_role == "user":
|
|
print("\n⚠️ WARNING: The last event in the transcript is an UNANSWERED USER MESSAGE!")
|
|
print(" The agent turn either crashed, timed out, or was killed before producing a response.")
|
|
|
|
print("\n" + "=" * 72)
|
|
return True
|
|
|
|
def main():
|
|
args = parse_args()
|
|
if not args.topic and not args.agent:
|
|
print("Usage: openclaw-triage.py (--topic <ID> | --agent <NAME>) [--limit <N>]")
|
|
sys.exit(1)
|
|
|
|
agent = args.agent
|
|
topic = args.topic
|
|
|
|
if topic and not agent:
|
|
agent = TOPIC_TO_AGENT.get(str(topic))
|
|
if not agent:
|
|
# Search all DBs
|
|
for a, t in TOPIC_TO_AGENT.items():
|
|
if t == topic:
|
|
agent = a
|
|
break
|
|
|
|
if not agent:
|
|
# Scan all DBs for the topic
|
|
for db in sorted(glob.glob(os.path.join(args.db_root, "*/agent/openclaw-agent.sqlite"))):
|
|
cand_agent = db.split(os.sep)[-3]
|
|
try:
|
|
con = sqlite3.connect(db, timeout=2.0)
|
|
cur = con.cursor()
|
|
cur.execute(
|
|
"SELECT session_key FROM session_nodes WHERE session_key LIKE ? LIMIT 1",
|
|
(f"%topic:{topic}%",)
|
|
)
|
|
if cur.fetchone():
|
|
agent = cand_agent
|
|
con.close()
|
|
break
|
|
con.close()
|
|
except Exception:
|
|
pass
|
|
|
|
if not agent:
|
|
print(f"[-] Could not map topic '{topic}' to any known OpenClaw agent.")
|
|
sys.exit(1)
|
|
|
|
success = triage_agent_session(agent, topic_id=topic, limit=args.limit, db_root=args.db_root)
|
|
sys.exit(0 if success else 1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|