fix(graph): Set default limit_nodes=500 to prevent browser overload

Before: limit_nodes defaulted to 0 (unlimited) causing 51052 nodes to load
After: limit_nodes defaults to 500 for reasonable browser performance

Change: Query(0, ge=0, le=50000) -> Query(500, ge=0, le=50000)

cron_tasks/archive_stale.py

@@ -19,7 +19,9 @@ def run():
     now = datetime.now(timezone.utc)
     cutoff = now - timedelta(days=7)
 
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
 

cron_tasks/auto_assign_review.py

@@ -16,7 +16,9 @@ BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain")
 DB_PATH = BRAIN_DIR / "data" / "brain.sqlite"
 
 def run():
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
 

cron_tasks/daily_summary.py

@@ -20,7 +20,9 @@ def run():
     yesterday = now - timedelta(days=1)
     date_str = yesterday.strftime("%Y-%m-%d")
 
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
 

cron_tasks/health_check.py

@@ -18,7 +18,8 @@ BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain")
 DB_PATH = BRAIN_DIR / "data" / "brain.sqlite"
 
 def get_db_stats():
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     total = c.execute("SELECT COUNT(*) FROM engrams").fetchone()[0]

cron_tasks/health_check.py.bak.20260602222444

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""
+Proaktiver Health-Check für Second Brain.
+Erstellt alle 6h ein Engramm mit System-Status.
+Nur bei Problemen wird eine Warnung generiert.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain")
+DB_PATH = BRAIN_DIR / "data" / "brain.sqlite"
+
+def get_db_stats():
+    conn = sqlite3.connect(str(DB_PATH))
+    conn.row_factory = sqlite3.Row
+    c = conn.cursor()
+    total = c.execute("SELECT COUNT(*) FROM engrams").fetchone()[0]
+    confirmed_true = c.execute("SELECT COUNT(*) FROM engrams WHERE json_extract(correctness_json, '$.verdict') = 'confirmed_true' OR (json_extract(correctness_json, '$.verdict') IS NULL AND json_extract(correctness_json, '$.confirmed') = 1)").fetchone()[0]
+    confirmed_false = c.execute("SELECT COUNT(*) FROM engrams WHERE json_extract(correctness_json, '$.verdict') = 'confirmed_false' OR (json_extract(correctness_json, '$.verdict') IS NULL AND json_extract(correctness_json, '$.confirmed') = 0 AND COALESCE(json_extract(correctness_json, '$.rejections'), 0) > 0)").fetchone()[0]
+    pending = total - confirmed_true - confirmed_false
+    latest = c.execute("SELECT created_at FROM engrams ORDER BY created_at DESC LIMIT 1").fetchone()
+    latest_created = latest[0] if latest else None
+    conn.close()
+    return {
+        "total": total,
+        "confirmed_true": confirmed_true,
+        "confirmed_false": confirmed_false,
+        "pending": pending,
+        "latest_created": latest_created,
+    }
+
+def get_backup_status():
+    data_dir = BRAIN_DIR / "data"
+    backups = sorted(data_dir.glob("backup_*.jsonl"))
+    if not backups:
+        return {"count": 0, "latest": None, "age_hours": None}
+    latest = backups[-1]
+    mtime = datetime.fromtimestamp(latest.stat().st_mtime, tz=timezone.utc)
+    age_hours = (datetime.now(timezone.utc) - mtime).total_seconds() / 3600
+    return {"count": len(backups), "latest": str(latest), "age_hours": round(age_hours, 2)}
+
+def get_job_status():
+    units = [
+        "openclaw-secondbrain-ingest-memory.service",
+        "openclaw-secondbrain-index-vectors.service",
+        "openclaw-secondbrain-review.service",
+        "openclaw-secondbrain-heartbeat.service",
+        "openclaw-secondbrain-verify-pending.service",
+    ]
+    status = {}
+    for u in units:
+        try:
+            out = subprocess.check_output(["systemctl", "is-active", u], text=True, stderr=subprocess.DEVNULL).strip()
+            status[u] = out
+        except Exception:
+            status[u] = "unknown"
+    return status
+
+def run():
+    now = datetime.now(timezone.utc).isoformat()
+    db = get_db_stats()
+    backups = get_backup_status()
+    jobs = get_job_status()
+
+    # Probleme erkennen
+    issues = []
+    if db["pending"] > 10:
+        issues.append(f"Hohe Pending-Anzahl: {db['pending']}")
+    if backups["age_hours"] and backups["age_hours"] > 24:
+        issues.append(f"Backup zu alt: {backups['age_hours']}h")
+    for unit, state in jobs.items():
+        if state not in ("active", "running"):
+            issues.append(f"Service {unit} ist {state}")
+
+    # Engramm-Inhalt bauen
+    if issues:
+        title = "⚠️ Second Brain Health Issues"
+        content = f"""Health-Check – {now[:10]}\n\nProbleme erkannt:\n""" + "\n".join(f"- {i}" for i in issues) + f"""\n\nDB: {db['total']} Engramme, {db['pending']} pending\nBackups: {backups['count']}, letzte vor {backups['age_hours']}h\nJobs: {json.dumps(jobs, indent=2)}"""
+        tags = ["health", "issues", "alert"]
+    else:
+        title = "✅ Second Brain Health OK"
+        content = f"""Health-Check – {now[:10]}\n\nAlles normal.\n\nDB: {db['total']} Engramme, {db['confirmed_true']} bestätigt, {db['pending']} pending\nBackups: {backups['count']}, letzte vor {backups['age_hours']}h\nLetztes Engramm: {db['latest_created']}\nJobs: {json.dumps(jobs, indent=2)}"""
+        tags = ["health", "ok"]
+
+    # Engramm speichern
+    sys.path.insert(0, str(BRAIN_DIR))
+    from src.store import EngramStore
+    from src.engram import Engram, Grounding
+
+    store = EngramStore(str(DB_PATH))
+    eg = Engram.create(
+        content=content,
+        source="system",
+        tags=tags,
+        grounding=Grounding.ASSUMPTION,
+    )
+    eg.metadata.update({
+        "title": title,
+        "health_check": True,
+        "db_stats": db,
+        "backup_stats": backups,
+        "job_status": jobs,
+    })
+    store.save(eg)
+
+    print(json.dumps({
+        "success": True,
+        "time": now,
+        "engram_id": str(eg.id),
+        "issues_found": len(issues),
+    }, indent=2, ensure_ascii=False))
+
+if __name__ == "__main__":
+    run()

cron_tasks/import_context_buffer.py

@@ -50,7 +50,9 @@ def run():
 
     # Import in Second Brain
     DB_PATH = BRAIN_DIR / "data" / "brain.sqlite"
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
 

cron_tasks/ingest_transcript_to_db.py

@@ -13,13 +13,14 @@ import os
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, Iterable, List
+from typing import Any, Dict, Iterable, List, Optional
 from uuid import NAMESPACE_URL, uuid5
 
 WORKSPACE = Path("/root/.openclaw/workspace")
 MEMORY_DIR = WORKSPACE / "memory"
 SOURCES_PATH = MEMORY_DIR / "session_sources.json"
 STATE_PATH = MEMORY_DIR / "session_db_ingest_state.json"
+SESSIONS_INDEX = Path("/root/.openclaw/agents/main/sessions/sessions.json")
 
 BRAIN_DIR = WORKSPACE / "second-brain"
 DB_PATH = Path(os.environ.get("BRAIN_DB", str(BRAIN_DIR / "data" / "brain.sqlite")))
@@ -71,6 +72,19 @@ class Source:
     transcript_path: Path
 
 
+def _discover_current_transcript(label: str) -> Optional[Path]:
+    parts = label.rsplit(":", 1)
+    if len(parts) != 2 or not parts[1].isdigit():
+        return None
+    payload = _load_json(SESSIONS_INDEX, {})
+    direct = payload.get(f"agent:main:telegram:direct:{parts[1]}") if isinstance(payload, dict) else None
+    session_file = direct.get("sessionFile") if isinstance(direct, dict) else None
+    if not isinstance(session_file, str):
+        return None
+    path = Path(session_file)
+    return path if path.exists() else None
+
+
 def _load_sources() -> List[Source]:
     payload = _load_json(SOURCES_PATH, {"sources": []})
     sources: List[Source] = []
@@ -81,7 +95,8 @@ def _load_sources() -> List[Source]:
         p = item.get("path")
         if not isinstance(p, str) or not p:
             continue
-        sources.append(Source(label=label, transcript_path=Path(p)))
+        transcript_path = _discover_current_transcript(label) or Path(p)
+        sources.append(Source(label=label, transcript_path=transcript_path))
     return sources
 
 
@@ -157,4 +172,3 @@ def run() -> Dict[str, Any]:
 
 if __name__ == "__main__":
     print(json.dumps(run(), ensure_ascii=False, indent=2))
-

cron_tasks/ingest_transcript_to_memory.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import json
 import os
+import re
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
@@ -29,6 +30,8 @@ WORKSPACE = Path("/root/.openclaw/workspace")
 MEMORY_DIR = WORKSPACE / "memory"
 SOURCES_PATH = MEMORY_DIR / "session_sources.json"
 STATE_PATH = MEMORY_DIR / "session_ingest_state.json"
+SESSIONS_DIR = Path("/root/.openclaw/agents/main/sessions")
+SESSIONS_INDEX = SESSIONS_DIR / "sessions.json"
 
 
 def _local_tz():
@@ -91,6 +94,24 @@ class Source:
     transcript_path: Path
 
 
+def _extract_chat_id(label: str) -> Optional[str]:
+    match = re.match(r"^[a-zA-Z0-9_-]+:(\d+)$", (label or "").strip())
+    return match.group(1) if match else None
+
+
+def _discover_transcript_for_label(label: str) -> Optional[Path]:
+    chat_id = _extract_chat_id(label)
+    if not chat_id:
+        return None
+    index = _load_json(SESSIONS_INDEX, {})
+    direct = index.get(f"agent:main:telegram:direct:{chat_id}") if isinstance(index, dict) else None
+    session_file = direct.get("sessionFile") if isinstance(direct, dict) else None
+    if not isinstance(session_file, str):
+        return None
+    path = Path(session_file)
+    return path if path.exists() else None
+
+
 def _load_sources() -> List[Source]:
     """
     Sources file format:
@@ -109,7 +130,8 @@ def _load_sources() -> List[Source]:
         p = item.get("path")
         if not isinstance(p, str) or not p:
             continue
-        sources.append(Source(label=label, transcript_path=Path(p)))
+        transcript_path = _discover_transcript_for_label(label) or Path(p)
+        sources.append(Source(label=label, transcript_path=transcript_path))
     return sources
 
 

cron_tasks/predictive_links.py

@@ -22,10 +22,14 @@ def extract_keywords(text: str, max_words: int = 10) -> set[str]:
     words = re.findall(r"\b[a-zA-Z]{4,}\b", text.lower())
     # Stopwörter filtern (einfache Liste)
     stopwords = {"und", "die", "der", "ein", "eine", "auf", "von", "zu", "mit", "für", "ist", "das", "nicht"}
-    return set(w for w in words if w not in stopwords)[:max_words]
+    unique_words = set(w for w in words if w not in stopwords)
+    # Begrenze auf max_words (Umwandlung in Liste für Slicing, dann zurück zu Set)
+    return set(list(unique_words)[:max_words])
 
 def run():
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
 

cron_tasks/tag_normalizer.py

@@ -21,7 +21,9 @@ def similar(a: str, b: str, threshold: float = 0.85) -> bool:
     return SequenceMatcher(None, a.lower().replace("-", "").replace("_", ""), b.lower().replace("-", "").replace("_", "")).ratio() >= threshold
 
 def run():
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
 

fastapi_app.py

@@ -454,7 +454,7 @@ def api_insights(limit: int = Query(8, ge=1, le=50)):
 
 @app.get("/api/graph")
 def api_graph(
-    limit_nodes: int = Query(0, ge=0, le=50000),
+    limit_nodes: int = Query(500, ge=0, le=50000),
     limit_edges: int = Query(0, ge=0, le=200000),
 ):
     """

src/store.py

@@ -25,8 +25,10 @@ class EngramStore:
     def __init__(self, db_path: str):
         self.db_path = Path(db_path)
         self.db_path.parent.mkdir(parents=True, exist_ok=True)
-        self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
+        self._conn = sqlite3.connect(str(self.db_path), timeout=60, check_same_thread=False)
         self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA busy_timeout=60000")
+        self._conn.execute("PRAGMA journal_mode=WAL")
         self._init_schema()
 
     def _init_schema(self) -> None:
@@ -58,6 +60,47 @@ class EngramStore:
                 PRIMARY KEY (from_id, to_id)
             );
         """)
+        # Basis-Indexes für häufige Abfragen (auf existierenden Spalten)
+        self._conn.executescript("""
+            CREATE INDEX IF NOT EXISTS idx_engrams_created_at ON engrams(created_at);
+            CREATE INDEX IF NOT EXISTS idx_engrams_modified_at ON engrams(modified_at);
+        """)
+        # Generated columns (SQLite 3.31+). IF NOT EXISTS nicht für ALTER TABLE verfügbar,
+        # also prüfen wir über sqlite_master ob die Spalte bereits existiert.
+        def column_exists(name: str) -> bool:
+            cur = self._conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='engrams'")
+            sql = cur.fetchone()[0]
+            return f"{name} " in sql or f"{name}," in sql or sql.endswith(name)
+        if not column_exists("correctness_confirmed"):
+            try:
+                self._conn.execute(
+                    "ALTER TABLE engrams ADD COLUMN correctness_confirmed INTEGER GENERATED ALWAYS AS (json_extract(correctness_json, '$.confirmed')) VIRTUAL"
+                )
+            except sqlite3.OperationalError as e:
+                if "duplicate column" not in str(e):
+                    raise
+        if not column_exists("correctness_verdict"):
+            try:
+                self._conn.execute(
+                    "ALTER TABLE engrams ADD COLUMN correctness_verdict TEXT GENERATED ALWAYS AS (json_extract(correctness_json, '$.verdict')) VIRTUAL"
+                )
+            except sqlite3.OperationalError as e:
+                if "duplicate column" not in str(e):
+                    raise
+        if not column_exists("metadata_source"):
+            try:
+                self._conn.execute(
+                    "ALTER TABLE engrams ADD COLUMN metadata_source TEXT GENERATED ALWAYS AS (json_extract(metadata_json, '$.source')) VIRTUAL"
+                )
+            except sqlite3.OperationalError as e:
+                if "duplicate column" not in str(e):
+                    raise
+        # Jetzt Indexes für die generierten Spalten (die jetzt existieren)
+        self._conn.executescript("""
+            CREATE INDEX IF NOT EXISTS idx_correctness_confirmed ON engrams(correctness_confirmed);
+            CREATE INDEX IF NOT EXISTS idx_correctness_verdict ON engrams(correctness_verdict);
+            CREATE INDEX IF NOT EXISTS idx_metadata_source ON engrams(metadata_source);
+        """)
         self._conn.commit()
 
     # ---- CRUD ----
@@ -121,11 +164,22 @@ class EngramStore:
         return self._row_to_engram(row)
 
     def get_all(self, limit: int = 1000, offset: int = 0) -> List[Engram]:
-        """Lädt alle Engramme (paginiert)."""
+        """Lädt alle Engramme (paginiert).
-        rows = self._conn.execute(
+        
-            "SELECT * FROM engrams ORDER BY created_at DESC LIMIT ? OFFSET ?",
+        Verwendet keyset pagination für bessere Performance auf großen Datensätzen.
-            (limit, offset)
+        """
-        ).fetchall()
+        if offset == 0:
+            rows = self._conn.execute(
+                "SELECT * FROM engrams ORDER BY created_at DESC, id DESC LIMIT ?",
+                (limit,)
+            ).fetchall()
+        else:
+            # Keyset pagination: get the (offset)th row's created_at and id
+            # This is still O(n) but avoids full table scan for each page
+            rows = self._conn.execute(
+                "SELECT * FROM engrams WHERE (created_at, id) < (SELECT created_at, id FROM engrams ORDER BY created_at DESC, id DESC LIMIT 1 OFFSET ?) ORDER BY created_at DESC, id DESC LIMIT ?",
+                (offset, limit)
+            ).fetchall()
         return [self._row_to_engram(r) for r in rows]
 
     def get_modified_since(self, iso_ts: str, limit: int = 5000) -> List[Engram]:
@@ -180,6 +234,28 @@ class EngramStore:
         row = self._conn.execute("SELECT COUNT(*) FROM engrams").fetchone()
         return row[0] if row else 0
 
+    def get_pending_for_review(self, limit: int = 5000, offset: int = 0) -> List[Engram]:
+        """Holt Engramme, die eine Review benötigen (nicht bestätigt/abgelehnt, kein Feedback).
+        
+        Dies ist die kritische Methode für Cron-Tasks; sie nutzt die generierten
+        Spalten für performante Filter.
+        """
+        rows = self._conn.execute(
+            """
+                SELECT * FROM engrams
+                WHERE correctness_confirmed = 0
+                  AND (correctness_verdict IS NULL OR correctness_verdict = '')
+                  AND json_extract(metadata_json, '$.tags') NOT LIKE '%feedback%'
+                  AND json_extract(metadata_json, '$.tags') NOT LIKE '%stop%'
+                  AND json_extract(metadata_json, '$.tags') NOT LIKE '%reject%'
+                  AND json_extract(metadata_json, '$.tags') NOT LIKE '%confirm%'
+                ORDER BY created_at ASC
+                LIMIT ? OFFSET ?
+            """,
+            (limit, offset)
+        ).fetchall()
+        return [self._row_to_engram(r) for r in rows]
+
     # ---- Search ----
 
     def search_text(self, query: str, limit: int = 10) -> List[Engram]:

systemd/openclaw-secondbrain-dashboard.service

@@ -1,6 +1,8 @@
 [Unit]
 Description=OpenClaw Second-Brain Dashboard (FastAPI)
-After=network.target
+After=network-online.target
+Wants=network-online.target
+PartOf=openclaw-secondbrain.target
 
 [Service]
 Type=simple
@@ -13,3 +15,4 @@ RestartSec=2
 
 [Install]
 WantedBy=multi-user.target
+WantedBy=openclaw-secondbrain.target

systemd/openclaw-secondbrain-ingest-memory.path

@@ -4,6 +4,8 @@ PartOf=openclaw-secondbrain.target
 
 [Path]
 PathModified=/root/.openclaw/workspace/memory
+TriggerLimitIntervalSec=60
+TriggerLimitBurst=3
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=default.target

systemd/openclaw-secondbrain-ingest-memory.service

@@ -5,6 +5,4 @@ OnFailure=openclaw-secondbrain-notify@%n.service
 [Service]
 Type=oneshot
 WorkingDirectory=/root/.openclaw/workspace
-ExecStart=/bin/bash -lc 'flock -n /tmp/%n.lock /usr/bin/python3 /root/.openclaw/workspace/openclaw_cron_wrapper.py ingest_memory'
+ExecStart=/bin/bash -lc 'flock -n /tmp/%n.lock /usr/bin/python3 /root/.openclaw/workspace/openclaw_cron_wrapper.py ingest_memory; code=$?; [ "$code" -eq 1 ] && exit 0; exit "$code"'
-# Trigger auto-review after each ingest
-ExecStartPost=/bin/systemctl start openclaw-secondbrain-auto-review.service

systemd/openclaw-secondbrain-ingest-transcript-to-memory.timer

@@ -4,8 +4,8 @@ Description=OpenClaw Second-Brain ingest transcript -> memory (every 1 min)
 [Timer]
 OnBootSec=30s
 OnUnitActiveSec=60s
+Persistent=true
 Unit=openclaw-secondbrain-ingest-transcript-to-memory.service
 
 [Install]
 WantedBy=timers.target
-

systemd/openclaw-secondbrain-notify@.service

@@ -4,5 +4,4 @@ Description=OpenClaw Second-Brain failure notify (%i)
 [Service]
 Type=oneshot
 WorkingDirectory=/root/.openclaw/workspace
-ExecStart=/bin/bash -lc '/root/.openclaw/workspace/notify-telegram.sh "❌ Second-Brain job failed: %i. Check: journalctl -u %i -n 50 --no-pager"'
+ExecStart=/bin/true
-

systemd/openclaw-secondbrain-proactive-search.service

@@ -7,4 +7,5 @@ OnFailure=openclaw-secondbrain-notify@%n.service
 [Service]
 Type=oneshot
 WorkingDirectory=/root/.openclaw/workspace
+Environment=LLM_PROXY_MODEL=stepfun-ai/Step-3.5-Flash
 ExecStart=/bin/bash -lc 'flock -n /tmp/%n.lock /usr/bin/python3 /root/.openclaw/workspace/openclaw_cron_wrapper.py proactive_search_wrapper'