335 lines
15 KiB
Bash
Executable File
335 lines
15 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
# ==============================================================================
|
|
# setup-compaction-cron.sh
|
|
# Declaratively registers the Multi-Agent Nightly Session Compaction in OpenClaw.
|
|
# Schedule: Every night at 04:45 Europe/Berlin (02:45 UTC) (cron: 45 4 * * *)
|
|
#
|
|
# Architecture:
|
|
# - Tier A: Daily Hard-Reset to 0 for koch (Topic 21), hebamme (Topic 22), shopping (Topic 115)
|
|
# (100% externalized state in DuckDB/JSONL; resets prevent cross-day hallucination)
|
|
# - Tier B: Smart Compaction for long-running project agents (itguy, recherche, regensburg, fitness, doktor)
|
|
# - WAL checkpoints and SQLite optimization without blocking full VACUUM
|
|
# ==============================================================================
|
|
|
|
set -euo pipefail
|
|
|
|
NAMESPACE="${NAMESPACE:-openclaw}"
|
|
POD_NAME="${POD_NAME:-openclaw-0}"
|
|
CONTAINER_NAME="${CONTAINER_NAME:-openclaw}"
|
|
|
|
JOB_NAME="nightly-session-compaction"
|
|
CRON_SCHEDULE="45 4 * * *"
|
|
TIMEZONE="Europe/Berlin"
|
|
|
|
echo "=== Registering Multi-Agent Nightly Session Maintenance in OpenClaw ($POD_NAME) ==="
|
|
|
|
# 1. Install or ensure python maintenance script is present on PVC
|
|
echo "Ensuring /home/node/.openclaw/scripts/compact_sessions.py exists..."
|
|
kubectl exec -i -n "$NAMESPACE" "$POD_NAME" -c "$CONTAINER_NAME" -- /bin/sh -c 'cat << "EOF" > /home/node/.openclaw/scripts/compact_sessions.py
|
|
#!/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()
|
|
EOF
|
|
chmod +x /home/node/.openclaw/scripts/compact_sessions.py
|
|
'
|
|
|
|
# 2. Register / Update the OpenClaw cron job
|
|
echo "Checking existing OpenClaw Cron automation '$JOB_NAME'..."
|
|
EXISTING_ID=$(kubectl exec -n "$NAMESPACE" "$POD_NAME" -c "$CONTAINER_NAME" -- openclaw cron list --json 2>/dev/null | python3 -c 'import sys, json; data=json.load(sys.stdin); print(next((j["id"] for j in data.get("jobs", []) if j.get("declarationKey") == "'"$JOB_NAME"'"), ""))' 2>/dev/null || true)
|
|
|
|
if [ -n "$EXISTING_ID" ]; then
|
|
echo "Updating existing OpenClaw Cron automation '$EXISTING_ID' ($CRON_SCHEDULE @ $TIMEZONE)..."
|
|
kubectl exec -n "$NAMESPACE" "$POD_NAME" -c "$CONTAINER_NAME" -- openclaw cron edit "$EXISTING_ID" \
|
|
--cron "$CRON_SCHEDULE" \
|
|
--tz "$TIMEZONE" \
|
|
--command "python3 /home/node/.openclaw/scripts/compact_sessions.py"
|
|
else
|
|
echo "Registering new OpenClaw Cron automation '$JOB_NAME' ($CRON_SCHEDULE @ $TIMEZONE)..."
|
|
kubectl exec -n "$NAMESPACE" "$POD_NAME" -c "$CONTAINER_NAME" -- openclaw cron add \
|
|
--name "Nightly Session Compaction" \
|
|
--declaration-key "$JOB_NAME" \
|
|
--description "Nightly multi-agent session compaction and SQLite WAL checkpointing" \
|
|
--cron "$CRON_SCHEDULE" \
|
|
--tz "$TIMEZONE" \
|
|
--session "isolated" \
|
|
--no-deliver \
|
|
--command "python3 /home/node/.openclaw/scripts/compact_sessions.py"
|
|
fi
|
|
|
|
echo "=== Compaction Cron automation successfully registered/updated! ==="
|
|
if [ -n "$EXISTING_ID" ]; then
|
|
kubectl exec -n "$NAMESPACE" "$POD_NAME" -c "$CONTAINER_NAME" -- openclaw cron get "$EXISTING_ID" 2>/dev/null || true
|
|
fi
|