diff --git a/k8s/openclaw/configmap-scripts.yaml b/k8s/openclaw/configmap-scripts.yaml index 5bfe25e..733226a 100644 --- a/k8s/openclaw/configmap-scripts.yaml +++ b/k8s/openclaw/configmap-scripts.yaml @@ -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]\ - \ ...\\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]\ + \ ...\\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}] \")\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\"\ diff --git a/k8s/openclaw/scripts/sync_crons.py b/k8s/openclaw/scripts/sync_crons.py index fbf6f71..2748746 100644 --- a/k8s/openclaw/scripts/sync_crons.py +++ b/k8s/openclaw/scripts/sync_crons.py @@ -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...")