#!/usr/bin/env python3 """ Nightly multi-agent session maintenance and compaction script for OpenClaw. 1. Pre-warms OS page cache for SQLite databases on spinning disks. 2. Recovers any wedged or locked sessions via recover_sessions.py. 3. Prunes abandoned isolated cron sessions (>48h old) to prevent token/state bloat. 4. Tier A Hard-Reset: Resets koch, hebamme, and shopping topics to 0 tokens (clean archive). 5. Tier B Smart-Compaction: Compacts active sessions exceeding token thresholds (5s pacing). 6. Checkpoints SQLite WAL journals across all agent databases to disk. 7. Optimizes SQLite query planner without blocking full VACUUM. """ import glob import json import os import sqlite3 import subprocess import sys import time THRESHOLD_TOKENS = 20000 THRESHOLD_TRANSCRIPT_BYTES = 300000 THRESHOLD_EVENT_COUNT = 40 CRON_MAX_AGE_HOURS = 48 RPC_TIMEOUT_MS = 180000 # Tier A: Event-logging agents whose state is 100% externalized (DuckDB / JSONL / Groceries DB). # Nightly reset to 0 prevents cross-day hallucination and eliminates LLM compaction token costs. TIER_A_RESET_TARGETS = [ ("koch", "agent:koch:telegram:group:-1004326921510:topic:21"), ("hebamme", "agent:hebamme:telegram:group:-1004326921510:topic:22"), ("shopping", "agent:shopping:telegram:group:-1004326921510:topic:115"), ] def prewarm_page_cache(): print("=== Pre-warming SQLite databases into OS page cache ===") t0 = time.time() try: subprocess.run( "cat /home/node/.openclaw/state/openclaw.sqlite /home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite > /dev/null 2>&1 || true", shell=True, timeout=90.0 ) print(f"Page cache pre-warmed in {time.time() - t0:.2f}s") except Exception as e: print(f"Page cache pre-warming warning: {e}") def prune_stale_cron_sessions(): print(f"=== Pruning isolated cron sessions older than {CRON_MAX_AGE_HOURS}h ===") cutoff_ms = int(time.time() * 1000) - (CRON_MAX_AGE_HOURS * 3600 * 1000) total_pruned = 0 total_tokens_reclaimed = 0 for db in sorted(glob.glob("/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite")): agent = db.split("/")[5] try: conn = sqlite3.connect(db, timeout=30.0) conn.execute("PRAGMA foreign_keys = ON;") cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type=\"table\"") all_tables = set(r[0] for r in cursor.fetchall()) cursor.execute("SELECT session_key, current_session_id, updated_at, entry_json FROM session_nodes WHERE session_key LIKE \"%:cron:%\"") stale = [] for k, sid, up, ej in cursor.fetchall(): if up and up < cutoff_ms: tok = 0 if ej: try: tok = json.loads(ej).get("totalTokens", 0) or 0 except Exception: pass stale.append((k, sid, tok)) if stale: tokens_reclaimed = sum(t[2] for t in stale) for k, sid, _ in stale: 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"]: if t in all_tables and sid: cursor.execute(f"DELETE FROM {t} WHERE session_id = ?", (sid,)) for t in ["session_participants", "heartbeat_outcomes", "session_progress_cards", "board_widgets", "board_tabs"]: if t in all_tables: cursor.execute(f"DELETE FROM {t} WHERE session_key = ?", (k,)) if "memory_entry_origins" in all_tables: cursor.execute("DELETE FROM memory_entry_origins WHERE session_key = ? OR session_id = ?", (k, sid or "")) if "session_windows" in all_tables: cursor.execute("DELETE FROM session_windows WHERE session_key = ? OR session_id = ?", (k, sid or "")) cursor.execute("DELETE FROM session_nodes WHERE session_key = ?", (k,)) conn.commit() total_pruned += len(stale) total_tokens_reclaimed += tokens_reclaimed print(f" [{agent}] Pruned {len(stale)} dead cron sessions ({tokens_reclaimed} tokens reclaimed)") # Unconditionally resolve any dangling foreign key references across all tables 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,)) conn.commit() conn.close() except Exception as e: print(f" [{agent}] Pruning error: {e}") # Clear resolved quarantine entries if all databases are healthy try: q_path = "/home/node/.openclaw/state/openclaw-quarantine.sqlite" if os.path.exists(q_path): q_conn = sqlite3.connect(q_path) q_cur = q_conn.cursor() q_cur.execute("DELETE FROM quarantined_databases") q_conn.commit() q_conn.close() except Exception as e: print(f"Quarantine cleanup error: {e}") print(f"Cron pruning complete: {total_pruned} sessions pruned, {total_tokens_reclaimed} tokens reclaimed.") def reset_tier_a_sessions(): print("=== Tier A: Daily Hard-Reset to 0 (koch, hebamme, shopping) ===") for agent, key in TIER_A_RESET_TARGETS: db_path = f"/home/node/.openclaw/agents/{agent}/agent/openclaw-agent.sqlite" if not os.path.exists(db_path): continue try: con = sqlite3.connect(db_path, timeout=10.0) row = con.execute("SELECT current_session_id FROM session_nodes WHERE session_key=?", (key,)).fetchone() con.close() if not row or not row[0]: print(f" [Tier A] {agent} ({key}) already at 0 (no active session).") continue print(f" [Tier A] Resetting {agent} ({key})...") r = subprocess.run( ["openclaw", "sessions", "delete", key, "--agent", agent, "--yes"], capture_output=True, text=True, timeout=60.0 ) if r.returncode == 0: print(f" ✓ Reset success: {r.stdout.strip()}") else: print(f" ✗ Reset error: {r.stderr.strip()}") except Exception as e: print(f" ✗ Exception resetting {agent}: {e}") time.sleep(1) def get_session_metrics(agent, key, sid=None): db_path = f"/home/node/.openclaw/agents/{agent}/agent/openclaw-agent.sqlite" if not os.path.exists(db_path): return 0, 0 try: con = sqlite3.connect(db_path, timeout=5.0) cur = con.cursor() if not sid: row = cur.execute("SELECT current_session_id FROM session_nodes WHERE session_key=?", (key,)).fetchone() sid = row[0] if row else None if not sid: con.close() return 0, 0 ev_cnt, ev_bytes = cur.execute( "SELECT count(*), coalesce(sum(length(event_json)), 0) FROM transcript_events WHERE session_id=?", (sid,) ).fetchone() con.close() return ev_cnt, ev_bytes except Exception: return 0, 0 def compact_active_sessions(): tier_a_keys = set(t[1] for t in TIER_A_RESET_TARGETS) print(f"=== Tier B: Running Multi-Agent Compaction (threshold: {THRESHOLD_TOKENS} tokens / {THRESHOLD_TRANSCRIPT_BYTES//1024}KB, LLM summarization) ===") res = subprocess.run(["openclaw", "sessions", "--all-agents", "--json"], capture_output=True, text=True) if res.returncode != 0: print(f"Failed to list sessions: {res.stderr}") return sessions = json.loads(res.stdout).get("sessions", []) compacted = 0 skipped = 0 for s in sessions: key = s.get("key", "") agent = s.get("agentId") or "main" tokens = s.get("totalTokens") or 0 sid = s.get("sessionId") # Skip Tier A agents as they are handled via hard-reset if key in tier_a_keys: continue # Target main sessions and all telegram sessions (topics and direct) is_target = ( key == "agent:main:main" or ":telegram:" in key or (key.startswith("agent:") and key.endswith(":main")) ) if is_target and not key.endswith(":heartbeat"): ev_cnt, ev_bytes = get_session_metrics(agent, key, sid) needs_compaction = ( tokens >= THRESHOLD_TOKENS or ev_bytes >= THRESHOLD_TRANSCRIPT_BYTES or (tokens == 0 and ev_cnt >= THRESHOLD_EVENT_COUNT) ) if needs_compaction: stat_str = f"{tokens} tokens" if tokens else f"{ev_bytes//1024}KB / {ev_cnt} events" print(f"Compacting {agent} ({stat_str}) -> {key}...") c_res = subprocess.run( ["openclaw", "sessions", "compact", key, "--agent", agent, "--timeout", str(RPC_TIMEOUT_MS)], capture_output=True, text=True ) if c_res.returncode == 0: print(f" ✓ {c_res.stdout.strip()}") compacted += 1 elif "Already compacted" in c_res.stderr or "Already compacted" in c_res.stdout: print(f" ✓ Already compacted.") skipped += 1 else: print(f" ✗ {c_res.stderr.strip()}; retrying with --max-lines 80 fallback...") fb_res = subprocess.run( ["openclaw", "sessions", "compact", key, "--agent", agent, "--max-lines", "80", "--timeout", str(RPC_TIMEOUT_MS)], capture_output=True, text=True ) if fb_res.returncode == 0: print(f" ✓ (fallback) {fb_res.stdout.strip()}") compacted += 1 else: print(f" ✗ Fallback failed: {fb_res.stderr.strip()}") time.sleep(5) # 5s pause to avoid rotational HDD I/O saturation else: skipped += 1 print(f"Compaction complete: {compacted} compacted, {skipped} skipped.") def checkpoint_wal_journals(): print("=== Checkpointing SQLite WAL journals across all agent databases ===") dbs = ["/home/node/.openclaw/state/openclaw.sqlite"] + glob.glob("/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite") wal_freed = 0 for db in dbs: wal = db + "-wal" if os.path.exists(wal): sz = os.path.getsize(wal) try: con = sqlite3.connect(db, timeout=30.0) con.execute("PRAGMA wal_checkpoint(TRUNCATE);") con.close() wal_freed += sz except Exception as e: print(f" ✗ Error checkpointing {db}: {e}") print(f"WAL Checkpoint complete: {wal_freed / (1024*1024):.2f} MB merged to disk.") def vacuum_and_optimize(): print("=== Running SQLite OPTIMIZE across all databases ===") dbs = ["/home/node/.openclaw/state/openclaw.sqlite"] + glob.glob("/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite") for db in dbs: try: con = sqlite3.connect(db, timeout=30.0) con.execute("PRAGMA optimize;") con.close() except Exception as e: print(f" ✗ Error optimizing {db}: {e}") print("SQLite optimization complete.") def main(): prewarm_page_cache() try: subprocess.run(["python3", "/home/node/.openclaw/scripts/recover_sessions.py"], check=False) except Exception: pass prune_stale_cron_sessions() reset_tier_a_sessions() compact_active_sessions() checkpoint_wal_journals() vacuum_and_optimize() if __name__ == "__main__": main()