Files
infrapuzzle/k8s/openclaw/scripts/compact_tracking_sessions.py
T
Moritz Graf 02fcbd3174 feat(openclaw): strip tracking agent prompt bloat and add 15m deterministic sliding-window compaction
- Disable bootstrap context injection (contextInjection: never) and skill catalog injection (skills: []) on mila_tracker and nutrition_tracker, dropping baseline prompt by 95% (<350 tokens)
- Disable reasoning loops on tracking agents (thinkingDefault: off) for instantaneous sub-second execution
- Add compact_tracking_sessions.py with deterministic 8-line sliding window (--max-lines 8) requiring 0 LLM tokens
- Register automated 15-minute cron job (tracking-session-sliding-window) in OpenClaw gateway
- Streamline setup-compaction-cron.sh to sync python scripts directly and manage both 15m and nightly maintenance
2026-10-14 14:14:35 +02:00

67 lines
2.5 KiB
Python

#!/usr/bin/env python3
"""
Lightweight sliding-window compaction for OpenClaw tracking channels:
- mila_tracker (topic 2796)
- nutrition_tracker (topic 2797)
Maintains a strict sliding window of ~5-8 transcript events (max 2-3 turns).
Uses native gateway truncation (--max-lines 8) requiring 0 LLM tokens and 0 external API calls.
"""
import os
import sqlite3
import subprocess
import sys
import time
MAX_TRACKING_EVENTS = 8
MAX_TRACKING_LINES = 8
TIMEOUT_MS = 30000
TRACKING_TARGETS = [
("mila_tracker", "agent:mila_tracker:telegram:group:-1004326921510:topic:2796"),
("nutrition_tracker", "agent:nutrition_tracker:telegram:group:-1004326921510:topic:2797"),
]
def check_and_compact_tracking():
compacted_count = 0
skipped_count = 0
for agent, key in TRACKING_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=5.0)
row = con.execute("SELECT current_session_id FROM session_nodes WHERE session_key=?", (key,)).fetchone()
if not row or not row[0]:
con.close()
continue
sid = row[0]
cnt = con.execute("SELECT count(*) FROM transcript_events WHERE session_id=?", (sid,)).fetchone()[0]
con.close()
if cnt > MAX_TRACKING_EVENTS:
print(f"[{agent}] Session {key} has {cnt} events (threshold {MAX_TRACKING_EVENTS}). Truncating to last {MAX_TRACKING_LINES} lines...")
cmd = [
"openclaw", "sessions", "compact", key,
"--agent", agent,
"--max-lines", str(MAX_TRACKING_LINES),
"--timeout", str(TIMEOUT_MS)
]
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode == 0:
print(f" ✓ [{agent}] Truncated successfully: {res.stdout.strip()}")
compacted_count += 1
else:
print(f" ✗ [{agent}] Truncation failed: {res.stderr.strip()}", file=sys.stderr)
else:
skipped_count += 1
except Exception as e:
print(f"[{agent}] Error checking session: {e}", file=sys.stderr)
if compacted_count > 0:
print(f"Tracking maintenance: {compacted_count} sessions truncated, {skipped_count} within window.")
if __name__ == "__main__":
check_and_compact_tracking()