moritz/infrapuzzle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(openclaw): add automatic mcp-venv fallback to sync_crons.py

Browse Source
This commit is contained in:
Moritz Graf
2026-10-16 21:28:15 +02:00
parent ed3529a9d7
commit ef50d87bd7
2 changed files with 335 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files
k8s/openclaw/configmap-scripts.yaml
+322 -355
View File
@@ -122,296 +122,6 @@ data:
\ \"Pre-warming SQLite database files into OS page cache for instant cold start...\"\
\ncat /home/node/.openclaw/state/openclaw.sqlite /home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\
\ > /dev/null 2>&1 || true\n\necho \"=== Bootstrap Complete ===\"\n"
gateway-preflight.mjs: " import fs from \"node:fs\";\n import\
\ path from \"node:path\";\n import { execSync, spawn } from \"node:child_process\"\
;\n\n if (!process.env.__OPENCLAW_PREFLIGHT_DONE) {\n process.env.__OPENCLAW_PREFLIGHT_DONE\
\ = \"1\";\n\n const isGateway = process.argv.some(arg => arg.endsWith(\"\
openclaw.mjs\") || arg.endsWith(\"openclaw\")) && process.argv.includes(\"gateway\"\
);\n\n if (isGateway) {\n const t0 = Date.now();\n \
\ console.log(\"[gateway-preflight] Initializing container self-healing\
\ preflight...\");\n\n // 1. Clean up stale gateway lock files from\
\ ungraceful kills\n for (const dir of [\"/tmp/openclaw-1000\", \"\
/tmp\"]) {\n if (fs.existsSync(dir)) {\n try {\n\
\ const files = fs.readdirSync(dir);\n for\
\ (const file of files) {\n if (file.startsWith(\"gateway\"\
) && file.endsWith(\".lock\")) {\n const fullPath = path.join(dir,\
\ file);\n fs.unlinkSync(fullPath);\n \
\ console.log(`[gateway-preflight] Removed stale lock: ${fullPath}`);\n\
\ }\n }\n } catch (e)\
\ {\n console.warn(`[gateway-preflight] Lock cleanup warning\
\ in ${dir}:`, e.message);\n }\n }\n \
\ }\n\n // 2. Clear false database quarantines\n \
\ const qPath = \"/home/node/.openclaw/state/openclaw-quarantine.sqlite\";\n\
\ if (fs.existsSync(qPath)) {\n try {\n \
\ 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\" });\n console.log(\"\
[gateway-preflight] Cleared database quarantine.\");\n } catch\
\ (_) {}\n }\n\n // 3. Deduplicate subagent vector databases\
\ (keeps non-main agent DBs < 15MB)\n try {\n execSync(\"\
/home/node/.openclaw/.local/bin/python3 /home/node/.openclaw/scripts/dedup_subagent_vectors.py\"\
, { stdio: \"inherit\" });\n } catch (e) {\n console.warn(\"\
[gateway-preflight] Vector deduplication warning:\", e.message);\n \
\ }\n\n // 4. Pre-warm page cache sequentially (transforms random\
\ 70 IOPS disk reads into sequential streaming)\n try {\n \
\ const pw0 = Date.now();\n execSync(\"cat /home/node/.openclaw/state/openclaw.sqlite\
\ /home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite > /dev/null 2>&1 ||\
\ true\");\n console.log(`[gateway-preflight] Pre-warmed SQLite\
\ databases in ${Date.now() - pw0}ms`);\n } catch (_) {}\n\n \
\ // 5. Recover any wedged running sessions left over from ungraceful\
\ container restart\n try {\n execSync(\"/home/node/.openclaw/.local/bin/python3\
\ /home/node/.openclaw/scripts/recover_sessions.py\", { stdio: \"inherit\" });\n\
\ } catch (e) {\n console.warn(\"[gateway-preflight]\
\ Session recovery warning:\", e.message);\n }\n\n //\
\ 6. Spawn background turn replay daemon if interrupted turns were detected\n\
\ try {\n if (fs.existsSync(\"/tmp/openclaw-interrupted-turns.json\"\
)) {\n console.log(\"[gateway-preflight] Spawning background\
\ turn replay daemon...\");\n const child = spawn(\"/home/node/.openclaw/.local/bin/python3\"\
, [\"/home/node/.openclaw/scripts/replay_interrupted_turns.py\"], {\n \
\ detached: true,\n stdio: \"ignore\"\n \
\ });\n child.unref();\n }\n \
\ } catch (e) {\n console.warn(\"[gateway-preflight] Could\
\ not spawn turn replay daemon:\", e.message);\n }\n\n \
\ console.log(`[gateway-preflight] Self-healing completed in ${Date.now() -\
\ t0}ms.`);\n }\n }\n"
recover_sessions.py: " #!/usr/bin/env python3\n import glob, sqlite3,\
\ json, os, sys, time\n\n t0 = time.time()\n recovered = 0\n\
\ interrupted_turns = []\n\n def extract_prompt(msg):\n \
\ if not msg: return \"\"\n content = msg.get(\"content\"\
)\n if isinstance(content, str): return content\n if\
\ isinstance(content, list):\n parts = []\n \
\ for c in content:\n if isinstance(c, dict): parts.append(c.get(\"\
text\") or c.get(\"content\") or \"\")\n elif isinstance(c,\
\ str): parts.append(c)\n return \"\\n\".join(filter(None, parts))\n\
\ return str(content or \"\")\n\n for db in sorted(glob.glob(\"\
/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\")):\n \
\ try:\n conn = sqlite3.connect(db, timeout=10.0)\n \
\ cursor = conn.cursor()\n while True:\n \
\ cursor.execute(\"PRAGMA foreign_key_check;\")\n \
\ violations = cursor.fetchall()\n if not violations: break\n\
\ for table, rowid, parent, fkid in violations:\n \
\ cursor.execute(f\"DELETE FROM {table} WHERE rowid = ?\", (rowid,))\n\
\ cursor.execute(\"SELECT session_key, entry_json, current_session_id\
\ FROM session_nodes WHERE status = \\\"running\\\"\")\n rows\
\ = cursor.fetchall()\n agent_id = os.path.basename(os.path.dirname(os.path.dirname(db)))\n\
\ for sk, ej, csid in rows:\n try:\n \
\ p = json.loads(ej) if ej else {}\n \
\ sid = csid or p.get(\"sessionId\")\n if sid:\n\
\ ev = cursor.execute(\"SELECT seq, event_json FROM\
\ transcript_events WHERE session_id = ? ORDER BY seq DESC LIMIT 1\", (sid,)).fetchone()\n\
\ if ev:\n ev_seq,\
\ ev_raw = ev\n try:\n \
\ ev_data = json.loads(ev_raw)\n \
\ msg = ev_data.get(\"message\") or ev_data\n \
\ if msg.get(\"role\") == \"user\":\n \
\ prompt_text = extract_prompt(msg)\n \
\ ctx = p.get(\"restartRecoveryDeliveryContext\") or p.get(\"delivery\"\
, {}).get(\"context\") or p.get(\"delivery\", {}).get(\"route\", {}).get(\"context\"\
) or {}\n interrupted_turns.append({\n\
\ \"agent\": agent_id,\n \
\ \"session_key\": sk,\n \
\ \"session_id\": sid,\n \
\ \"context\": ctx,\n \
\ \"prompt\": prompt_text\n \
\ })\n cursor.execute(\"PRAGMA foreign_keys\
\ = ON\")\n cursor.execute(\"DELETE FROM\
\ transcript_events WHERE session_id = ? AND seq = ?\", (sid, ev_seq))\n \
\ except Exception as ex:\n \
\ print(f\"[recover_sessions] Error parsing event: {ex}\", file=sys.stderr)\n\
\ p[\"status\"] = \"done\"\n \
\ p[\"activeWriterRunId\"] = None\n p[\"activeClaimRunId\"\
] = None\n p[\"restartRecoveryDeliveryRunId\"] = None\n\
\ cursor.execute(\"UPDATE session_nodes SET status =\
\ \\\"done\\\", entry_json = ? WHERE session_key = ?\", (json.dumps(p), sk))\n\
\ cursor.execute(\"UPDATE session_nodes SET entry_valid\
\ = 1 WHERE session_key = ?\", (sk,))\n recovered +=\
\ 1\n except Exception: pass\n cursor.execute(\"\
UPDATE session_windows SET status = \\\"done\\\" WHERE status = \\\"running\\\"\
\")\n conn.commit()\n conn.close()\n \
\ except Exception as e:\n print(f\"[recover_sessions]\
\ Warning for {db}: {e}\", file=sys.stderr)\n\n if interrupted_turns:\n\
\ try:\n with open(\"/tmp/openclaw-interrupted-turns.json\"\
, \"w\") as f:\n json.dump(interrupted_turns, f, indent=2)\n\
\ print(f\"[recover_sessions] Detected {len(interrupted_turns)}\
\ interrupted user turn(s) saved to /tmp/openclaw-interrupted-turns.json\")\n\
\ except Exception as e:\n print(f\"[recover_sessions]\
\ Error writing interrupted turns: {e}\", file=sys.stderr)\n\n print(f\"\
[recover_sessions] Completed in {time.time() - t0:.2f}s ({recovered} wedged sessions\
\ recovered)\")\n"
dedup_subagent_vectors.py: " #!/usr/bin/env python3\n import glob,\
\ os, sqlite3, sys\n\n vec_path = \"/app/node_modules/sqlite-vec-linux-x64/vec0.so\"\
\n for db in sorted(glob.glob(\"/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\"\
)):\n agent = os.path.basename(os.path.dirname(os.path.dirname(db)))\n\
\ if agent == \"main\":\n continue\n \
\ try:\n conn = sqlite3.connect(db, timeout=10.0)\n \
\ if os.path.exists(vec_path):\n conn.enable_load_extension(True)\n\
\ conn.load_extension(vec_path)\n cur =\
\ conn.cursor()\n chk = cur.execute(\"SELECT count(*) FROM memory_index_chunks\"\
).fetchone()[0]\n if chk > 0:\n print(f\"\
[dedup_subagents] Purging {chk} duplicated vector chunks from agent {agent}...\"\
)\n for t in [\n \"memory_index_sources\"\
,\n \"memory_index_chunks\",\n \
\ \"memory_index_chunk_recall_metadata\",\n \"memory_index_chunk_provenance\"\
,\n \"memory_entry_origins\",\n \
\ \"memory_embedding_cache\",\n \"memory_index_chunks_vec\"\
\n ]:\n try: cur.execute(f\"DELETE\
\ FROM {t}\")\n except Exception: pass\n \
\ conn.commit()\n cur.execute(\"VACUUM\")\n \
\ print(f\"[dedup_subagents] Reclaimed disk space for agent {agent}\"\
)\n conn.close()\n except Exception as e:\n \
\ print(f\"[dedup_subagents] Notice for {agent}: {e}\", file=sys.stderr)\n"
replay_interrupted_turns.py: " #!/usr/bin/env python3\n import\
\ json, os, subprocess, sys, time, urllib.request\n\n INTERRUPTED_FILE\
\ = \"/tmp/openclaw-interrupted-turns.json\"\n\n if not os.path.exists(INTERRUPTED_FILE):\n\
\ sys.exit(0)\n\n try:\n with open(INTERRUPTED_FILE,\
\ \"r\") as f:\n turns = json.load(f)\n except Exception\
\ as e:\n print(f\"[replay_interrupted_turns] Failed to read {INTERRUPTED_FILE}:\
\ {e}\", file=sys.stderr)\n sys.exit(0)\n\n if not turns:\n\
\ sys.exit(0)\n\n print(f\"[replay_interrupted_turns] Found\
\ {len(turns)} interrupted turn(s). Waiting for gateway /healthz...\")\n\n \
\ gateway_ready = False\n for _ in range(60):\n try:\n\
\ with urllib.request.urlopen(\"http://127.0.0.1:18790/healthz\"\
, timeout=3) as resp:\n if resp.status == 200:\n \
\ gateway_ready = True\n break\n \
\ except Exception:\n pass\n time.sleep(2)\n\
\n if not gateway_ready:\n print(\"[replay_interrupted_turns]\
\ Gateway did not become healthy in 120s. Aborting.\", file=sys.stderr)\n \
\ sys.exit(1)\n\n # Allow 5s for Telegram ingress to bind\n\
\ time.sleep(5)\n\n for turn in turns:\n sk = turn.get(\"\
session_key\")\n sid = turn.get(\"session_id\")\n agent_id\
\ = turn.get(\"agent\")\n ctx = turn.get(\"context\") or {}\n \
\ prompt = turn.get(\"prompt\") or \"\"\n \n \
\ channel = ctx.get(\"channel\") or \"telegram\"\n to_target =\
\ ctx.get(\"to\") or \"\"\n thread_id = ctx.get(\"threadId\")\n \
\ \n target = to_target\n if target.startswith(\"\
telegram:\"):\n parts = target.split(\":\")\n \
\ target = parts[1] if len(parts) > 1 else target\n\n print(f\"\
[replay_interrupted_turns] Processing agent '{agent_id}' on session '{sk}' target='{target}'\
\ thread='{thread_id}'\")\n \n # 1. Send status notification\
\ to Telegram if target exists\n if target:\n notice\
\ = \"\u26A0\uFE0F OpenClaw wurde w\xE4hrend der Bearbeitung neu gestartet \u2013\
\ deine Anfrage wird jetzt nachgeholt...\"\n notify_cmd = [\n\
\ \"openclaw\", \"message\", \"send\",\n \
\ \"--channel\", channel,\n \"--target\", str(target),\n\
\ \"-m\", notice\n ]\n \
\ if thread_id is not None:\n notify_cmd.extend([\"--thread-id\"\
, str(thread_id)])\n try:\n subprocess.run(notify_cmd,\
\ check=False, timeout=30)\n print(f\"[replay_interrupted_turns]\
\ Sent recovery notice to {target}\")\n except Exception as e:\n\
\ print(f\"[replay_interrupted_turns] Notice failed: {e}\"\
, file=sys.stderr)\n\n # 2. Replay agent turn\n if sk\
\ and prompt:\n p_short = (prompt[:60] + \"...\") if len(prompt)\
\ > 60 else prompt\n print(f\"[replay_interrupted_turns] Replaying\
\ agent turn: {p_short}\")\n agent_cmd = [\n \
\ \"openclaw\", \"agent\",\n \"--session-key\", sk,\n\
\ \"-m\", prompt,\n \"--deliver\"\n\
\ ]\n try:\n res = subprocess.run(agent_cmd,\
\ capture_output=True, text=True, timeout=300)\n if res.returncode\
\ == 0:\n print(f\"[replay_interrupted_turns] Turn replay\
\ succeeded for {sk}\")\n else:\n \
\ print(f\"[replay_interrupted_turns] Turn replay exited with code {res.returncode}:\
\ {res.stderr[:200]}\", file=sys.stderr)\n except Exception as\
\ e:\n print(f\"[replay_interrupted_turns] Replay execution\
\ error: {e}\", file=sys.stderr)\n\n try:\n os.remove(INTERRUPTED_FILE)\n\
\ print(f\"[replay_interrupted_turns] Cleaned up {INTERRUPTED_FILE}\"\
)\n except Exception:\n pass\n"
patch_garmin_mcp.py: "import os\n\np = \"/home/node/.openclaw/mcp-venv/lib/python3.12/site-packages/garmin_mcp/workouts.py\"\
\nif os.path.exists(p):\n with open(p, \"r\") as f:\n c = f.read()\n\
\ if \"_normalize_sport_type\" not in c:\n patch = '''\nSPORT_TYPE_MAPPINGS\
\ = {\n \"running\": {\"sportTypeId\": 1, \"sportTypeKey\": \"running\", \"\
displayOrder\": 1},\n \"cycling\": {\"sportTypeId\": 2, \"sportTypeKey\": \"\
cycling\", \"displayOrder\": 2},\n \"other\": {\"sportTypeId\": 3, \"sportTypeKey\"\
: \"other\", \"displayOrder\": 3},\n \"lap_swimming\": {\"sportTypeId\": 4,\
\ \"sportTypeKey\": \"lap_swimming\", \"displayOrder\": 4},\n \"swimming\"\
: {\"sportTypeId\": 4, \"sportTypeKey\": \"lap_swimming\", \"displayOrder\": 4},\n\
\ \"strength_training\": {\"sportTypeId\": 5, \"sportTypeKey\": \"strength_training\"\
, \"displayOrder\": 5},\n \"cardio_training\": {\"sportTypeId\": 6, \"sportTypeKey\"\
: \"cardio_training\", \"displayOrder\": 6},\n \"cardio\": {\"sportTypeId\"\
: 6, \"sportTypeKey\": \"cardio_training\", \"displayOrder\": 6},\n \"yoga\"\
: {\"sportTypeId\": 7, \"sportTypeKey\": \"yoga\", \"displayOrder\": 8},\n \
\ \"pilates\": {\"sportTypeId\": 8, \"sportTypeKey\": \"pilates\", \"displayOrder\"\
: 9},\n \"hiit\": {\"sportTypeId\": 9, \"sportTypeKey\": \"hiit\", \"displayOrder\"\
: 7},\n \"mobility\": {\"sportTypeId\": 11, \"sportTypeKey\": \"mobility\"\
, \"displayOrder\": 11},\n \"walking\": {\"sportTypeId\": 12, \"sportTypeKey\"\
: \"walking\", \"displayOrder\": 10},\n \"rucking\": {\"sportTypeId\": 13,\
\ \"sportTypeKey\": \"rucking\", \"displayOrder\": 12},\n}\nSPORT_ID_MAPPINGS\
\ = {v[\"sportTypeId\"]: v for v in SPORT_TYPE_MAPPINGS.values()}\n\ndef _resolve_sport_block(st):\n\
\ if isinstance(st, str):\n k = st.strip().lower()\n if k in\
\ SPORT_TYPE_MAPPINGS:\n return dict(SPORT_TYPE_MAPPINGS[k])\n elif\
\ isinstance(st, dict):\n key = st.get(\"sportTypeKey\")\n sid =\
\ st.get(\"sportTypeId\")\n if key and str(key).strip().lower() in SPORT_TYPE_MAPPINGS:\n\
\ base = dict(SPORT_TYPE_MAPPINGS[str(key).strip().lower()])\n \
\ if sid is not None and sid != 0 and sid != base[\"sportTypeId\"]:\n \
\ raise ValueError(f\"sportType mismatch: {key} vs {sid}\")\n \
\ return base\n if sid is not None and sid in SPORT_ID_MAPPINGS:\n\
\ return dict(SPORT_ID_MAPPINGS[sid])\n return None\n\ndef _normalize_sport_type(workout_data:\
\ dict) -> None:\n resolved = _resolve_sport_block(workout_data.get(\"sportType\"\
))\n segments = workout_data.get(\"workoutSegments\") or []\n if not resolved\
\ and segments:\n for seg in segments:\n resolved = _resolve_sport_block(seg.get(\"\
sportType\"))\n if resolved:\n break\n if not resolved:\n\
\ raise ValueError(\"Missing or invalid sportType. Both sportTypeId and\
\ sportTypeKey are required.\")\n workout_data[\"sportType\"] = resolved\n\
\ for seg in segments:\n st = _resolve_sport_block(seg.get(\"sportType\"\
))\n seg[\"sportType\"] = st if st else dict(resolved)\n\n'''\n \
\ t = \"def _normalize_workout_steps(workout_data: dict) -> None:\"\n if\
\ t in c:\n with open(p, \"w\") as f:\n f.write(c.replace(t,\
\ patch + t + \"\\n _normalize_sport_type(workout_data)\", 1))\n \
\ print(\"Successfully patched garmin_mcp workouts.py!\")\n"
patch_perplexity.mjs: "import fs from \"node:fs\";\nimport path from \"node:path\"\
;\nimport { globSync } from \"node:fs\";\n\nconst baseDir = \"/home/node/.openclaw/extensions/perplexity/dist\"\
;\nif (fs.existsSync(baseDir)) {\n const files = fs.readdirSync(baseDir).filter(f\
\ => f.startsWith(\"perplexity-web-search-provider.runtime-\") && f.endsWith(\"\
.js\"));\n for (const filename of files) {\n const file = path.join(baseDir,\
\ filename);\n let code = fs.readFileSync(file, \"utf8\");\n if (!code.includes(\"\
/* PATCHED_CHAT_COMPLETIONS */\")) {\n const target = \"const structured\
\ = runtime.transport === \\\"search_api\\\";\";\n const fastPath = `const\
\ structured = runtime.transport === \"search_api\"; /* PATCHED_CHAT_COMPLETIONS\
\ */\n if (!structured) {\n const cacheKey = buildSearchCacheKey([\n\
\ \"perplexity\",\n runtime.transport,\n runtime.baseUrl,\n\
\ runtime.model,\n query,\n freshness\n \
\ ]);\n const cacheTtlMs = resolveSearchCacheTtlMs(searchConfig);\n\
\ const cached = readCachedSearchPayload(cacheKey, cacheTtlMs);\n \
\ if (cached) return cached;\n const start = Date.now();\n \
\ const timeoutSeconds = resolveSearchTimeoutSeconds(searchConfig);\n \
\ const result = await runPerplexitySearch({\n query,\n \
\ apiKey: runtime.apiKey,\n baseUrl: runtime.baseUrl,\n \
\ model: runtime.model,\n timeoutSeconds,\n signal,\n\
\ freshness\n });\n const payload = {\n \
\ query,\n provider: \"perplexity\",\n model: runtime.model,\n\
\ tookMs: Date.now() - start,\n externalContent: {\n \
\ untrusted: true,\n source: \"web_search\",\n \
\ provider: \"perplexity\",\n wrapped: true\n },\n\
\ content: wrapWebContent(result.content, \"web_search\"),\n \
\ citations: result.citations\n };\n writeCachedSearchPayload(cacheKey,\
\ payload);\n return payload;\n }`;\n if (code.includes(target))\
\ {\n code = code.replace(target, fastPath);\n fs.writeFileSync(file,\
\ code, \"utf8\");\n console.log(\"Patched perplexity runtime for chat\
\ completions:\", file);\n }\n }\n }\n}\n"
file_cli.py: "#!/home/node/.openclaw/mcp-venv/bin/python3\nimport sys, os, mimetypes,\
\ filetype\n\ndef get_file_info(filepath, mime_only=False):\n if not os.path.exists(filepath):\n\
\ return \"cannot open (No such file or directory)\", \"cannot open (No\
\ such file or directory)\"\n if os.path.isdir(filepath):\n return \"\
directory\", \"inode/directory\"\n kind = filetype.guess(filepath)\n if\
\ kind is not None:\n return f\"{kind.extension.upper()} image/media data\
\ ({kind.mime})\", kind.mime\n guessed_mime, _ = mimetypes.guess_type(filepath)\n\
\ try:\n with open(filepath, \"rb\") as f:\n sample = f.read(4096)\n\
\ if len(sample) == 0:\n return \"empty\", \"application/x-empty\"\
\n try:\n sample.decode(\"utf-8\")\n if guessed_mime:\n\
\ return f\"{guessed_mime} text\", guessed_mime\n return\
\ \"ASCII text\", \"text/plain\"\n except UnicodeDecodeError:\n \
\ return \"data\", \"application/octet-stream\"\n except Exception as e:\n\
\ return f\"cannot read ({e})\", \"application/octet-stream\"\n\ndef main():\n\
\ args = sys.argv[1:]\n brief, mime_only, files = False, False, []\n \
\ for arg in args:\n if arg in (\"-b\", \"--brief\"):\n brief\
\ = True\n elif arg in (\"-i\", \"--mime\", \"--mime-type\"):\n \
\ mime_only = True\n elif arg.startswith(\"-\") and len(arg) > 1 and\
\ not arg.startswith(\"--\"):\n if \"b\" in arg: brief = True\n \
\ if \"i\" in arg: mime_only = True\n else:\n files.append(arg)\n\
\ if not files:\n sys.stderr.write(\"Usage: file [-b] [-i|--mime-type]\
\ <file>...\\n\")\n sys.exit(1)\n for f in files:\n desc, mime\
\ = get_file_info(f, mime_only)\n res = mime if mime_only else desc\n \
\ print(res if brief else f\"{f}: {res}\")\n\nif __name__ == \"__main__\"\
:\n main()\n"
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\
@@ -642,68 +352,92 @@ data:
\ if compacted_count > 0:\n print(f\"Tracking maintenance: {compacted_count}\
\ sessions truncated, {skipped_count} within window.\")\n\nif __name__ == \"__main__\"\
:\n check_and_compact_tracking()\n"
sync_crons.py: "#!/usr/bin/env python3\n\"\"\"\nsync_crons.py\nIdempotently reconciles\
\ declarative OpenClaw automations defined in crons.yaml\nwith the running OpenClaw\
\ gateway state.\n\nCan run directly inside the container or remotely via kubectl\
\ exec.\n\"\"\"\n\nimport json\nimport os\nimport subprocess\nimport sys\nimport\
\ yaml\n\ndef find_crons_file():\n candidates = [\n os.environ.get(\"\
CRONS_FILE\"),\n \"crons.yaml\",\n \"k8s/openclaw/crons.yaml\",\n\
\ \"/home/node/.openclaw/scripts/crons.yaml\",\n os.path.join(os.path.dirname(__file__),\
\ \"../crons.yaml\"),\n os.path.join(os.path.dirname(__file__), \"crons.yaml\"\
),\n ]\n for c in candidates:\n if c and os.path.exists(c):\n \
\ return os.path.abspath(c)\n raise FileNotFoundError(\"Could not find\
\ crons.yaml. Pass via CRONS_FILE environment variable.\")\n\ndef is_inside_container():\n\
\ return os.path.exists(\"/home/node/.openclaw\") or os.environ.get(\"OPENCLAW_HOME\"\
) is not None\n\ndef run_openclaw_cmd(args):\n if is_inside_container():\n\
\ cmd = [\"openclaw\"] + args\n else:\n namespace = os.environ.get(\"\
NAMESPACE\", \"openclaw\")\n pod = os.environ.get(\"POD_NAME\", \"openclaw-0\"\
)\n container = os.environ.get(\"CONTAINER_NAME\", \"openclaw\")\n \
\ cmd = [\"kubectl\", \"exec\", \"-n\", namespace, pod, \"-c\", container,\
\ \"--\", \"openclaw\"] + args\n \n res = subprocess.run(cmd, capture_output=True,\
\ text=True)\n if res.returncode != 0:\n raise RuntimeError(f\"OpenClaw\
\ CLI command failed ({' '.join(cmd)}):\\n{res.stderr}\\n{res.stdout}\")\n \
\ return res.stdout\n\ndef list_existing_jobs():\n raw = run_openclaw_cmd([\"\
cron\", \"list\", \"--json\"])\n try:\n data = json.loads(raw)\n \
\ return data.get(\"jobs\", [])\n except Exception as e:\n print(f\"\
Warning: could not parse cron list json: {e}\\nRaw output: {raw[:300]}\", file=sys.stderr)\n\
\ return []\n\ndef reconcile_jobs():\n file_path = find_crons_file()\n\
\ print(f\"=== Syncing OpenClaw Crons-as-Code from {file_path} ===\")\n \
\ \n with open(file_path, \"r\", encoding=\"utf-8\") as f:\n spec =\
\ yaml.safe_load(f)\n\n declarations = spec.get(\"jobs\", [])\n existing_jobs\
\ = list_existing_jobs()\n \n print(f\"Found {len(declarations)} declared\
\ jobs in YAML, {len(existing_jobs)} active jobs on gateway.\")\n\n for job\
\ in declarations:\n name = job.get(\"name\")\n decl_key = job.get(\"\
declarationKey\") or name\n desc = job.get(\"description\", \"\")\n \
\ sched = job.get(\"schedule\", {})\n \n # Match existing job\
\ by declarationKey or exact name\n match = next((j for j in existing_jobs\
\ if j.get(\"declarationKey\") == decl_key or j.get(\"name\") == name), None)\n\
\ \n if match:\n existing_id = match.get(\"id\")\n \
\ print(f\"Removing existing job '{name}' (ID {existing_id}) for idempotent\
\ recreation...\")\n try:\n run_openclaw_cmd([\"cron\"\
, \"rm\", existing_id])\n except Exception as e:\n print(f\"\
Warning removing {existing_id}: {e}\")\n\n # Build add command\n \
\ cmd = [\n \"cron\", \"add\",\n \"--name\", name,\n \
\ \"--declaration-key\", decl_key,\n \"--description\", desc,\n\
\ ]\n\n if sched.get(\"kind\") == \"every\":\n cmd.extend([\"\
--every\", sched.get(\"every\", \"15m\")])\n elif sched.get(\"kind\") ==\
\ \"cron\":\n cmd.extend([\"--cron\", sched.get(\"expr\", \"* * * *\
\ *\")])\n if sched.get(\"tz\"):\n cmd.extend([\"--tz\"\
, sched.get(\"tz\")])\n\n if job.get(\"session\"):\n cmd.extend([\"\
--session\", job.get(\"session\")])\n\n if job.get(\"delivery\") == \"\
none\":\n cmd.append(\"--no-deliver\")\n\n if job.get(\"command\"\
):\n cmd.extend([\"--command\", job.get(\"command\")])\n\n if\
\ job.get(\"agent\"):\n cmd.extend([\"--agent\", job.get(\"agent\"\
)])\n\n if job.get(\"model\"):\n cmd.extend([\"--model\", job.get(\"\
model\")])\n\n if job.get(\"channel\"):\n cmd.extend([\"--channel\"\
, job.get(\"channel\")])\n\n if job.get(\"to\"):\n cmd.extend([\"\
--to\", job.get(\"to\")])\n\n if job.get(\"threadId\"):\n cmd.extend([\"\
--thread-id\", str(job.get(\"threadId\"))])\n\n if job.get(\"announce\"\
):\n cmd.append(\"--announce\")\n\n if job.get(\"message\"):\n\
\ cmd.extend([\"--message\", job.get(\"message\")])\n\n print(f\"\
Registering job '{name}'...\")\n run_openclaw_cmd(cmd)\n print(f\"\
Successfully registered '{name}'.\")\n\n print(\"=== OpenClaw Crons-as-Code\
\ reconciliation complete! ===\")\n\nif __name__ == \"__main__\":\n reconcile_jobs()\n"
triage.py: "#!/usr/bin/env python3\n\"\"\"\nopenclaw-triage.py: Rapid zero-dependency\
dedup_subagent_vectors.py: "#!/usr/bin/env python3\nimport glob, os, sqlite3, sys\n\
\nvec_path = \"/app/node_modules/sqlite-vec-linux-x64/vec0.so\"\nfor db in sorted(glob.glob(\"\
/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\")):\n agent = os.path.basename(os.path.dirname(os.path.dirname(db)))\n\
\ if agent == \"main\":\n continue\n try:\n conn = sqlite3.connect(db,\
\ timeout=10.0)\n if os.path.exists(vec_path):\n conn.enable_load_extension(True)\n\
\ conn.load_extension(vec_path)\n cur = conn.cursor()\n \
\ chk = cur.execute(\"SELECT count(*) FROM memory_index_chunks\").fetchone()[0]\n\
\ if chk > 0:\n print(f\"[dedup_subagents] Purging {chk} duplicated\
\ vector chunks from agent {agent}...\")\n for t in [\n \
\ \"memory_index_sources\",\n \"memory_index_chunks\",\n \
\ \"memory_index_chunk_recall_metadata\",\n \"memory_index_chunk_provenance\"\
,\n \"memory_entry_origins\",\n \"memory_embedding_cache\"\
,\n \"memory_index_chunks_vec\"\n ]:\n \
\ try: cur.execute(f\"DELETE FROM {t}\")\n except Exception:\
\ pass\n conn.commit()\n cur.execute(\"VACUUM\")\n \
\ print(f\"[dedup_subagents] Reclaimed disk space for agent {agent}\")\n\
\ conn.close()\n except Exception as e:\n print(f\"[dedup_subagents]\
\ Notice for {agent}: {e}\", file=sys.stderr)\n"
file_cli.py: "#!/home/node/.openclaw/mcp-venv/bin/python3\nimport sys, os, mimetypes,\
\ filetype\n\ndef get_file_info(filepath, mime_only=False):\n if not os.path.exists(filepath):\n\
\ return \"cannot open (No such file or directory)\", \"cannot open (No\
\ such file or directory)\"\n if os.path.isdir(filepath):\n return \"\
directory\", \"inode/directory\"\n kind = filetype.guess(filepath)\n if\
\ kind is not None:\n return f\"{kind.extension.upper()} image/media data\
\ ({kind.mime})\", kind.mime\n guessed_mime, _ = mimetypes.guess_type(filepath)\n\
\ try:\n with open(filepath, \"rb\") as f:\n sample = f.read(4096)\n\
\ if len(sample) == 0:\n return \"empty\", \"application/x-empty\"\
\n try:\n sample.decode(\"utf-8\")\n if guessed_mime:\n\
\ return f\"{guessed_mime} text\", guessed_mime\n return\
\ \"ASCII text\", \"text/plain\"\n except UnicodeDecodeError:\n \
\ return \"data\", \"application/octet-stream\"\n except Exception as e:\n\
\ return f\"cannot read ({e})\", \"application/octet-stream\"\n\ndef main():\n\
\ args = sys.argv[1:]\n brief, mime_only, files = False, False, []\n \
\ for arg in args:\n if arg in (\"-b\", \"--brief\"):\n brief\
\ = True\n elif arg in (\"-i\", \"--mime\", \"--mime-type\"):\n \
\ mime_only = True\n elif arg.startswith(\"-\") and len(arg) > 1 and\
\ not arg.startswith(\"--\"):\n if \"b\" in arg: brief = True\n \
\ if \"i\" in arg: mime_only = True\n else:\n files.append(arg)\n\
\ if not files:\n sys.stderr.write(\"Usage: file [-b] [-i|--mime-type]\
\ <file>...\\n\")\n sys.exit(1)\n for f in files:\n desc, mime\
\ = get_file_info(f, mime_only)\n res = mime if mime_only else desc\n \
\ print(res if brief else f\"{f}: {res}\")\n\nif __name__ == \"__main__\"\
:\n main()\n"
gateway-preflight.mjs: "import fs from \"node:fs\";\nimport path from \"node:path\"\
;\nimport { execSync, spawn } from \"node:child_process\";\n\nif (!process.env.__OPENCLAW_PREFLIGHT_DONE)\
\ {\n process.env.__OPENCLAW_PREFLIGHT_DONE = \"1\";\n\n const isGateway = process.argv.some(arg\
\ => arg.endsWith(\"openclaw.mjs\") || arg.endsWith(\"openclaw\")) && process.argv.includes(\"\
gateway\");\n\n if (isGateway) {\n const t0 = Date.now();\n console.log(\"\
[gateway-preflight] Initializing container self-healing preflight...\");\n\n \
\ // 1. Clean up stale gateway lock files from ungraceful kills\n for (const\
\ dir of [\"/tmp/openclaw-1000\", \"/tmp\"]) {\n if (fs.existsSync(dir))\
\ {\n try {\n const files = fs.readdirSync(dir);\n for\
\ (const file of files) {\n if (file.startsWith(\"gateway\") && file.endsWith(\"\
.lock\")) {\n const fullPath = path.join(dir, file);\n \
\ fs.unlinkSync(fullPath);\n console.log(`[gateway-preflight]\
\ Removed stale lock: ${fullPath}`);\n }\n }\n } catch\
\ (e) {\n console.warn(`[gateway-preflight] Lock cleanup warning in ${dir}:`,\
\ e.message);\n }\n }\n }\n\n // 2. Clear false database quarantines\n\
\ const qPath = \"/home/node/.openclaw/state/openclaw-quarantine.sqlite\";\n\
\ if (fs.existsSync(qPath)) {\n try {\n 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\"\
\ });\n console.log(\"[gateway-preflight] Cleared database quarantine.\"\
);\n } catch (_) {}\n }\n\n // 3. Deduplicate subagent vector databases\
\ (keeps non-main agent DBs < 15MB)\n try {\n execSync(\"/home/node/.openclaw/.local/bin/python3\
\ /home/node/.openclaw/scripts/dedup_subagent_vectors.py\", { stdio: \"inherit\"\
\ });\n } catch (e) {\n console.warn(\"[gateway-preflight] Vector deduplication\
\ warning:\", e.message);\n }\n\n // 4. Pre-warm page cache sequentially\
\ (transforms random 70 IOPS disk reads into sequential streaming)\n try {\n\
\ const pw0 = Date.now();\n execSync(\"cat /home/node/.openclaw/state/openclaw.sqlite\
\ /home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite > /dev/null 2>&1 ||\
\ true\");\n console.log(`[gateway-preflight] Pre-warmed SQLite databases\
\ in ${Date.now() - pw0}ms`);\n } catch (_) {}\n\n // 5. Recover any wedged\
\ running sessions left over from ungraceful container restart\n try {\n \
\ execSync(\"/home/node/.openclaw/.local/bin/python3 /home/node/.openclaw/scripts/recover_sessions.py\"\
, { stdio: \"inherit\" });\n } catch (e) {\n console.warn(\"[gateway-preflight]\
\ Session recovery warning:\", e.message);\n }\n\n // 6. Spawn background\
\ turn replay daemon if interrupted turns were detected\n try {\n if (fs.existsSync(\"\
/tmp/openclaw-interrupted-turns.json\")) {\n console.log(\"[gateway-preflight]\
\ Spawning background turn replay daemon...\");\n const child = spawn(\"\
/home/node/.openclaw/.local/bin/python3\", [\"/home/node/.openclaw/scripts/replay_interrupted_turns.py\"\
], {\n detached: true,\n stdio: \"ignore\"\n });\n \
\ child.unref();\n }\n } catch (e) {\n console.warn(\"[gateway-preflight]\
\ Could not spawn turn replay daemon:\", e.message);\n }\n\n console.log(`[gateway-preflight]\
\ Self-healing completed in ${Date.now() - t0}ms.`);\n }\n}\n"
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\n\
import glob\nimport json\nimport os\nimport sqlite3\nimport sys\nimport time\n\
@@ -765,9 +499,9 @@ data:
\ 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\"\
\ 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}] <unparseable event json>\")\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\
@@ -825,6 +559,239 @@ data:
\ 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"
patch_garmin_mcp.py: "import os\n\np = \"/home/node/.openclaw/mcp-venv/lib/python3.12/site-packages/garmin_mcp/workouts.py\"\
\nif os.path.exists(p):\n with open(p, \"r\") as f:\n c = f.read()\n\
\ if \"_normalize_sport_type\" not in c:\n patch = '''\nSPORT_TYPE_MAPPINGS\
\ = {\n \"running\": {\"sportTypeId\": 1, \"sportTypeKey\": \"running\", \"\
displayOrder\": 1},\n \"cycling\": {\"sportTypeId\": 2, \"sportTypeKey\": \"\
cycling\", \"displayOrder\": 2},\n \"other\": {\"sportTypeId\": 3, \"sportTypeKey\"\
: \"other\", \"displayOrder\": 3},\n \"lap_swimming\": {\"sportTypeId\": 4,\
\ \"sportTypeKey\": \"lap_swimming\", \"displayOrder\": 4},\n \"swimming\"\
: {\"sportTypeId\": 4, \"sportTypeKey\": \"lap_swimming\", \"displayOrder\": 4},\n\
\ \"strength_training\": {\"sportTypeId\": 5, \"sportTypeKey\": \"strength_training\"\
, \"displayOrder\": 5},\n \"cardio_training\": {\"sportTypeId\": 6, \"sportTypeKey\"\
: \"cardio_training\", \"displayOrder\": 6},\n \"cardio\": {\"sportTypeId\"\
: 6, \"sportTypeKey\": \"cardio_training\", \"displayOrder\": 6},\n \"yoga\"\
: {\"sportTypeId\": 7, \"sportTypeKey\": \"yoga\", \"displayOrder\": 8},\n \
\ \"pilates\": {\"sportTypeId\": 8, \"sportTypeKey\": \"pilates\", \"displayOrder\"\
: 9},\n \"hiit\": {\"sportTypeId\": 9, \"sportTypeKey\": \"hiit\", \"displayOrder\"\
: 7},\n \"mobility\": {\"sportTypeId\": 11, \"sportTypeKey\": \"mobility\"\
, \"displayOrder\": 11},\n \"walking\": {\"sportTypeId\": 12, \"sportTypeKey\"\
: \"walking\", \"displayOrder\": 10},\n \"rucking\": {\"sportTypeId\": 13,\
\ \"sportTypeKey\": \"rucking\", \"displayOrder\": 12},\n}\nSPORT_ID_MAPPINGS\
\ = {v[\"sportTypeId\"]: v for v in SPORT_TYPE_MAPPINGS.values()}\n\ndef _resolve_sport_block(st):\n\
\ if isinstance(st, str):\n k = st.strip().lower()\n if k in\
\ SPORT_TYPE_MAPPINGS:\n return dict(SPORT_TYPE_MAPPINGS[k])\n elif\
\ isinstance(st, dict):\n key = st.get(\"sportTypeKey\")\n sid =\
\ st.get(\"sportTypeId\")\n if key and str(key).strip().lower() in SPORT_TYPE_MAPPINGS:\n\
\ base = dict(SPORT_TYPE_MAPPINGS[str(key).strip().lower()])\n \
\ if sid is not None and sid != 0 and sid != base[\"sportTypeId\"]:\n \
\ raise ValueError(f\"sportType mismatch: {key} vs {sid}\")\n \
\ return base\n if sid is not None and sid in SPORT_ID_MAPPINGS:\n\
\ return dict(SPORT_ID_MAPPINGS[sid])\n return None\n\ndef _normalize_sport_type(workout_data:\
\ dict) -> None:\n resolved = _resolve_sport_block(workout_data.get(\"sportType\"\
))\n segments = workout_data.get(\"workoutSegments\") or []\n if not resolved\
\ and segments:\n for seg in segments:\n resolved = _resolve_sport_block(seg.get(\"\
sportType\"))\n if resolved:\n break\n if not resolved:\n\
\ raise ValueError(\"Missing or invalid sportType. Both sportTypeId and\
\ sportTypeKey are required.\")\n workout_data[\"sportType\"] = resolved\n\
\ for seg in segments:\n st = _resolve_sport_block(seg.get(\"sportType\"\
))\n seg[\"sportType\"] = st if st else dict(resolved)\n\n'''\n \
\ t = \"def _normalize_workout_steps(workout_data: dict) -> None:\"\n if\
\ t in c:\n with open(p, \"w\") as f:\n f.write(c.replace(t,\
\ patch + t + \"\\n _normalize_sport_type(workout_data)\", 1))\n \
\ print(\"Successfully patched garmin_mcp workouts.py!\")\n"
patch_perplexity.mjs: "import fs from \"node:fs\";\nimport path from \"node:path\"\
;\nimport { globSync } from \"node:fs\";\n\nconst baseDir = \"/home/node/.openclaw/extensions/perplexity/dist\"\
;\nif (fs.existsSync(baseDir)) {\n const files = fs.readdirSync(baseDir).filter(f\
\ => f.startsWith(\"perplexity-web-search-provider.runtime-\") && f.endsWith(\"\
.js\"));\n for (const filename of files) {\n const file = path.join(baseDir,\
\ filename);\n let code = fs.readFileSync(file, \"utf8\");\n if (!code.includes(\"\
/* PATCHED_CHAT_COMPLETIONS */\")) {\n const target = \"const structured\
\ = runtime.transport === \\\"search_api\\\";\";\n const fastPath = `const\
\ structured = runtime.transport === \"search_api\"; /* PATCHED_CHAT_COMPLETIONS\
\ */\n if (!structured) {\n const cacheKey = buildSearchCacheKey([\n\
\ \"perplexity\",\n runtime.transport,\n runtime.baseUrl,\n\
\ runtime.model,\n query,\n freshness\n \
\ ]);\n const cacheTtlMs = resolveSearchCacheTtlMs(searchConfig);\n\
\ const cached = readCachedSearchPayload(cacheKey, cacheTtlMs);\n \
\ if (cached) return cached;\n const start = Date.now();\n \
\ const timeoutSeconds = resolveSearchTimeoutSeconds(searchConfig);\n \
\ const result = await runPerplexitySearch({\n query,\n \
\ apiKey: runtime.apiKey,\n baseUrl: runtime.baseUrl,\n \
\ model: runtime.model,\n timeoutSeconds,\n signal,\n\
\ freshness\n });\n const payload = {\n \
\ query,\n provider: \"perplexity\",\n model: runtime.model,\n\
\ tookMs: Date.now() - start,\n externalContent: {\n \
\ untrusted: true,\n source: \"web_search\",\n \
\ provider: \"perplexity\",\n wrapped: true\n },\n\
\ content: wrapWebContent(result.content, \"web_search\"),\n \
\ citations: result.citations\n };\n writeCachedSearchPayload(cacheKey,\
\ payload);\n return payload;\n }`;\n if (code.includes(target))\
\ {\n code = code.replace(target, fastPath);\n fs.writeFileSync(file,\
\ code, \"utf8\");\n console.log(\"Patched perplexity runtime for chat\
\ completions:\", file);\n }\n }\n }\n}\n"
recover_sessions.py: "#!/usr/bin/env python3\nimport glob, sqlite3, json, os, sys,\
\ time\n\nt0 = time.time()\nrecovered = 0\ninterrupted_turns = []\n\ndef extract_prompt(msg):\n\
\ if not msg: return \"\"\n content = msg.get(\"content\")\n if isinstance(content,\
\ str): return content\n if isinstance(content, list):\n parts = []\n\
\ for c in content:\n if isinstance(c, dict): parts.append(c.get(\"\
text\") or c.get(\"content\") or \"\")\n elif isinstance(c, str): parts.append(c)\n\
\ return \"\\n\".join(filter(None, parts))\n return str(content or \"\
\")\n\nfor db in sorted(glob.glob(\"/home/node/.openclaw/agents/*/agent/openclaw-agent.sqlite\"\
)):\n try:\n conn = sqlite3.connect(db, timeout=10.0)\n cursor\
\ = conn.cursor()\n while True:\n cursor.execute(\"PRAGMA foreign_key_check;\"\
)\n violations = cursor.fetchall()\n if not violations:\
\ break\n for table, rowid, parent, fkid in violations:\n \
\ cursor.execute(f\"DELETE FROM {table} WHERE rowid = ?\", (rowid,))\n \
\ cursor.execute(\"SELECT session_key, entry_json, current_session_id FROM\
\ session_nodes WHERE status = \\\"running\\\"\")\n rows = cursor.fetchall()\n\
\ agent_id = os.path.basename(os.path.dirname(os.path.dirname(db)))\n \
\ for sk, ej, csid in rows:\n try:\n p = json.loads(ej)\
\ if ej else {}\n sid = csid or p.get(\"sessionId\")\n \
\ if sid:\n ev = cursor.execute(\"SELECT seq, event_json\
\ FROM transcript_events WHERE session_id = ? ORDER BY seq DESC LIMIT 1\", (sid,)).fetchone()\n\
\ if ev:\n ev_seq, ev_raw = ev\n \
\ try:\n ev_data = json.loads(ev_raw)\n\
\ msg = ev_data.get(\"message\") or ev_data\n \
\ if msg.get(\"role\") == \"user\":\n \
\ prompt_text = extract_prompt(msg)\n \
\ ctx = p.get(\"restartRecoveryDeliveryContext\") or p.get(\"delivery\", {}).get(\"\
context\") or p.get(\"delivery\", {}).get(\"route\", {}).get(\"context\") or {}\n\
\ interrupted_turns.append({\n \
\ \"agent\": agent_id,\n \
\ \"session_key\": sk,\n \"session_id\": sid,\n\
\ \"context\": ctx,\n \
\ \"prompt\": prompt_text\n })\n\
\ cursor.execute(\"PRAGMA foreign_keys = ON\")\n\
\ cursor.execute(\"DELETE FROM transcript_events\
\ WHERE session_id = ? AND seq = ?\", (sid, ev_seq))\n \
\ except Exception as ex:\n print(f\"[recover_sessions]\
\ Error parsing event: {ex}\", file=sys.stderr)\n p[\"status\"\
] = \"done\"\n p[\"activeWriterRunId\"] = None\n \
\ p[\"activeClaimRunId\"] = None\n p[\"restartRecoveryDeliveryRunId\"\
] = None\n cursor.execute(\"UPDATE session_nodes SET status = \\\
\"done\\\", entry_json = ? WHERE session_key = ?\", (json.dumps(p), sk))\n \
\ cursor.execute(\"UPDATE session_nodes SET entry_valid = 1 WHERE\
\ session_key = ?\", (sk,))\n recovered += 1\n except\
\ Exception: pass\n cursor.execute(\"UPDATE session_windows SET status\
\ = \\\"done\\\" WHERE status = \\\"running\\\"\")\n conn.commit()\n \
\ conn.close()\n except Exception as e:\n print(f\"[recover_sessions]\
\ Warning for {db}: {e}\", file=sys.stderr)\n\nif interrupted_turns:\n try:\n\
\ with open(\"/tmp/openclaw-interrupted-turns.json\", \"w\") as f:\n \
\ json.dump(interrupted_turns, f, indent=2)\n print(f\"[recover_sessions]\
\ Detected {len(interrupted_turns)} interrupted user turn(s) saved to /tmp/openclaw-interrupted-turns.json\"\
)\n except Exception as e:\n print(f\"[recover_sessions] Error writing\
\ interrupted turns: {e}\", file=sys.stderr)\n\nprint(f\"[recover_sessions] Completed\
\ in {time.time() - t0:.2f}s ({recovered} wedged sessions recovered)\")\n"
replay_interrupted_turns.py: "#!/usr/bin/env python3\nimport json, os, subprocess,\
\ sys, time, urllib.request\n\nINTERRUPTED_FILE = \"/tmp/openclaw-interrupted-turns.json\"\
\n\nif not os.path.exists(INTERRUPTED_FILE):\n sys.exit(0)\n\ntry:\n with\
\ open(INTERRUPTED_FILE, \"r\") as f:\n turns = json.load(f)\nexcept Exception\
\ as e:\n print(f\"[replay_interrupted_turns] Failed to read {INTERRUPTED_FILE}:\
\ {e}\", file=sys.stderr)\n sys.exit(0)\n\nif not turns:\n sys.exit(0)\n\
\nprint(f\"[replay_interrupted_turns] Found {len(turns)} interrupted turn(s).\
\ Waiting for gateway /healthz...\")\n\ngateway_ready = False\nfor _ in range(60):\n\
\ try:\n with urllib.request.urlopen(\"http://127.0.0.1:18790/healthz\"\
, timeout=3) as resp:\n if resp.status == 200:\n gateway_ready\
\ = True\n break\n except Exception:\n pass\n time.sleep(2)\n\
\nif not gateway_ready:\n print(\"[replay_interrupted_turns] Gateway did not\
\ become healthy in 120s. Aborting.\", file=sys.stderr)\n sys.exit(1)\n\n#\
\ Allow 5s for Telegram ingress to bind\ntime.sleep(5)\n\nfor turn in turns:\n\
\ sk = turn.get(\"session_key\")\n sid = turn.get(\"session_id\")\n agent_id\
\ = turn.get(\"agent\")\n ctx = turn.get(\"context\") or {}\n prompt = turn.get(\"\
prompt\") or \"\"\n\n channel = ctx.get(\"channel\") or \"telegram\"\n to_target\
\ = ctx.get(\"to\") or \"\"\n thread_id = ctx.get(\"threadId\")\n\n target\
\ = to_target\n if target.startswith(\"telegram:\"):\n parts = target.split(\"\
:\")\n target = parts[1] if len(parts) > 1 else target\n\n print(f\"\
[replay_interrupted_turns] Processing agent '{agent_id}' on session '{sk}' target='{target}'\
\ thread='{thread_id}'\")\n\n # 1. Send status notification to Telegram if\
\ target exists\n if target:\n notice = \"\u26A0\uFE0F OpenClaw wurde\
\ w\xE4hrend der Bearbeitung neu gestartet \u2013 deine Anfrage wird jetzt nachgeholt...\"\
\n notify_cmd = [\n \"openclaw\", \"message\", \"send\",\n \
\ \"--channel\", channel,\n \"--target\", str(target),\n\
\ \"-m\", notice\n ]\n if thread_id is not None:\n \
\ notify_cmd.extend([\"--thread-id\", str(thread_id)])\n try:\n\
\ subprocess.run(notify_cmd, check=False, timeout=30)\n \
\ print(f\"[replay_interrupted_turns] Sent recovery notice to {target}\")\n \
\ except Exception as e:\n print(f\"[replay_interrupted_turns]\
\ Notice failed: {e}\", file=sys.stderr)\n\n # 2. Replay agent turn\n if\
\ sk and prompt:\n p_short = (prompt[:60] + \"...\") if len(prompt) > 60\
\ else prompt\n print(f\"[replay_interrupted_turns] Replaying agent turn:\
\ {p_short}\")\n agent_cmd = [\n \"openclaw\", \"agent\",\n\
\ \"--session-key\", sk,\n \"-m\", prompt,\n \
\ \"--deliver\"\n ]\n try:\n res = subprocess.run(agent_cmd,\
\ capture_output=True, text=True, timeout=300)\n if res.returncode\
\ == 0:\n print(f\"[replay_interrupted_turns] Turn replay succeeded\
\ for {sk}\")\n else:\n print(f\"[replay_interrupted_turns]\
\ Turn replay exited with code {res.returncode}: {res.stderr[:200]}\", file=sys.stderr)\n\
\ except Exception as e:\n print(f\"[replay_interrupted_turns]\
\ Replay execution error: {e}\", file=sys.stderr)\n\ntry:\n os.remove(INTERRUPTED_FILE)\n\
\ print(f\"[replay_interrupted_turns] Cleaned up {INTERRUPTED_FILE}\")\nexcept\
\ Exception:\n pass\n"
sync_crons.py: "#!/usr/bin/env python3\n\"\"\"\nsync_crons.py\nIdempotently reconciles\
\ declarative OpenClaw automations defined in crons.yaml\nwith the running OpenClaw\
\ gateway state.\n\nCan run directly inside the container or remotely via kubectl\
\ exec.\n\"\"\"\n\nimport json\nimport os\nimport subprocess\nimport sys\n\ntry:\n\
\ import yaml\nexcept ImportError:\n venv_py = \"/home/node/.openclaw/mcp-venv/bin/python3\"\
\n if os.path.exists(venv_py) and sys.executable != venv_py:\n os.execv(venv_py,\
\ [venv_py] + sys.argv)\n raise\n\ndef find_crons_file():\n candidates =\
\ [\n os.environ.get(\"CRONS_FILE\"),\n \"crons.yaml\",\n \
\ \"k8s/openclaw/crons.yaml\",\n \"/home/node/.openclaw/scripts/crons.yaml\"\
,\n os.path.join(os.path.dirname(__file__), \"../crons.yaml\"),\n \
\ os.path.join(os.path.dirname(__file__), \"crons.yaml\"),\n ]\n for c\
\ in candidates:\n if c and os.path.exists(c):\n return os.path.abspath(c)\n\
\ raise FileNotFoundError(\"Could not find crons.yaml. Pass via CRONS_FILE\
\ environment variable.\")\n\ndef is_inside_container():\n return os.path.exists(\"\
/home/node/.openclaw\") or os.environ.get(\"OPENCLAW_HOME\") is not None\n\ndef\
\ run_openclaw_cmd(args):\n if is_inside_container():\n cmd = [\"openclaw\"\
] + args\n else:\n namespace = os.environ.get(\"NAMESPACE\", \"openclaw\"\
)\n pod = os.environ.get(\"POD_NAME\", \"openclaw-0\")\n container\
\ = os.environ.get(\"CONTAINER_NAME\", \"openclaw\")\n cmd = [\"kubectl\"\
, \"exec\", \"-n\", namespace, pod, \"-c\", container, \"--\", \"openclaw\"] +\
\ args\n\n res = subprocess.run(cmd, capture_output=True, text=True)\n if\
\ res.returncode != 0:\n raise RuntimeError(f\"OpenClaw CLI command failed\
\ ({' '.join(cmd)}):\\n{res.stderr}\\n{res.stdout}\")\n return res.stdout\n\
\ndef list_existing_jobs():\n raw = run_openclaw_cmd([\"cron\", \"list\", \"\
--json\"])\n try:\n data = json.loads(raw)\n return data.get(\"\
jobs\", [])\n except Exception as e:\n print(f\"Warning: could not parse\
\ cron list json: {e}\\nRaw output: {raw[:300]}\", file=sys.stderr)\n return\
\ []\n\ndef reconcile_jobs():\n file_path = find_crons_file()\n print(f\"\
=== Syncing OpenClaw Crons-as-Code from {file_path} ===\")\n\n with open(file_path,\
\ \"r\", encoding=\"utf-8\") as f:\n spec = yaml.safe_load(f)\n\n declarations\
\ = spec.get(\"jobs\", [])\n existing_jobs = list_existing_jobs()\n\n print(f\"\
Found {len(declarations)} declared jobs in YAML, {len(existing_jobs)} active jobs\
\ on gateway.\")\n\n for job in declarations:\n name = job.get(\"name\"\
)\n decl_key = job.get(\"declarationKey\") or name\n desc = job.get(\"\
description\", \"\")\n sched = job.get(\"schedule\", {})\n\n # Match\
\ existing job by declarationKey or exact name\n match = next((j for j\
\ in existing_jobs if j.get(\"declarationKey\") == decl_key or j.get(\"name\"\
) == name), None)\n\n if match:\n existing_id = match.get(\"\
id\")\n print(f\"Removing existing job '{name}' (ID {existing_id})\
\ for idempotent recreation...\")\n try:\n run_openclaw_cmd([\"\
cron\", \"rm\", existing_id])\n except Exception as e:\n \
\ print(f\"Warning removing {existing_id}: {e}\")\n\n # Build add\
\ command\n cmd = [\n \"cron\", \"add\",\n \"--name\"\
, name,\n \"--declaration-key\", decl_key,\n \"--description\"\
, desc,\n ]\n\n if sched.get(\"kind\") == \"every\":\n \
\ cmd.extend([\"--every\", sched.get(\"every\", \"15m\")])\n elif sched.get(\"\
kind\") == \"cron\":\n cmd.extend([\"--cron\", sched.get(\"expr\",\
\ \"* * * * *\")])\n if sched.get(\"tz\"):\n cmd.extend([\"\
--tz\", sched.get(\"tz\")])\n\n if job.get(\"session\"):\n cmd.extend([\"\
--session\", job.get(\"session\")])\n\n if job.get(\"delivery\") == \"\
none\":\n cmd.append(\"--no-deliver\")\n\n if job.get(\"command\"\
):\n cmd.extend([\"--command\", job.get(\"command\")])\n\n if\
\ job.get(\"agent\"):\n cmd.extend([\"--agent\", job.get(\"agent\"\
)])\n\n if job.get(\"model\"):\n cmd.extend([\"--model\", job.get(\"\
model\")])\n\n if job.get(\"channel\"):\n cmd.extend([\"--channel\"\
, job.get(\"channel\")])\n\n if job.get(\"to\"):\n cmd.extend([\"\
--to\", job.get(\"to\")])\n\n if job.get(\"threadId\"):\n cmd.extend([\"\
--thread-id\", str(job.get(\"threadId\"))])\n\n if job.get(\"announce\"\
):\n cmd.append(\"--announce\")\n\n if job.get(\"message\"):\n\
\ cmd.extend([\"--message\", job.get(\"message\")])\n\n print(f\"\
Registering job '{name}'...\")\n run_openclaw_cmd(cmd)\n print(f\"\
Successfully registered '{name}'.\")\n\n print(\"=== OpenClaw Crons-as-Code\
\ reconciliation complete! ===\")\n\nif __name__ == \"__main__\":\n reconcile_jobs()\n"
crons.yaml: "# Declarative Crons-as-Code for OpenClaw (Haumdaucher)\n# Reconciled\
\ idempotently via k8s/openclaw/scripts/sync_crons.py\n\nversion: 1\njobs:\n \
\ # 1. Maintenance & State Truncation\n - name: \"tracking-session-sliding-window\"\
k8s/openclaw/scripts/sync_crons.py
+13 -6
View File
@@ -11,7 +11,14 @@ import json
import os
import subprocess
import sys
import yaml
try:
import yaml
except ImportError:
venv_py = "/home/node/.openclaw/mcp-venv/bin/python3"
if os.path.exists(venv_py) and sys.executable != venv_py:
os.execv(venv_py, [venv_py] + sys.argv)
raise
def find_crons_file():
candidates = [
@@ -38,7 +45,7 @@ def run_openclaw_cmd(args):
pod = os.environ.get("POD_NAME", "openclaw-0")
container = os.environ.get("CONTAINER_NAME", "openclaw")
cmd = ["kubectl", "exec", "-n", namespace, pod, "-c", container, "--", "openclaw"] + args
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode != 0:
raise RuntimeError(f"OpenClaw CLI command failed ({' '.join(cmd)}):\n{res.stderr}\n{res.stdout}")
@@ -56,13 +63,13 @@ def list_existing_jobs():
def reconcile_jobs():
file_path = find_crons_file()
print(f"=== Syncing OpenClaw Crons-as-Code from {file_path} ===")
with open(file_path, "r", encoding="utf-8") as f:
spec = yaml.safe_load(f)
declarations = spec.get("jobs", [])
existing_jobs = list_existing_jobs()
print(f"Found {len(declarations)} declared jobs in YAML, {len(existing_jobs)} active jobs on gateway.")
for job in declarations:
@@ -70,10 +77,10 @@ def reconcile_jobs():
decl_key = job.get("declarationKey") or name
desc = job.get("description", "")
sched = job.get("schedule", {})
# Match existing job by declarationKey or exact name
match = next((j for j in existing_jobs if j.get("declarationKey") == decl_key or j.get("name") == name), None)
if match:
existing_id = match.get("id")
print(f"Removing existing job '{name}' (ID {existing_id}) for idempotent recreation...")