apiVersion: v1 data: compact_sessions.py: "#!/usr/bin/env python3\n\"\"\"\nNightly multi-agent session maintenance and compaction script for OpenClaw.\n1. Pre-warms OS page cache for SQLite databases on spinning disks.\n2. Recovers any wedged or locked sessions via recover_sessions.py.\n3. Prunes abandoned isolated cron sessions (>48h old) to prevent token/state bloat.\n4. Tier A Hard-Reset: Resets koch, hebamme, and shopping topics to 0 tokens (clean archive).\n5. Tier B Smart-Compaction: Compacts active sessions exceeding token thresholds (5s pacing).\n6. Checkpoints SQLite WAL journals across all agent databases to disk.\n7. Optimizes SQLite query planner without blocking full VACUUM.\n\"\"\"\nimport glob\nimport json\nimport os\nimport sqlite3\nimport subprocess\nimport sys\nimport time\n\nTHRESHOLD_TOKENS = 15000\nTHRESHOLD_TRANSCRIPT_BYTES = 200000\nTHRESHOLD_EVENT_COUNT = 30\nCRON_MAX_AGE_HOURS = 48\nRPC_TIMEOUT_MS = 180000\n\n# Tracking Channels: Deterministic sliding-window (mila_tracker, nutrition_tracker).\n# Strict sliding window of ~5-8 events (2-3 turns max), 0 LLM tokens, instant truncation.\nTRACKING_TARGETS = [\n (\"mila_tracker\", \"agent:mila_tracker:telegram:group:-1004326921510:topic:2796\"),\n \ (\"nutrition_tracker\", \"agent:nutrition_tracker:telegram:group:-1004326921510:topic:2797\"),\n]\nMAX_TRACKING_EVENTS = 8\nMAX_TRACKING_LINES = 8\n\n# Tier A: High-frequency interaction topics (koch, hebamme, shopping).\n# Deep-compaction keeps a compact summary + latest 2 turns without completely wiping conversational memory.\nTIER_A_TARGETS = [\n (\"koch\", \"agent:koch:telegram:group:-1004326921510:topic:21\"),\n (\"hebamme\", \"agent:hebamme:telegram:group:-1004326921510:topic:22\"),\n \ (\"shopping\", \"agent:shopping:telegram:group:-1004326921510:topic:115\"),\n]\n\ndef prewarm_page_cache():\n print(\"=== Pre-warming SQLite databases and Knowledge Wiki into OS page cache ===\")\n t0 = time.time()\n try:\n subprocess.run(\n \ \"cat /home/node/.openclaw/state/openclaw.sqlite /home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite > /dev/null 2>&1 || true; \"\n \"find /home/node/.openclaw/wiki/main /home/node/.openclaw/workspace* -maxdepth 3 -name '*.md' -exec cat {} + > /dev/null 2>&1 || true\",\n shell=True,\n timeout=90.0\n )\n \ print(f\"Page cache pre-warmed in {time.time() - t0:.2f}s\")\n except Exception as e:\n print(f\"Page cache pre-warming warning: {e}\")\n\ndef prune_stale_cron_sessions():\n print(f\"=== Pruning isolated cron sessions older than {CRON_MAX_AGE_HOURS}h ===\")\n cutoff_ms = int(time.time() * 1000) - (CRON_MAX_AGE_HOURS * 3600 * 1000)\n total_pruned = 0\n total_tokens_reclaimed = 0\n\n for db in sorted(glob.glob(\"/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\")):\n \ agent = db.split(\"/\")[5]\n try:\n conn = sqlite3.connect(db, timeout=30.0)\n conn.execute(\"PRAGMA foreign_keys = ON;\")\n cursor = conn.cursor()\n cursor.execute(\"SELECT name FROM sqlite_master WHERE type=\\\"table\\\"\")\n all_tables = set(r[0] for r in cursor.fetchall())\n\n \ cursor.execute(\"SELECT session_key, current_session_id, updated_at, entry_json FROM session_nodes WHERE session_key LIKE \\\"%:cron:%\\\"\")\n stale = []\n for k, sid, up, ej in cursor.fetchall():\n if up and up < cutoff_ms:\n tok = 0\n if ej:\n \ try:\n tok = json.loads(ej).get(\"totalTokens\", 0) or 0\n except Exception:\n pass\n \ stale.append((k, sid, tok))\n\n if stale:\n tokens_reclaimed = sum(t[2] for t in stale)\n for k, sid, _ in stale:\n for t in [\"trajectory_runtime_events\", \"transcript_events\", \"transcript_event_identities\", \"session_transcript_active_events\", \"transcript_rewrite_watermarks\", \"session_transcript_index_state\", \"session_transcript_fts\", \"session_conversations\"]:\n if t in all_tables and sid:\n cursor.execute(f\"DELETE FROM {t} WHERE session_id = ?\", (sid,))\n for t in [\"session_participants\", \"heartbeat_outcomes\", \"session_progress_cards\", \"board_widgets\", \"board_tabs\"]:\n \ if t in all_tables:\n cursor.execute(f\"DELETE FROM {t} WHERE session_key = ?\", (k,))\n if \"memory_entry_origins\" in all_tables:\n cursor.execute(\"DELETE FROM memory_entry_origins WHERE session_key = ? OR session_id = ?\", (k, sid or \"\"))\n if \"session_windows\" in all_tables:\n cursor.execute(\"DELETE FROM session_windows WHERE session_key = ? OR session_id = ?\", (k, sid or \"\"))\n \ cursor.execute(\"DELETE FROM session_nodes WHERE session_key = ?\", (k,))\n conn.commit()\n total_pruned += len(stale)\n \ total_tokens_reclaimed += tokens_reclaimed\n print(f\" \ [{agent}] Pruned {len(stale)} dead cron sessions ({tokens_reclaimed} tokens reclaimed)\")\n\n # Unconditionally resolve any dangling foreign key references across all tables\n while True:\n cursor.execute(\"PRAGMA foreign_key_check;\")\n violations = cursor.fetchall()\n if not violations:\n break\n for table, rowid, parent, fkid in violations:\n cursor.execute(f\"DELETE FROM {table} WHERE rowid = ?\", (rowid,))\n conn.commit()\n conn.close()\n \ except Exception as e:\n print(f\" [{agent}] Pruning error: {e}\")\n\n # Clear resolved quarantine entries if all databases are healthy\n \ try:\n q_path = \"/home/node/.openclaw/state/openclaw-quarantine.sqlite\"\n \ if os.path.exists(q_path):\n q_conn = sqlite3.connect(q_path)\n \ q_cur = q_conn.cursor()\n q_cur.execute(\"DELETE FROM quarantined_databases\")\n \ q_conn.commit()\n q_conn.close()\n except Exception as e:\n print(f\"Quarantine cleanup error: {e}\")\n\n print(f\"Cron pruning complete: {total_pruned} sessions pruned, {total_tokens_reclaimed} tokens reclaimed.\")\n\ndef compact_tracking_sessions():\n print(\"=== Tracking Channels Sliding-Window Maintenance (mila_tracker, nutrition_tracker) ===\")\n for agent, key in TRACKING_TARGETS:\n db_path = f\"/home/node/.openclaw/agents/{agent}/agent/openclaw-agent.sqlite\"\n \ if not os.path.exists(db_path):\n continue\n try:\n \ con = sqlite3.connect(db_path, timeout=5.0)\n row = con.execute(\"SELECT current_session_id FROM session_nodes WHERE session_key=?\", (key,)).fetchone()\n \ if not row or not row[0]:\n con.close()\n print(f\" \ [{agent}] No active session node found for {key}.\")\n continue\n \ sid = row[0]\n cnt = con.execute(\"SELECT count(*) FROM transcript_events WHERE session_id=?\", (sid,)).fetchone()[0]\n con.close()\n\n \ if cnt > MAX_TRACKING_EVENTS:\n print(f\" [{agent}] Session {key} has {cnt} events (> {MAX_TRACKING_EVENTS}). Truncating to last {MAX_TRACKING_LINES} lines...\")\n cmd = [\n \"openclaw\", \"sessions\", \"compact\", key,\n \"--agent\", agent,\n \"--max-lines\", str(MAX_TRACKING_LINES),\n \"--timeout\", \"30000\"\n ]\n \ r = subprocess.run(cmd, capture_output=True, text=True)\n if r.returncode == 0:\n print(f\" ✓ Truncated successfully: {r.stdout.strip()}\")\n else:\n print(f\" ✗ Truncation failed: {r.stderr.strip()}\")\n else:\n print(f\" \ [{agent}] Session {key} has {cnt} events (<= {MAX_TRACKING_EVENTS}). Clean.\")\n \ except Exception as e:\n print(f\" ✗ [{agent}] Error during tracking compaction: {e}\")\n\ndef compact_tier_a_sessions():\n print(\"=== Tier A: Deep-Compaction (koch, hebamme, shopping) ===\")\n for agent, key in TIER_A_TARGETS:\n db_path = f\"/home/node/.openclaw/agents/{agent}/agent/openclaw-agent.sqlite\"\n \ if not os.path.exists(db_path):\n continue\n try:\n \ con = sqlite3.connect(db_path, timeout=10.0)\n row = con.execute(\"SELECT current_session_id FROM session_nodes WHERE session_key=?\", (key,)).fetchone()\n \ con.close()\n if not row or not row[0]:\n print(f\" \ [Tier A] {agent} ({key}) already at 0 (no active session).\")\n continue\n \ ev_cnt, ev_bytes = get_session_metrics(agent, key, row[0])\n if ev_cnt <= 2 and ev_bytes < 20000:\n print(f\" [Tier A] {agent} ({key}) has only {ev_cnt} events ({ev_bytes} bytes), skipping.\")\n continue\n \ print(f\" [Tier A] Deep-compacting {agent} ({key}) [{ev_cnt} events, {ev_bytes//1024}KB]...\")\n r = subprocess.run(\n [\"openclaw\", \"sessions\", \"compact\", key, \"--agent\", agent, \"--timeout\", str(RPC_TIMEOUT_MS)],\n \ capture_output=True, text=True\n )\n if r.returncode == 0:\n print(f\" ✓ Deep-compacted: {r.stdout.strip()}\")\n \ elif \"Already compacted\" in r.stderr or \"Already compacted\" in r.stdout:\n print(f\" ✓ Already compacted.\")\n else:\n \ print(f\" ✗ Standard compact error: {r.stderr.strip()}; retrying with --max-lines 40...\")\n fb = subprocess.run(\n [\"openclaw\", \"sessions\", \"compact\", key, \"--agent\", agent, \"--max-lines\", \"40\", \"--timeout\", str(RPC_TIMEOUT_MS)],\n capture_output=True, text=True\n )\n \ if fb.returncode == 0:\n print(f\" ✓ Truncation fallback success: {fb.stdout.strip()}\")\n else:\n print(f\" \ ✗ Truncation fallback failed: {fb.stderr.strip()}\")\n except Exception as e:\n print(f\" ✗ Exception deep-compacting {agent}: {e}\")\n \ time.sleep(2)\n\ndef get_session_metrics(agent, key, sid=None):\n db_path = f\"/home/node/.openclaw/agents/{agent}/agent/openclaw-agent.sqlite\"\n if not os.path.exists(db_path):\n return 0, 0\n try:\n con = sqlite3.connect(db_path, timeout=5.0)\n cur = con.cursor()\n if not sid:\n row = cur.execute(\"SELECT current_session_id FROM session_nodes WHERE session_key=?\", (key,)).fetchone()\n sid = row[0] if row else None\n if not sid:\n con.close()\n return 0, 0\n ev_cnt, ev_bytes = cur.execute(\n \"SELECT count(*), coalesce(sum(length(event_json)), 0) FROM transcript_events WHERE session_id=?\", \n (sid,)\n ).fetchone()\n \ con.close()\n return ev_cnt, ev_bytes\n except Exception:\n \ return 0, 0\n\ndef compact_active_sessions():\n excluded_keys = set(t[1] for t in TIER_A_TARGETS) | set(t[1] for t in TRACKING_TARGETS)\n print(f\"=== Tier B: Running Multi-Agent Compaction (threshold: {THRESHOLD_TOKENS} tokens / {THRESHOLD_TRANSCRIPT_BYTES//1024}KB, LLM summarization) ===\")\n res = subprocess.run([\"openclaw\", \"sessions\", \"--all-agents\", \"--json\"], capture_output=True, text=True)\n \ if res.returncode != 0:\n print(f\"Failed to list sessions: {res.stderr}\")\n \ return\n\n sessions = json.loads(res.stdout).get(\"sessions\", [])\n \ compacted = 0\n skipped = 0\n\n for s in sessions:\n key = s.get(\"key\", \"\")\n agent = s.get(\"agentId\") or \"main\"\n tokens = s.get(\"totalTokens\") or 0\n sid = s.get(\"sessionId\")\n\n # Skip Tier A and Tracking agents as they have dedicated routines\n if key in excluded_keys:\n continue\n\n \ # Target main sessions and all telegram sessions (topics and direct)\n \ is_target = (\n key == \"agent:main:main\" or \n \":telegram:\" in key or\n (key.startswith(\"agent:\") and key.endswith(\":main\"))\n \ )\n if is_target and not key.endswith(\":heartbeat\"):\n ev_cnt, ev_bytes = get_session_metrics(agent, key, sid)\n needs_compaction = (\n tokens >= THRESHOLD_TOKENS or\n ev_bytes >= THRESHOLD_TRANSCRIPT_BYTES or\n (tokens == 0 and ev_cnt >= THRESHOLD_EVENT_COUNT)\n \ )\n if needs_compaction:\n stat_str = f\"{tokens} tokens\" if tokens else f\"{ev_bytes//1024}KB / {ev_cnt} events\"\n print(f\"Compacting {agent} ({stat_str}) -> {key}...\")\n c_res = subprocess.run(\n \ [\"openclaw\", \"sessions\", \"compact\", key, \"--agent\", agent, \"--timeout\", str(RPC_TIMEOUT_MS)],\n capture_output=True, text=True\n )\n if c_res.returncode == 0:\n print(f\" \ ✓ {c_res.stdout.strip()}\")\n compacted += 1\n elif \"Already compacted\" in c_res.stderr or \"Already compacted\" in c_res.stdout:\n \ print(f\" ✓ Already compacted.\")\n skipped += 1\n else:\n print(f\" ✗ {c_res.stderr.strip()}; retrying with --max-lines 80 fallback...\")\n fb_res = subprocess.run(\n \ [\"openclaw\", \"sessions\", \"compact\", key, \"--agent\", agent, \"--max-lines\", \"80\", \"--timeout\", str(RPC_TIMEOUT_MS)],\n capture_output=True, text=True\n )\n if fb_res.returncode == 0:\n print(f\" ✓ (fallback) {fb_res.stdout.strip()}\")\n \ compacted += 1\n else:\n print(f\" \ ✗ Fallback failed: {fb_res.stderr.strip()}\")\n time.sleep(5) \ # 5s pause to avoid rotational HDD I/O saturation\n else:\n skipped += 1\n\n print(f\"Compaction complete: {compacted} compacted, {skipped} skipped.\")\n\ndef checkpoint_wal_journals():\n print(\"=== Checkpointing SQLite WAL journals across all agent databases ===\")\n dbs = [\"/home/node/.openclaw/state/openclaw.sqlite\"] + glob.glob(\"/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\")\n wal_freed = 0\n for db in dbs:\n wal = db + \"-wal\"\n if os.path.exists(wal):\n \ sz = os.path.getsize(wal)\n try:\n con = sqlite3.connect(db, timeout=30.0)\n con.execute(\"PRAGMA wal_checkpoint(TRUNCATE);\")\n \ con.close()\n wal_freed += sz\n except Exception as e:\n print(f\" ✗ Error checkpointing {db}: {e}\")\n \ print(f\"WAL Checkpoint complete: {wal_freed / (1024*1024):.2f} MB merged to disk.\")\n\ndef vacuum_and_optimize():\n print(\"=== Running SQLite OPTIMIZE across all databases ===\")\n dbs = [\"/home/node/.openclaw/state/openclaw.sqlite\"] + glob.glob(\"/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\")\n for db in dbs:\n try:\n con = sqlite3.connect(db, timeout=30.0)\n \ con.execute(\"PRAGMA optimize;\")\n con.close()\n except Exception as e:\n print(f\" ✗ Error optimizing {db}: {e}\")\n print(\"SQLite optimization complete.\")\n\ndef main():\n prewarm_page_cache()\n try:\n \ subprocess.run([\"python3\", \"/home/node/.openclaw/scripts/recover_sessions.py\"], check=False)\n except Exception:\n pass\n prune_stale_cron_sessions()\n \ compact_tracking_sessions()\n compact_tier_a_sessions()\n compact_active_sessions()\n \ checkpoint_wal_journals()\n vacuum_and_optimize()\n prewarm_page_cache()\n\nif __name__ == \"__main__\":\n main()\n" dedup_subagent_vectors.py: | #!/usr/bin/env python3 import glob, json, os, sqlite3, sys cfg_path = "/home/node/.openclaw/openclaw.json" if os.path.exists(cfg_path): try: with open(cfg_path, "r") as f: cfg = json.load(f) mem_search = cfg.get("memory", {}).get("search", {}) if mem_search.get("extraPaths"): mem_search["extraPaths"] = [] with open(cfg_path, "w") as f: json.dump(cfg, f, indent=2) print("[dedup_subagents] Cleared top-level memory.search.extraPaths from openclaw.json (scoped to main only)") except Exception as e: print(f"[dedup_subagents] Warning checking openclaw.json: {e}", file=sys.stderr) vec_path = "/app/node_modules/sqlite-vec-linux-x64/vec0.so" for db in sorted(glob.glob("/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite")): agent = os.path.basename(os.path.dirname(os.path.dirname(db))) if agent == "main": continue try: conn = sqlite3.connect(db, timeout=10.0) if os.path.exists(vec_path): conn.enable_load_extension(True) conn.load_extension(vec_path) cur = conn.cursor() chk = cur.execute("SELECT count(*) FROM memory_index_chunks").fetchone()[0] if chk > 0: print(f"[dedup_subagents] Purging {chk} duplicated vector chunks from agent {agent}...") for t in [ "memory_index_sources", "memory_index_chunks", "memory_index_chunk_recall_metadata", "memory_index_chunk_provenance", "memory_entry_origins", "memory_embedding_cache", "memory_index_chunks_vec" ]: try: cur.execute(f"DELETE FROM {t}") except Exception: pass conn.commit() cur.execute("VACUUM") print(f"[dedup_subagents] Reclaimed disk space for agent {agent}") conn.close() except Exception as e: print(f"[dedup_subagents] Notice for {agent}: {e}", file=sys.stderr) gateway-preflight.mjs: | import fs from "node:fs"; import path from "node:path"; import { execSync, spawn } from "node:child_process"; if (!process.env.__OPENCLAW_PREFLIGHT_DONE) { process.env.__OPENCLAW_PREFLIGHT_DONE = "1"; const isGateway = process.argv.some(arg => arg.endsWith("openclaw.mjs") || arg.endsWith("openclaw")) && process.argv.includes("gateway"); if (isGateway) { const t0 = Date.now(); console.log("[gateway-preflight] Initializing container self-healing preflight..."); // 1. Clean up stale gateway lock files from ungraceful kills for (const dir of ["/tmp/openclaw-1000", "/tmp"]) { if (fs.existsSync(dir)) { try { const files = fs.readdirSync(dir); for (const file of files) { if (file.startsWith("gateway") && file.endsWith(".lock")) { const fullPath = path.join(dir, file); fs.unlinkSync(fullPath); console.log(`[gateway-preflight] Removed stale lock: ${fullPath}`); } } } catch (e) { console.warn(`[gateway-preflight] Lock cleanup warning in ${dir}:`, e.message); } } } // 2. Clear false database quarantines const qPath = "/home/node/.openclaw/state/openclaw-quarantine.sqlite"; if (fs.existsSync(qPath)) { try { execSync(`/home/node/.openclaw/.local/bin/python3 -c "import sqlite3; c=sqlite3.connect(\x27${qPath}\x27); c.execute(\x27DELETE FROM quarantined_databases\x27); c.commit(); c.close()"`, { stdio: "ignore" }); console.log("[gateway-preflight] Cleared database quarantine."); } catch (_) {} } // 3. Deduplicate subagent vector databases (keeps non-main agent DBs < 15MB) try { execSync("/home/node/.openclaw/.local/bin/python3 /home/node/.openclaw/scripts/dedup_subagent_vectors.py", { stdio: "inherit" }); } catch (e) { console.warn("[gateway-preflight] Vector deduplication warning:", e.message); } // 4. Pre-warm page cache sequentially (transforms random 70 IOPS disk reads into sequential streaming) try { const pw0 = Date.now(); execSync("cat /home/node/.openclaw/state/openclaw.sqlite /home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite > /dev/null 2>&1 || true; find /home/node/.openclaw/wiki/main /home/node/.openclaw/workspace* -maxdepth 3 -name '*.md' -exec cat {} + > /dev/null 2>&1 || true"); console.log(`[gateway-preflight] Pre-warmed SQLite databases and Wiki markdown in ${Date.now() - pw0}ms`); } catch (_) {} // 5. Recover any wedged running sessions left over from ungraceful container restart try { execSync("/home/node/.openclaw/.local/bin/python3 /home/node/.openclaw/scripts/recover_sessions.py", { stdio: "inherit" }); } catch (e) { console.warn("[gateway-preflight] Session recovery warning:", e.message); } // 6. Spawn background turn replay daemon if interrupted turns were detected try { if (fs.existsSync("/tmp/openclaw-interrupted-turns.json")) { console.log("[gateway-preflight] Spawning background turn replay daemon..."); const child = spawn("/home/node/.openclaw/.local/bin/python3", ["/home/node/.openclaw/scripts/replay_interrupted_turns.py"], { detached: true, stdio: "ignore" }); child.unref(); } } catch (e) { console.warn("[gateway-preflight] Could not spawn turn replay daemon:", e.message); } console.log(`[gateway-preflight] Self-healing completed in ${Date.now() - t0}ms.`); } } openclaw-triage.py: "#!/usr/bin/env python3\n\"\"\"\nopenclaw-triage.py: Rapid zero-dependency incident triage for OpenClaw multi-agent sessions.\nDirectly inspects ground-truth SQLite session state and transcript events in < 1 second.\n\"\"\"\nimport argparse\nimport glob\nimport json\nimport os\nimport sqlite3\nimport sys\nimport time\n\nTOPIC_TO_AGENT = {\n \"20\": \"fitness\",\n \"21\": \"koch\",\n \"22\": \"hebamme\",\n \ \"51\": \"travel\",\n \"115\": \"shopping\",\n \"116\": \"regensburg\",\n \ \"117\": \"recherche\",\n \"118\": \"doktor\",\n \"119\": \"itguy\",\n \ \"164\": \"grabengasse_media\",\n \"2796\": \"mila_tracker\",\n \"2797\": \"nutrition_tracker\",\n}\n\ndef parse_args():\n parser = argparse.ArgumentParser(description=\"Rapid OpenClaw multi-agent session triage.\")\n parser.add_argument(\"--topic\", \"-t\", help=\"Telegram forum topic ID (e.g. 21, 22, 117)\")\n parser.add_argument(\"--agent\", \"-a\", help=\"Agent identifier (e.g. koch, hebamme, recherche)\")\n parser.add_argument(\"--limit\", \"-n\", type=int, default=10, help=\"Number of recent events to display (default: 10)\")\n parser.add_argument(\"--db-root\", default=\"/home/node/.openclaw/agents\", help=\"Root path of agent SQLite databases\")\n return parser.parse_args()\n\ndef triage_agent_session(agent, topic_id=None, limit=10, db_root=\"/home/node/.openclaw/agents\"):\n \ db_path = os.path.join(db_root, agent, \"agent\", \"openclaw-agent.sqlite\")\n \ if not os.path.exists(db_path):\n print(f\"[-] Database not found for agent '{agent}': {db_path}\")\n return False\n\n try:\n con = sqlite3.connect(db_path, timeout=5.0)\n except Exception as e:\n print(f\"[-] Failed to connect to SQLite DB for agent '{agent}': {e}\")\n return False\n\n \ cur = con.cursor()\n # Find matching session\n if topic_id:\n target_pattern = f\"%topic:{topic_id}%\"\n cur.execute(\n \"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\",\n \ (target_pattern,)\n )\n else:\n cur.execute(\n \"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\"\n \ )\n\n row = cur.fetchone()\n if not row:\n print(f\"[-] No session found in {agent} DB (filter: topic={topic_id})\")\n con.close()\n \ return False\n\n skey, sid, status, updated_at_ms, entry_json = row\n \ now_ms = time.time() * 1000\n age_min = (now_ms - updated_at_ms) / 60000.0 if updated_at_ms else 0.0\n\n entry = {}\n if entry_json:\n try:\n \ entry = json.loads(entry_json)\n except Exception:\n pass\n\n \ tokens = entry.get(\"totalTokens\", \"?\")\n model = entry.get(\"model\", \"default\")\n last_run_id = entry.get(\"activeWriterRunId\") or entry.get(\"activeRunId\", None)\n\n # Detect anomalies\n badges = []\n if isinstance(tokens, int):\n \ if tokens > 60000:\n badges.append(\"\U0001F6A8 [CRITICAL_CONTEXT_BLOAT: >60k tokens]\")\n elif tokens > 20000:\n badges.append(\"⚠️ [HIGH_CONTEXT_TOKENS: >20k tokens]\")\n if status == \"running\" and age_min > 5.0:\n badges.append(\"\U0001F512 [POSSIBLY_WEDGED_RUNNING: updated >5m ago]\")\n elif status == \"processing\":\n badges.append(\"⚡ [ACTIVE_PROCESSING]\")\n\n \ # Check ingress queue status\n state_db = \"/home/node/.openclaw/state/openclaw.sqlite\"\n \ if topic_id and os.path.exists(state_db):\n try:\n s_con = sqlite3.connect(state_db, timeout=2.0)\n lane_pat = f\"%topic:{topic_id}%\"\n \ pending_cnt = s_con.execute(\n \"SELECT count(*) FROM channel_ingress_events WHERE lane_key LIKE ? AND status IN ('pending', 'claimed')\",\n \ (lane_pat,)\n ).fetchone()[0]\n s_con.close()\n \ if pending_cnt > 0:\n badges.append(f\"\U0001F4EC [INGRESS_STALLED: {pending_cnt} message(s) pending]\")\n except Exception:\n pass\n\n \ print(\"=\" * 72)\n print(f\"\U0001F3AF AGENT: {agent.upper():<12} | TOPIC: {topic_id or 'N/A':<6} | STATUS: {status.upper()}\")\n print(f\" SESSION KEY: {skey}\")\n print(f\" SESSION ID: {sid}\")\n print(f\" MODEL: \ {model}\")\n print(f\" TOKENS: {tokens} | LAST UPDATED: {age_min:.1f}m ago\")\n if badges:\n print(f\" ANOMALIES: {' '.join(badges)}\")\n \ print(\"=\" * 72)\n\n # Query transcript events\n cur.execute(\n \"SELECT seq, created_at, event_json FROM transcript_events WHERE session_id = ? ORDER BY seq DESC LIMIT ?\",\n (sid, limit)\n )\n events = cur.fetchall()\n \ con.close()\n\n if not events:\n print(\" (No transcript events found)\")\n return True\n\n events.reverse()\n print(\"\\n\U0001F4DC RECENT TRANSCRIPT EVENTS (Chronological):\")\n\n last_role = None\n for seq, ca_ms, ej_str in events:\n try:\n ev = json.loads(ej_str)\n \ except Exception:\n print(f\"[{seq}] \")\n \ continue\n\n etype = ev.get(\"type\")\n if etype != \"message\":\n continue\n\n msg = ev.get(\"message\", {})\n \ role = msg.get(\"role\")\n last_role = role\n content = msg.get(\"content\", [])\n\n if role == \"user\":\n txt = content if isinstance(content, str) else json.dumps(content)\n # Check for audio transcript\n if isinstance(content, str) and \"[Audio]\" in content:\n \ print(f\"\\n\U0001F449 [SEQ {seq}] \U0001F3A4 USER AUDIO PROMPT:\")\n \ for line in content.splitlines()[:6]:\n print(f\" \ {line}\")\n if len(content.splitlines()) > 6:\n print(\" \ ...\")\n else:\n txt_preview = txt.replace(\"\\n\", \" \")[:240]\n print(f\"\\n\U0001F449 [SEQ {seq}] \U0001F464 USER: {txt_preview}\")\n\n elif role == \"assistant\":\n print(f\"\\n\U0001F916 [SEQ {seq}] ASSISTANT TURN:\")\n if isinstance(content, list):\n for item in content:\n if not isinstance(item, dict):\n continue\n \ itype = item.get(\"type\")\n if itype == \"thinking\":\n th = item.get(\"thinking\", \"\").replace(\"\\n\", \" \")\n print(f\" \U0001F9E0 THINKING: {th[:280]}...\")\n \ elif itype == \"toolCall\":\n tname = item.get(\"name\", \"unknown\")\n targs = json.dumps(item.get(\"arguments\", {}))\n print(f\" ⚙️ TOOL CALL: {tname}({targs[:160]})\")\n \ elif itype == \"text\":\n tx = item.get(\"text\", \"\")\n lines = tx.splitlines()\n prev = lines[0][:150] if lines else \"\"\n print(f\" \U0001F4AC TEXT OUTPUT: {prev}...\")\n elif isinstance(content, str):\n print(f\" \ \U0001F4AC TEXT OUTPUT: {content[:200]}...\")\n\n elif role == \"toolResult\":\n \ is_err = ev.get(\"isError\", False)\n tname = ev.get(\"details\", {}).get(\"tool\") or msg.get(\"toolName\", \"tool\")\n res_txt = json.dumps(content) if not isinstance(content, str) else content\n if is_err or \"error\" in res_txt.lower() or \"must have required property\" in res_txt:\n print(f\" \ ❌ [SEQ {seq}] TOOL ERROR ({tname}): {res_txt[:260]}\")\n else:\n \ print(f\" ✅ [SEQ {seq}] TOOL RESULT ({tname}): {res_txt[:180]}\")\n\n \ if last_role == \"user\":\n print(\"\\n⚠️ WARNING: The last event in the transcript is an UNANSWERED USER MESSAGE!\")\n print(\" The agent turn either crashed, timed out, or was killed before producing a response.\")\n\n \ print(\"\\n\" + \"=\" * 72)\n return True\n\ndef main():\n args = parse_args()\n \ if not args.topic and not args.agent:\n print(\"Usage: openclaw-triage.py (--topic | --agent ) [--limit ]\")\n sys.exit(1)\n\n agent = args.agent\n topic = args.topic\n\n if topic and not agent:\n agent = TOPIC_TO_AGENT.get(str(topic))\n if not agent:\n # Search all DBs\n for a, t in TOPIC_TO_AGENT.items():\n if t == topic:\n agent = a\n break\n\n if not agent:\n # Scan all DBs for the topic\n for db in sorted(glob.glob(os.path.join(args.db_root, \"*/agent/openclaw-agent.sqlite\"))):\n cand_agent = db.split(os.sep)[-3]\n \ try:\n con = sqlite3.connect(db, timeout=2.0)\n cur = con.cursor()\n cur.execute(\n \"SELECT session_key FROM session_nodes WHERE session_key LIKE ? LIMIT 1\",\n (f\"%topic:{topic}%\",)\n \ )\n if cur.fetchone():\n agent = cand_agent\n con.close()\n break\n con.close()\n \ except Exception:\n pass\n\n if not agent:\n print(f\"[-] Could not map topic '{topic}' to any known OpenClaw agent.\")\n sys.exit(1)\n\n \ success = triage_agent_session(agent, topic_id=topic, limit=args.limit, db_root=args.db_root)\n \ sys.exit(0 if success else 1)\n\nif __name__ == \"__main__\":\n main()\n" recover_sessions.py: |+ #!/usr/bin/env python3 import glob, sqlite3, json, os, sys, time t0 = time.time() recovered = 0 interrupted_turns = [] def extract_prompt(msg): if not msg: return "" content = msg.get("content") if isinstance(content, str): return content if isinstance(content, list): parts = [] for c in content: if isinstance(c, dict): parts.append(c.get("text") or c.get("content") or "") elif isinstance(c, str): parts.append(c) return "\n".join(filter(None, parts)) return str(content or "") for db in sorted(glob.glob("/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite")): try: conn = sqlite3.connect(db, timeout=10.0) cursor = conn.cursor() while True: cursor.execute("PRAGMA foreign_key_check;") violations = cursor.fetchall() if not violations: break for table, rowid, parent, fkid in violations: cursor.execute(f"DELETE FROM {table} WHERE rowid = ?", (rowid,)) cursor.execute("SELECT session_key, entry_json, current_session_id, status FROM session_nodes WHERE status IN (\"running\", \"failed\", \"error\")") rows = cursor.fetchall() agent_id = os.path.basename(os.path.dirname(os.path.dirname(db))) for sk, ej, csid, st in rows: try: p = json.loads(ej) if ej else {} sid = csid or p.get("sessionId") if sid: ev = cursor.execute("SELECT seq, event_json FROM transcript_events WHERE session_id = ? ORDER BY seq DESC LIMIT 1", (sid,)).fetchone() if ev: ev_seq, ev_raw = ev try: ev_data = json.loads(ev_raw) msg = ev_data.get("message") or ev_data if st == "running" and msg.get("role") == "user": prompt_text = extract_prompt(msg) ctx = p.get("restartRecoveryDeliveryContext") or p.get("delivery", {}).get("context") or p.get("delivery", {}).get("route", {}).get("context") or {} interrupted_turns.append({ "agent": agent_id, "session_key": sk, "session_id": sid, "context": ctx, "prompt": prompt_text }) cursor.execute("PRAGMA foreign_keys = ON") cursor.execute("DELETE FROM transcript_events WHERE session_id = ? AND seq = ?", (sid, ev_seq)) elif msg.get("role") == "assistant" and msg.get("stopReason") == "error": cursor.execute("PRAGMA foreign_keys = ON") cursor.execute("DELETE FROM transcript_events WHERE session_id = ? AND seq = ?", (sid, ev_seq)) except Exception as ex: print(f"[recover_sessions] Error parsing event: {ex}", file=sys.stderr) p["status"] = "done" p["activeWriterRunId"] = None p["activeClaimRunId"] = None p["restartRecoveryDeliveryRunId"] = None cursor.execute("UPDATE session_nodes SET status = \"done\", entry_json = ? WHERE session_key = ?", (json.dumps(p), sk)) cursor.execute("UPDATE session_nodes SET entry_valid = 1 WHERE session_key = ?", (sk,)) recovered += 1 except Exception: pass cursor.execute("UPDATE session_nodes SET entry_valid = 1 WHERE entry_valid != 1") cursor.execute("UPDATE session_windows SET status = \"done\" WHERE status IN (\"running\", \"failed\", \"error\")") conn.commit() conn.close() except Exception as e: print(f"[recover_sessions] Warning for {db}: {e}", file=sys.stderr) state_db = "/home/node/.openclaw/state/openclaw.sqlite" if os.path.exists(state_db): try: sconn = sqlite3.connect(state_db, timeout=10.0) scur = sconn.cursor() scur.execute("UPDATE channel_ingress_events SET attempts = 0, last_attempt_at = NULL, last_error = NULL WHERE status = 'pending' AND attempts > 0") sconn.commit() sconn.close() except Exception as e: print(f"[recover_sessions] Warning resetting ingress backoff: {e}", file=sys.stderr) if interrupted_turns: try: with open("/tmp/openclaw-interrupted-turns.json", "w") as f: json.dump(interrupted_turns, f, indent=2) print(f"[recover_sessions] Detected {len(interrupted_turns)} interrupted user turn(s) saved to /tmp/openclaw-interrupted-turns.json") except Exception as e: print(f"[recover_sessions] Error writing interrupted turns: {e}", file=sys.stderr) print(f"[recover_sessions] Completed in {time.time() - t0:.2f}s ({recovered} wedged sessions recovered)") replay_interrupted_turns.py: | #!/usr/bin/env python3 import json, os, subprocess, sys, time, urllib.request INTERRUPTED_FILE = "/tmp/openclaw-interrupted-turns.json" if not os.path.exists(INTERRUPTED_FILE): sys.exit(0) try: with open(INTERRUPTED_FILE, "r") as f: turns = json.load(f) except Exception as e: print(f"[replay_interrupted_turns] Failed to read {INTERRUPTED_FILE}: {e}", file=sys.stderr) sys.exit(0) if not turns: sys.exit(0) print(f"[replay_interrupted_turns] Found {len(turns)} interrupted turn(s). Waiting for gateway /healthz...") gateway_ready = False for _ in range(60): try: with urllib.request.urlopen("http://127.0.0.1:18790/healthz", timeout=3) as resp: if resp.status == 200: gateway_ready = True break except Exception: pass time.sleep(2) if not gateway_ready: print("[replay_interrupted_turns] Gateway did not become healthy in 120s. Aborting.", file=sys.stderr) sys.exit(1) # Allow 5s for Telegram ingress to bind time.sleep(5) for turn in turns: sk = turn.get("session_key") sid = turn.get("session_id") agent_id = turn.get("agent") ctx = turn.get("context") or {} prompt = turn.get("prompt") or "" channel = ctx.get("channel") or "telegram" to_target = ctx.get("to") or "" thread_id = ctx.get("threadId") target = to_target if target.startswith("telegram:"): parts = target.split(":") target = parts[1] if len(parts) > 1 else target print(f"[replay_interrupted_turns] Processing agent '{agent_id}' on session '{sk}' target='{target}' thread='{thread_id}'") # 1. Send status notification to Telegram if target exists if target: notice = "⚠️ OpenClaw wurde während der Bearbeitung neu gestartet – deine Anfrage wird jetzt nachgeholt..." notify_cmd = [ "openclaw", "message", "send", "--channel", channel, "--target", str(target), "-m", notice ] if thread_id is not None: notify_cmd.extend(["--thread-id", str(thread_id)]) try: subprocess.run(notify_cmd, check=False, timeout=30) print(f"[replay_interrupted_turns] Sent recovery notice to {target}") except Exception as e: print(f"[replay_interrupted_turns] Notice failed: {e}", file=sys.stderr) # 2. Replay agent turn if sk and prompt: p_short = (prompt[:60] + "...") if len(prompt) > 60 else prompt print(f"[replay_interrupted_turns] Replaying agent turn: {p_short}") agent_cmd = [ "openclaw", "agent", "--session-key", sk, "-m", prompt, "--deliver" ] try: res = subprocess.run(agent_cmd, capture_output=True, text=True, timeout=300) if res.returncode == 0: print(f"[replay_interrupted_turns] Turn replay succeeded for {sk}") else: print(f"[replay_interrupted_turns] Turn replay exited with code {res.returncode}: {res.stderr[:200]}", file=sys.stderr) except Exception as e: print(f"[replay_interrupted_turns] Replay execution error: {e}", file=sys.stderr) try: os.remove(INTERRUPTED_FILE) print(f"[replay_interrupted_turns] Cleaned up {INTERRUPTED_FILE}") except Exception: pass kind: ConfigMap metadata: name: openclaw-scripts namespace: openclaw