gardener.py

# -*- coding: utf-8 -*-
"""
Gardener -- LLM-natives Betriebssystem
======================================

Vier Funktionen. Eine Suche. Zwei Datenbanken.

    from gardener import Gardener
    af = Gardener()
    af.find("steuer")
    af.get("beleg-scanner")
    af.put("notiz", content="Wichtig!", type="memory")
    af.run("beleg-scanner", input={"pfad": "rechnung.pdf"})

Konzept: KONZEPT.md
"""
import json
import hashlib
import os
import re
import shutil
import sqlite3
import subprocess
import sys
import tempfile
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union


# ---------------------------------------------------------------------------
# Konfiguration
# ---------------------------------------------------------------------------

# DB lebt lokal (nicht in Cloud-Sync)
LOCAL_DATA_DIR = Path(os.environ.get(
    "GARDENER_DATA",
    os.path.expanduser("~/.gardener")
))

# Der Ordner den der User sieht (darf in einem Cloud-Sync-Ordner liegen)
DEFAULT_HOME = Path(os.environ.get(
    "GARDENER_HOME",
    os.path.expanduser("~/gardener")
))

# Schwellenwerte für Blob-Halde
BLOB_THRESHOLD_DIRECT = 1_000_000      # < 1MB: direkt in DB
BLOB_THRESHOLD_WARN   = 50_000_000     # < 50MB: BLOB in DB mit Warnung
# > 50MB: nur Index + Halde

# Path segments that observe()/sync() must never index. Derived from the
# never-index list in sources.py so there is exactly one list to maintain:
# what a federated source adapter refuses to read, the home-folder walk
# refuses too (credentials, .ssh, node_modules, .git, .gardener, ...).
try:
    from sources import EXCLUDED_PATH_SEGMENTS as _EXCLUDED_SEGMENTS
    from sources import is_excluded as _is_excluded_path
except ImportError:  # pragma: no cover - sources.py is a sibling module
    _EXCLUDED_SEGMENTS = frozenset({
        ".absorber", ".output", ".gardener", "__pycache__",
        "credentials", ".ssh", "node_modules", ".git",
    })
    _is_excluded_path = None
INTERNAL_SKIP_PREFIXES = tuple(sorted(_EXCLUDED_SEGMENTS))
# Internal top-level files that observe()/sync() must never index or absorb
INTERNAL_SKIP_FILES = ("config.json",)

# A credential signature found inside a CLOUD-SYNCED document is a security
# finding in its own right: the value has left the machine. The same
# signature inside a local agent transcript (~/.codex, ~/.claude) is not --
# it never left. So the alert fires on source path, not on the finding.
CLOUD_ALERT_ROOT = Path(os.environ.get(
    "GARDENER_CLOUD_ROOT", os.path.expanduser("~/OneDrive")))
CLOUD_ALERT_FILE = Path(os.environ.get(
    "GARDENER_CLOUD_ALERT_FILE",
    os.path.expanduser(
        "~/OneDrive/.SYNC/laptop/SECURITY-ALERT_TOKEN-IN-ONEDRIVE.md")))


# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------

SCHEMA_SYSTEM = """
CREATE TABLE IF NOT EXISTS everything (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    type TEXT NOT NULL DEFAULT 'knowledge',
    name TEXT NOT NULL UNIQUE,
    content TEXT DEFAULT '',
    tags TEXT DEFAULT '',
    meta TEXT DEFAULT '{}',
    pinned INTEGER DEFAULT 0,
    created TEXT NOT NULL,
    updated TEXT NOT NULL
);

CREATE VIRTUAL TABLE IF NOT EXISTS everything_fts
    USING fts5(name, content, tags, content=everything, content_rowid=id);

-- Trigger für FTS-Sync
CREATE TRIGGER IF NOT EXISTS everything_ai AFTER INSERT ON everything BEGIN
    INSERT INTO everything_fts(rowid, name, content, tags)
    VALUES (new.id, new.name, new.content, new.tags);
END;

CREATE TRIGGER IF NOT EXISTS everything_ad AFTER DELETE ON everything BEGIN
    INSERT INTO everything_fts(everything_fts, rowid, name, content, tags)
    VALUES ('delete', old.id, old.name, old.content, old.tags);
END;

CREATE TRIGGER IF NOT EXISTS everything_au AFTER UPDATE ON everything BEGIN
    INSERT INTO everything_fts(everything_fts, rowid, name, content, tags)
    VALUES ('delete', old.id, old.name, old.content, old.tags);
    INSERT INTO everything_fts(rowid, name, content, tags)
    VALUES (new.id, new.name, new.content, new.tags);
END;

-- Optionale Fachtabellen: Regal-Registry
CREATE TABLE IF NOT EXISTS shelves (
    name TEXT PRIMARY KEY,
    description TEXT DEFAULT '',
    schema_json TEXT DEFAULT '{}',
    created TEXT NOT NULL
);

-- Hinweis: Blob-Metadaten (blob_path, blob_hash, size, mimetype,
-- original_name) liegen bewusst im meta-JSON des everything-Eintrags.
-- Eine frühere separate blobs-Tabelle wurde nie befüllt und entfernt
-- (siehe CHANGELOG 2026-06-12).
"""


# ---------------------------------------------------------------------------
# Gardener Core
# ---------------------------------------------------------------------------

class Gardener:
    """Das LLM-native Betriebssystem. Vier Funktionen, eine Suche."""

    def __init__(self, home: Optional[Path] = None, data_dir: Optional[Path] = None):
        self.home = Path(home) if home else DEFAULT_HOME
        self.data_dir = Path(data_dir) if data_dir else LOCAL_DATA_DIR
        self.blob_dir = self.data_dir / "blobs"
        self.workspace_dir = self.data_dir / "workspace"
        self.absorber_dir = self.home / ".absorber"
        self.output_dir = self.home / ".output"

        # Verzeichnisse anlegen
        for d in (self.home, self.data_dir, self.blob_dir,
                  self.workspace_dir, self.absorber_dir, self.output_dir):
            d.mkdir(parents=True, exist_ok=True)

        # Config laden
        self.config = self._load_config()

        # Datenbanken initialisieren
        self.system_db_path = self.data_dir / "gardener.db"
        self.user_db_path = self.data_dir / "user.db"

        self._init_db(self.system_db_path)
        self._init_db(self.user_db_path)

    # ------------------------------------------------------------------
    # DB Setup
    # ------------------------------------------------------------------

    def _init_db(self, db_path: Path):
        """Initialisiert eine Datenbank mit dem Schema."""
        conn = sqlite3.connect(str(db_path))
        try:
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute("PRAGMA foreign_keys=ON")
            conn.executescript(SCHEMA_SYSTEM)
            conn.commit()
        finally:
            conn.close()

    def _conn(self, target: str = "user") -> sqlite3.Connection:
        """Gibt eine Connection zurück. 'user' oder 'system'."""
        path = self.user_db_path if target == "user" else self.system_db_path
        conn = sqlite3.connect(str(path))
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA foreign_keys=ON")

        # Andere DB attachen für übergreifende Suche
        other = self.system_db_path if target == "user" else self.user_db_path
        conn.execute(f"ATTACH DATABASE ? AS other", (str(other),))
        return conn

    @contextmanager
    def connection(self, target: str = "user"):
        """Context Manager fuer DB-Connections mit garantierter Schliessung."""
        conn = self._conn(target)
        try:
            yield conn
        finally:
            conn.close()

    def _now(self) -> str:
        return datetime.now().isoformat(timespec="seconds")

    # ------------------------------------------------------------------
    # API: find()
    # ------------------------------------------------------------------

    @staticmethod
    def _build_fts_or_query(query: str) -> Optional[str]:
        """Baut aus einer Mehrwort-Query eine FTS5-OR-Query mit Anführungszeichen.

        Gibt None zurück, wenn die Query bereits explizite FTS-Operatoren oder
        Anführungszeichen enthält oder aus nur einem Wort besteht.
        """
        if '"' in query or any(op in query.upper().split() for op in ["OR", "AND", "NOT", "NEAR"]):
            return None

        tokens = [t.strip() for t in query.split() if t.strip()]
        if len(tokens) <= 1:
            return None

        cleaned = []
        for t in tokens:
            t_clean = t.replace('"', '""')
            if t_clean:
                cleaned.append(f'"{t_clean}"')

        if len(cleaned) <= 1:
            return None

        return " OR ".join(cleaned)

    @staticmethod
    def _like_escape(value: str) -> str:
        """Maskiert LIKE-Platzhalter, damit ein Präfix wörtlich gilt.

        Quell-IDs enthalten regelmäßig Bindestriche, aber `_` wäre in LIKE
        ein Platzhalter für ein beliebiges Zeichen. Passend dazu setzen alle
        Abfragen, die diesen Helfer nutzen, ``ESCAPE '\\'``.
        """
        return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

    @classmethod
    def _normalize_sources(cls, source) -> List[str]:
        """Normalisiert eine --source-Angabe zu einer Liste von Quell-IDs.

        Akzeptiert 'usmc-working', 'observed/usmc-working', 'a,b' oder
        ['a', 'b']; ein vorangestelltes 'observed/' und Schrägstriche am Rand
        werden abgeschnitten, damit beide Schreibweisen dasselbe treffen.
        """
        if source is None:
            return []
        raw = source.split(",") if isinstance(source, str) else list(source)
        ids = []
        for item in raw:
            sid = str(item).strip().strip("/")
            if sid.lower().startswith("observed/"):
                sid = sid[len("observed/"):]
            sid = sid.strip("/")
            if sid:
                ids.append(sid)
        return ids

    @classmethod
    def _source_filter(cls, source, column: str = "e.name"):
        """Baut eine WHERE-Bedingung auf den Namensraum einer Quelle.

        Beobachtete Einträge heißen ``observed/<quell-id>/<rest>``; gefiltert
        wird deshalb auf genau dieses Präfix (plus den Eintrag der Quelle
        selbst, falls es keinen Rest gibt). Mehrere Quellen werden ODER-verknüpft.

        Returns:
            (sql_fragment, params) oder (None, []) wenn nichts zu filtern ist
        """
        ids = cls._normalize_sources(source)
        if not ids:
            return None, []
        parts, params = [], []
        for sid in ids:
            parts.append(f"({column} LIKE ? ESCAPE '\\' OR {column} = ?)")
            params.extend([cls._like_escape(f"observed/{sid}/") + "%", f"observed/{sid}"])
        return "(" + " OR ".join(parts) + ")", params

    def _fts_query(self, conn: sqlite3.Connection, match_query: str,
                   type: Optional[str] = None, limit: int = 20,
                   with_snippets: bool = False,
                   source=None) -> List[Dict]:
        """Führt FTS5-Suche über beide Datenbanken aus."""
        results = []
        # snippet()/rank und die MATCH-Spalte brauchen den unqualifizierten
        # FTS-Tabellennamen (SQLite löst Auxiliary-Functions sonst nicht
        # auf); pro Schleifendurchlauf ist genau eine everything_fts im
        # Query, daher ist der Name eindeutig.
        snippet_sel = (
            ", snippet(everything_fts, 1, '>>>', '<<<', ' … ', 16) AS snippet"
            if with_snippets else ""
        )
        for db_prefix, db_label in [("main", "user"), ("other", "system")]:
            sql = f"""
                SELECT e.*, fts.rank AS rank, '{db_label}' as source{snippet_sel}
                FROM {db_prefix}.everything e
                JOIN {db_prefix}.everything_fts fts ON e.id = fts.rowid
                WHERE everything_fts MATCH ?
            """
            params = [match_query]

            if type:
                sql += " AND e.type = ?"
                params.append(type)

            src_sql, src_params = self._source_filter(source)
            if src_sql:
                sql += f" AND {src_sql}"
                params.extend(src_params)

            sql += " ORDER BY rank LIMIT ?"
            params.append(limit)

            rows = conn.execute(sql, params).fetchall()
            for row in rows:
                results.append(self._row_to_dict(row))
        return results

    def _like_query(self, conn: sqlite3.Connection, query: str,
                    type: Optional[str] = None, limit: int = 20,
                    source=None) -> List[Dict]:
        """Fallback-LIKE-Suche über beide Datenbanken."""
        results = []
        tokens = [t.strip() for t in query.split() if t.strip()]
        for db_prefix, db_label in [("main", "user"), ("other", "system")]:
            sql = f"""
                SELECT e.*, '{db_label}' as source
                FROM {db_prefix}.everything e
                WHERE (e.name LIKE ? OR e.content LIKE ? OR e.tags LIKE ?)
            """
            like = f"%{query}%"
            params = [like, like, like]

            if type:
                sql += " AND e.type = ?"
                params.append(type)

            src_sql, src_params = self._source_filter(source)
            if src_sql:
                sql += f" AND {src_sql}"
                params.extend(src_params)

            sql += " LIMIT ?"
            params.append(limit)

            rows = conn.execute(sql, params).fetchall()
            for row in rows:
                results.append(self._row_to_dict(row))

        if not results and len(tokens) > 1:
            for db_prefix, db_label in [("main", "user"), ("other", "system")]:
                conditions = []
                params = []
                for t in tokens:
                    l = f"%{t}%"
                    conditions.append("(e.name LIKE ? OR e.content LIKE ? OR e.tags LIKE ?)")
                    params.extend([l, l, l])

                sql = f"""
                    SELECT e.*, '{db_label}' as source
                    FROM {db_prefix}.everything e
                    WHERE ({' OR '.join(conditions)})
                """
                if type:
                    sql += " AND e.type = ?"
                    params.append(type)

                src_sql, src_params = self._source_filter(source)
                if src_sql:
                    sql += f" AND {src_sql}"
                    params.extend(src_params)

                sql += " LIMIT ?"
                params.append(limit)

                rows = conn.execute(sql, params).fetchall()
                for row in rows:
                    results.append(self._row_to_dict(row))

        return results

    # ------------------------------------------------------------------
    # API: find()
    # ------------------------------------------------------------------

    def _source_listing(self, conn: sqlite3.Connection, source,
                        type: Optional[str] = None, limit: int = 20) -> List[Dict]:
        """Listet eine Quelle ohne Volltextsuche (neueste zuerst).

        Für den Fall `find(query="", source="usmc-working")`: FTS5 braucht
        einen Suchbegriff, "zeig mir alles aus dieser Quelle" hat aber keinen.
        """
        src_sql, src_params = self._source_filter(source)
        if not src_sql:
            return []
        results = []
        for db_prefix, db_label in [("main", "user"), ("other", "system")]:
            sql = f"""
                SELECT e.*, '{db_label}' as source
                FROM {db_prefix}.everything e
                WHERE {src_sql}
            """
            params = list(src_params)
            if type:
                sql += " AND e.type = ?"
                params.append(type)
            sql += " ORDER BY e.updated DESC LIMIT ?"
            params.append(limit)

            for row in conn.execute(sql, params).fetchall():
                results.append(self._row_to_dict(row))
        return results

    def find(self, query: str, type: Optional[str] = None,
             limit: int = 20, with_snippets: bool = False,
             source=None) -> List[Dict]:
        """Durchsucht beide Datenbanken. Der primäre Zugang zu allem.

        Bei Mehrwort-Suchanfragen (z. B. 'Registry Mitgliedschaft') wird zunächst
        eine exakte/AND-Suche ausgeführt. Führt diese zu 0 Treffern, erfolgt
        automatisch ein Fallback auf eine OR-Verknüpfung der Einzelbegriffe mit
        FTS5-BM25-Ranking (Treffer mit allen/mehreren Begriffen stehen höher).

        Args:
            query: Suchbegriff (Volltextsuche). Leer erlaubt, wenn `source`
                gesetzt ist -- dann wird die Quelle schlicht aufgelistet.
            type: Optional filtern nach Typ (knowledge, tool, task, memory, ...)
            limit: Max. Ergebnisse
            with_snippets: Bei True enthalten FTS-Treffer ein 'snippet'-Feld
                mit Treffer-Kontext aus dem Inhalt (Marker '>>>'/'<<<',
                BACH-unified_search-Muster). LIKE-Fallback-Treffer haben
                kein Snippet.
            source: Auf eine oder mehrere observe-Quellen einschränken
                ('usmc-working' oder 'a,b'). Wirkt als WHERE-Bedingung auf den
                Namensraum `observed/<quell-id>/...`, also VOR dem Ranking:
                ohne das verdrängen die grossen Transkript-Quellen per BM25
                jede Fachsuche. NICHT zu verwechseln mit dem Feld 'source' im
                Ergebnis -- das benennt die Datenbank ('user'/'system').

        Returns:
            Liste von Einträgen als Dicts
        """
        conn = self._conn("user")
        results = []
        # Nur-Quelle-Auflistung: leere Query UND gesetzte Quelle. Ohne Quelle
        # bleibt die leere Query wie bisher der LIKE-Stufe überlassen (dort
        # matcht '%%' alles) -- das ist bestehendes Verhalten und ändert sich hier nicht.
        source_only = (not query or not query.strip()) and bool(self._normalize_sources(source))
        try:
            if source_only:
                results = self._source_listing(conn, source, type=type, limit=limit)
            else:
                # 1. Exakte / Standard-FTS5-Suche
                try:
                    results = self._fts_query(conn, query, type=type, limit=limit,
                                              with_snippets=with_snippets,
                                              source=source)
                except Exception:
                    results = []

                # 2. Mehrwort-Fallback: Wenn 0 Treffer und Mehrwort-Query, OR-Verknüpfung in FTS5 versuchen
                if not results:
                    or_query = self._build_fts_or_query(query)
                    if or_query:
                        try:
                            results = self._fts_query(conn, or_query, type=type, limit=limit,
                                                      with_snippets=with_snippets,
                                                      source=source)
                        except Exception:
                            results = []

                # 3. Fallback auf LIKE-Suche, falls FTS (auch OR) fehlschlug oder 0 Treffer ergab
                if not results:
                    try:
                        results = self._like_query(conn, query, type=type, limit=limit,
                                                   source=source)
                    except Exception:
                        results = []
        finally:
            conn.close()

        # Deduplizieren nach (id, source)
        seen = set()
        unique_results = []
        for r in results:
            key = (r.get("id"), r.get("source"))
            if key not in seen:
                seen.add(key)
                unique_results.append(r)
        results = unique_results

        # Nach Relevanz sortieren: pinned zuerst, dann FTS-Rank (bm25,
        # kleiner = relevanter); LIKE-Fallback-Treffer (ohne Rank) werden
        # untereinander neueste zuerst sortiert.
        results.sort(key=lambda x: x.get("updated", ""), reverse=True)
        results.sort(key=lambda x: (
            -x.get("pinned", 0),
            x.get("rank", 0.0)
        ))
        for r in results:
            r.pop("rank", None)

        return results[:limit]

    # ------------------------------------------------------------------
    # API: get()
    # ------------------------------------------------------------------

    def get(self, name: str) -> Optional[Dict]:
        """Holt einen einzelnen Eintrag nach Name.

        Sucht zuerst in user.db, dann in gardener.db.
        """
        conn = self._conn("user")
        try:
            for db_prefix, db_label in [("main", "user"), ("other", "system")]:
                row = conn.execute(
                    f"SELECT *, '{db_label}' as source FROM {db_prefix}.everything WHERE name = ?",
                    (name,)
                ).fetchone()
                if row:
                    return self._row_to_dict(row)
            return None
        finally:
            conn.close()

    # ------------------------------------------------------------------
    # API: put()
    # ------------------------------------------------------------------

    def put(self, name: str, content: str = "", type: str = "memory",
            tags: str = "", meta: Optional[Dict] = None,
            pinned: bool = False, target: str = "auto") -> Dict:
        """Schreibt oder aktualisiert einen Eintrag.

        Args:
            name: Eindeutiger Name
            content: Inhalt (Markdown, Text, Code)
            type: Typ (knowledge, tool, task, memory, config, document, export)
            tags: Komma-separierte Tags
            meta: Zusätzliche strukturierte Daten (JSON)
            pinned: Fest gespeichert (überlebt Sync)
            target: 'user', 'system', oder 'auto' (auto = user für die meisten)

        Returns:
            Der geschriebene Eintrag als Dict
        """
        now = self._now()
        meta_json = json.dumps(meta or {}, ensure_ascii=False)

        # Auto-Target: system nur für knowledge und tool
        if target == "auto":
            target = "system" if type in ("knowledge", "tool") else "user"

        conn = self._conn(target)
        db = "main"  # Immer in die primäre DB schreiben

        try:
            # Upsert
            existing = conn.execute(
                f"SELECT id FROM {db}.everything WHERE name = ?", (name,)
            ).fetchone()

            if existing:
                conn.execute(f"""
                    UPDATE {db}.everything
                    SET content = ?, type = ?, tags = ?, meta = ?,
                        pinned = ?, updated = ?
                    WHERE name = ?
                """, (content, type, tags, meta_json, int(pinned), now, name))
            else:
                conn.execute(f"""
                    INSERT INTO {db}.everything
                        (name, content, type, tags, meta, pinned, created, updated)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """, (name, content, type, tags, meta_json, int(pinned), now, now))

            conn.commit()
        finally:
            conn.close()

        return self.get(name)

    # ------------------------------------------------------------------
    # API: run()
    # ------------------------------------------------------------------

    def run(self, name: str, input: Optional[Dict] = None) -> Tuple[bool, str]:
        """Führt den Code-Block eines Eintrags aus.

        Materialisiert den Code in einen Workspace, führt ihn aus,
        gibt das Ergebnis zurück.

        Args:
            name: Name des Eintrags (muss type='tool' sein)
            input: Parameter für die Ausführung

        Returns:
            (success, output) Tuple
        """
        entry = self.get(name)
        if not entry:
            return False, f"Eintrag '{name}' nicht gefunden."

        # Code-Block aus Content extrahieren
        code = self._extract_code(entry["content"])
        if not code:
            return False, f"Kein ausführbarer Code-Block in '{name}' gefunden."

        # Workspace materialisieren
        ws_dir = self.workspace_dir / name.replace("/", "_")
        ws_dir.mkdir(parents=True, exist_ok=True)
        script_path = ws_dir / "run.py"

        # Runner-Script erstellen
        runner = self._build_runner(code, input or {})
        script_path.write_text(runner, encoding="utf-8")

        # Ausfuehren
        try:
            result = subprocess.run(
                [sys.executable, str(script_path)],
                capture_output=True,
                text=True,
                timeout=60,
                cwd=str(ws_dir),
                env={**os.environ, "PYTHONIOENCODING": "utf-8"},
            )

            output = result.stdout
            if result.returncode != 0:
                output += f"\n[STDERR] {result.stderr}" if result.stderr else ""
                return False, output.strip()

            return True, output.strip()

        except subprocess.TimeoutExpired:
            return False, f"Timeout: '{name}' hat länger als 60s gedauert."
        except Exception as e:
            return False, f"Fehler bei Ausführung von '{name}': {e}"

    # ------------------------------------------------------------------
    # Erweiterte Operationen
    # ------------------------------------------------------------------

    def absorb(self, file_path: Union[str, Path]) -> Dict:
        """Absorbiert eine Datei in die DB (Transporter: Materialisiert → DB).

        Kleine Dateien: Inhalt direkt in DB.
        Große Dateien: Index in DB + Datei auf Halde.
        """
        file_path = Path(file_path)
        if not file_path.is_file():
            raise FileNotFoundError(
                f"Keine reguläre Datei (fehlt oder ist ein Ordner): {file_path}")

        size = file_path.stat().st_size
        name = file_path.name
        content = ""
        meta = {
            "original_path": str(file_path),
            "original_name": name,
            "size": size,
            "absorbed": True,
            "absorbed_at": self._now(),
        }

        # Text extrahieren (wenn moeglich)
        try:
            if file_path.suffix.lower() in ('.txt', '.md', '.py', '.json', '.csv',
                                             '.yaml', '.yml', '.xml', '.html', '.css',
                                             '.js', '.ts', '.sh', '.bat', '.sql', '.log'):
                content = file_path.read_text(encoding="utf-8", errors="replace")
                meta["mimetype"] = "text/plain"
            else:
                meta["mimetype"] = self._guess_mimetype(file_path.suffix)
        except Exception:
            meta["mimetype"] = "application/octet-stream"

        # Größe entscheidet über Speicherort
        if size > BLOB_THRESHOLD_WARN:
            # Halde: nur Index + Datei kopieren
            blob_hash = self._hash_file(file_path)
            blob_dest = self.blob_dir / f"{blob_hash}{file_path.suffix}"
            shutil.copy2(str(file_path), str(blob_dest))
            meta["blob_path"] = str(blob_dest)
            meta["blob_hash"] = blob_hash
            meta["storage"] = "halde"
        elif size > BLOB_THRESHOLD_DIRECT:
            # In DB als BLOB (über blobs-Tabelle)
            meta["storage"] = "blob"
            blob_hash = self._hash_file(file_path)
            blob_dest = self.blob_dir / f"{blob_hash}{file_path.suffix}"
            shutil.copy2(str(file_path), str(blob_dest))
            meta["blob_path"] = str(blob_dest)
            meta["blob_hash"] = blob_hash
        else:
            # Klein genug: direkt in content (wenn Text) oder blob
            meta["storage"] = "inline"
            if not content:
                blob_hash = self._hash_file(file_path)
                blob_dest = self.blob_dir / f"{blob_hash}{file_path.suffix}"
                shutil.copy2(str(file_path), str(blob_dest))
                meta["blob_path"] = str(blob_dest)
                meta["blob_hash"] = blob_hash

        # Tags aus Dateiendung
        tags = f"datei,{file_path.suffix.lstrip('.')}"

        return self.put(name, content=content, type="document",
                        tags=tags, meta=meta, target="user", pinned=True)

    def materialize(self, name: str, dest: Optional[Path] = None) -> Optional[Path]:
        """Materialisiert einen DB-Eintrag als Datei (Transporter: DB → Datei).

        Args:
            name: Name des Eintrags
            dest: Zielordner (Default: ~/gardener/export/)

        Returns:
            Pfad zur materialisierten Datei oder None
        """
        entry = self.get(name)
        if not entry:
            return None

        dest = dest or self.output_dir
        dest.mkdir(parents=True, exist_ok=True)

        meta = entry.get("meta", {})
        if isinstance(meta, str):
            meta = json.loads(meta)

        # Blob von Halde?
        blob_path = meta.get("blob_path")
        if blob_path and Path(blob_path).exists():
            original_name = self._safe_filename(
                meta.get("original_name") or name, "blob.bin")
            out_path = dest / original_name
            shutil.copy2(blob_path, str(out_path))
            return out_path

        # Text-Content als Datei schreiben
        suffix = meta.get("format", "md")
        filename = self._safe_filename(
            meta.get("filename") or f"{name}.{suffix}", f"export.{suffix}")
        out_path = dest / filename
        out_path.write_text(entry["content"], encoding="utf-8")
        return out_path

    def observe(self, directory: Optional[Path] = None) -> List[Dict]:
        """Beobachtet den Ordner und aktualisiert die DB (Blick aus dem Fenster).

        Scannt den Home-Ordner und erstellt/aktualisiert observed-Einträge.

        Returns:
            Liste der beobachteten Dateien
        """
        directory = directory or self.home
        observed = []

        for file_path in directory.rglob("*"):
            if not file_path.is_file():
                continue
            # Interne Ordner überspringen
            rel = file_path.relative_to(directory)
            if self._is_internal(str(rel)):
                continue

            # as_posix(): plattformstabile Entry-Namen (Windows-Backslashes
            # würden bei Cross-System-Sync Duplikate statt Updates erzeugen)
            name = f"observed/{rel.as_posix()}"
            content = ""

            # Text extrahieren wenn moeglich
            if file_path.suffix.lower() in ('.txt', '.md', '.py', '.json', '.csv',
                                             '.yaml', '.yml', '.xml', '.html'):
                try:
                    content = file_path.read_text(encoding="utf-8", errors="replace")
                except Exception:
                    content = f"[Datei nicht lesbar: {file_path.suffix}]"
            else:
                content = f"[Binärdatei: {file_path.name}, {file_path.stat().st_size} Bytes]"

            meta = {
                "path": str(file_path),
                "size": file_path.stat().st_size,
                "modified": datetime.fromtimestamp(
                    file_path.stat().st_mtime
                ).isoformat(timespec="seconds"),
                "observed": True,
            }

            entry = self.put(name, content=content, type="observed",
                             tags=f"datei,{file_path.suffix.lstrip('.')}",
                             meta=meta, target="user")
            observed.append(entry)

        return observed

    def sync(self) -> Dict:
        """Führt einen Sync-Zyklus durch: Absorber leeren + Ordner beobachten.

        1. .absorber/ → Dateien absorbieren und aus Ordner entfernen
        2. Rest des Home-Ordners → beobachten (nur Text lesen)
        3. Falls mode='always_absorb' → alles absorbieren

        Returns:
            {"absorbed": int, "observed": int, "mode": str}
        """
        mode = self.config.get("mode", "selective")
        absorbed_count = 0
        observed_count = 0

        # --- 1. Absorber leeren ---
        if self.absorber_dir.exists():
            for file_path in list(self.absorber_dir.rglob("*")):
                if not file_path.is_file():
                    continue
                try:
                    self.absorb(file_path)
                    file_path.unlink()  # Datei verschwindet nach Absorption
                    absorbed_count += 1
                except Exception as e:
                    # Fehler loggen aber weitermachen
                    self.put(f"sync-error/{file_path.name}",
                             content=f"Absorb fehlgeschlagen: {e}",
                             type="memory", tags="error,sync")

        # --- 2. Ordner beobachten oder absorbieren ---
        for file_path in self.home.rglob("*"):
            if not file_path.is_file():
                continue

            # Interne Ordner überspringen
            rel = file_path.relative_to(self.home)
            if self._is_internal(str(rel)):
                continue

            if mode == "always_absorb":
                try:
                    self.absorb(file_path)
                    file_path.unlink()
                    absorbed_count += 1
                except Exception:
                    pass
            else:
                # Beobachten (nur Text extrahieren, Datei nicht anfassen)
                name = f"observed/{rel.as_posix()}"
                content = ""

                if file_path.suffix.lower() in ('.txt', '.md', '.py', '.json', '.csv',
                                                 '.yaml', '.yml', '.xml', '.html'):
                    try:
                        content = file_path.read_text(encoding="utf-8", errors="replace")
                    except Exception:
                        content = f"[Nicht lesbar: {file_path.suffix}]"
                else:
                    content = f"[Binärdatei: {file_path.name}, {file_path.stat().st_size} Bytes]"

                meta = {
                    "path": str(file_path),
                    "size": file_path.stat().st_size,
                    "modified": datetime.fromtimestamp(
                        file_path.stat().st_mtime
                    ).isoformat(timespec="seconds"),
                    "observed": True,
                }

                self.put(name, content=content, type="observed",
                         tags=f"datei,{file_path.suffix.lstrip('.')}",
                         meta=meta, target="user")
                observed_count += 1

        return {"absorbed": absorbed_count, "observed": observed_count, "mode": mode}

    # ------------------------------------------------------------------
    # Task-Komfort (Tasks = Einträge vom Typ 'task' in user.db)
    # ------------------------------------------------------------------

    def tasks(self, status: Optional[str] = None) -> List[Dict]:
        """Listet alle Tasks. Optional nach Status filtern.

        Status-Werte: open, doing, done, blocked, waiting
        """
        conn = self._conn("user")
        try:
            sql = "SELECT *, 'user' as source FROM main.everything WHERE type = 'task'"
            params = []

            if status:
                sql += " AND json_extract(meta, '$.status') = ?"
                params.append(status)

            # Semantic priority order (critical > high > normal > low),
            # not alphabetical string order
            sql += """
                ORDER BY CASE json_extract(meta, '$.priority')
                    WHEN 'critical' THEN 4
                    WHEN 'high' THEN 3
                    WHEN 'normal' THEN 2
                    WHEN 'low' THEN 1
                    ELSE 2
                END DESC, updated DESC
            """
            rows = conn.execute(sql, params).fetchall()
        finally:
            conn.close()
        return [self._row_to_dict(row) for row in rows]

    def task(self, name: str, content: str = "", priority: str = "normal",
             due: Optional[str] = None, tags: str = "") -> Dict:
        """Erstellt oder aktualisiert einen Task.

        Args:
            name: Task-Name (wird zum Key)
            content: Beschreibung
            priority: low, normal, high, critical
            due: Faelligkeitsdatum (ISO)
            tags: Komma-separierte Tags
        """
        meta = {"status": "open", "priority": priority}
        if due:
            meta["due"] = due

        return self.put(name, content=content, type="task",
                        tags=f"task,{tags}" if tags else "task",
                        meta=meta, target="user")

    def task_done(self, name: str) -> Optional[Dict]:
        """Markiert einen Task als erledigt."""
        entry = self.get(name)
        if not entry or entry.get("type") != "task":
            return None
        meta = entry.get("meta", {})
        if isinstance(meta, str):
            meta = json.loads(meta)
        meta["status"] = "done"
        meta["done_at"] = self._now()
        return self.put(name, content=entry["content"], type="task",
                        tags=entry.get("tags", ""), meta=meta, target="user")

    def task_status(self, name: str, status: str) -> Optional[Dict]:
        """Setzt den Status eines Tasks (open, doing, done, blocked, waiting)."""
        entry = self.get(name)
        if not entry or entry.get("type") != "task":
            return None
        meta = entry.get("meta", {})
        if isinstance(meta, str):
            meta = json.loads(meta)
        meta["status"] = status
        return self.put(name, content=entry["content"], type="task",
                        tags=entry.get("tags", ""), meta=meta, target="user")

    # ------------------------------------------------------------------
    # Memory & Lernen (alles in everything, keine Extra-Tabellen)
    # ------------------------------------------------------------------

    def memo(self, text: str, tags: str = "") -> Dict:
        """Schreibt ins Arbeitsgedächtnis (Working Memory).

        Kurzlebige Notizen für die aktuelle Session.
        Werden bei Konsolidierung verdichtet oder vergessen.
        """
        name = f"memo/{self._now().replace(':', '-')}"
        return self.put(name, content=text, type="memory",
                        tags=f"working,{tags}" if tags else "working",
                        meta={"weight": 0.5, "accessed": 0,
                              "session": self._now()[:10]},
                        target="user")

    def lesson(self, title: str, content: str = "",
               severity: str = "medium", tags: str = "") -> Dict:
        """Speichert eine Lektion (Best Practice / Lesson Learned).

        Lessons haben hohes Gewicht und verfallen langsamer.
        severity: low, medium, high, critical
        """
        weight_map = {"low": 0.5, "medium": 0.7, "high": 0.9, "critical": 1.0}
        name = f"lesson/{title.lower().replace(' ', '-')[:50]}"
        return self.put(name, content=content or title, type="lesson",
                        tags=f"lesson,{severity},{tags}" if tags else f"lesson,{severity}",
                        meta={"weight": weight_map.get(severity, 0.7),
                              "severity": severity, "accessed": 0,
                              "decay_rate": 0.99},  # Lessons verfallen langsam
                        target="user")

    def session_end(self, summary: str) -> Dict:
        """Speichert einen Session-Bericht (Episodisches Gedächtnis).

        Format empfohlen: "THEMA: Was. ERLEDIGT: Was. NAECHSTE: Was."
        """
        name = f"session/{self._now()[:10]}"
        return self.put(name, content=summary, type="session",
                        tags="session,episodisch",
                        meta={"weight": 0.8, "accessed": 0,
                              "decay_rate": 0.97},
                        target="user")

    def recall(self, query: str, limit: int = 5) -> List[Dict]:
        """Erinnert sich -- sucht in Memory, Lessons und Sessions.

        Wie find(), aber auf Gedächtnis-Typen beschränkt und
        erhöht das Gewicht abgerufener Einträge (Boost).
        """
        conn = self._conn("user")
        results = []

        try:
            for mem_type in ("memory", "lesson", "session"):
                try:
                    sql = """
                        SELECT *, 'user' as source
                        FROM main.everything e
                        JOIN main.everything_fts fts ON e.id = fts.rowid
                        WHERE everything_fts MATCH ? AND e.type = ?
                        ORDER BY rank LIMIT ?
                    """
                    rows = conn.execute(sql, (query, mem_type, limit)).fetchall()
                    for row in rows:
                        d = self._row_to_dict(row)
                        results.append(d)
                        # Boost: Gewicht erhoehen bei Abruf
                        self._boost(conn, row["id"])
                except Exception:
                    sql = """
                        SELECT *, 'user' as source
                        FROM main.everything e
                        WHERE (e.name LIKE ? OR e.content LIKE ?) AND e.type = ?
                        LIMIT ?
                    """
                    like = f"%{query}%"
                    rows = conn.execute(sql, (like, like, mem_type, limit)).fetchall()
                    for row in rows:
                        d = self._row_to_dict(row)
                        results.append(d)
                        self._boost(conn, row["id"])

            conn.commit()
        finally:
            conn.close()

        # Nach Gewicht sortieren
        results.sort(key=lambda x: x.get("meta", {}).get("weight", 0.5), reverse=True)
        return results[:limit]

    def consolidate(self) -> Dict:
        """Konsolidiert das Gedächtnis (= Schlaf).

        1. Decay: Gewicht aller Einträge reduzieren
        2. Forget: Einträge unter Schwellenwert deaktivieren
        3. Stats zurückgeben

        Einfacher als BACHs 6-Stufen-Pipeline, aber gleicher Effekt:
        Wichtiges bleibt, Unwichtiges verblasst.
        """
        conn = self._conn("user")
        now = self._now()
        stats = {"decayed": 0, "forgotten": 0, "kept": 0}

        try:
            # Alle Memory-artigen Einträge mit Gewicht.
            # Gepinnte Einträge sind von Decay und Forget ausgenommen.
            rows = conn.execute("""
                SELECT id, name, type, meta FROM main.everything
                WHERE type IN ('memory', 'lesson', 'session')
                  AND pinned = 0
            """).fetchall()

            for row in rows:
                meta = row["meta"]
                if isinstance(meta, str):
                    try:
                        meta = json.loads(meta)
                    except (json.JSONDecodeError, TypeError):
                        meta = {}

                weight = meta.get("weight", 0.5)
                decay_rate = meta.get("decay_rate", 0.95)

                # Decay anwenden
                new_weight = weight * decay_rate
                meta["weight"] = round(new_weight, 4)
                meta["last_decay"] = now

                if new_weight < 0.05:
                    # Vergessen: Eintrag löschen (unter Schwelle)
                    conn.execute("DELETE FROM main.everything WHERE id = ?", (row["id"],))
                    stats["forgotten"] += 1
                else:
                    # Gewicht aktualisieren
                    conn.execute(
                        "UPDATE main.everything SET meta = ? WHERE id = ?",
                        (json.dumps(meta, ensure_ascii=False), row["id"])
                    )
                    if new_weight < weight:
                        stats["decayed"] += 1
                    else:
                        stats["kept"] += 1

            conn.commit()
        finally:
            conn.close()
        return stats

    def _boost(self, conn: sqlite3.Connection, entry_id: int, amount: float = 0.1):
        """Erhoeht das Gewicht eines Eintrags (Boost bei Abruf)."""
        row = conn.execute(
            "SELECT meta FROM main.everything WHERE id = ?", (entry_id,)
        ).fetchone()
        if not row:
            return

        meta = row["meta"]
        if isinstance(meta, str):
            try:
                meta = json.loads(meta)
            except (json.JSONDecodeError, TypeError):
                meta = {}

        weight = min(1.0, meta.get("weight", 0.5) + amount)
        accessed = meta.get("accessed", 0) + 1
        meta["weight"] = round(weight, 4)
        meta["accessed"] = accessed
        meta["last_accessed"] = self._now()

        conn.execute(
            "UPDATE main.everything SET meta = ? WHERE id = ?",
            (json.dumps(meta, ensure_ascii=False), entry_id)
        )

    # ------------------------------------------------------------------
    # Status
    # ------------------------------------------------------------------

    def delete(self, name: str) -> bool:
        """Löscht einen Eintrag aus der DB.

        Sucht in user.db, dann in system.db.
        Returns: True wenn gelöscht, False wenn nicht gefunden.
        """
        for target in ("user", "system"):
            conn = self._conn(target)
            try:
                row = conn.execute(
                    "SELECT id FROM main.everything WHERE name = ?", (name,)
                ).fetchone()
                if row:
                    conn.execute("DELETE FROM main.everything WHERE id = ?", (row["id"],))
                    conn.commit()
                    return True
            finally:
                conn.close()
        return False

    def list(self, type: Optional[str] = None, limit: int = 50) -> List[Dict]:
        """Listet alle Einträge. Optional nach Typ filtern.

        Ohne Suchbegriff -- einfach alles zeigen.
        """
        conn = self._conn("user")
        results = []

        try:
            for db_prefix, db_label in [("main", "user"), ("other", "system")]:
                sql = f"SELECT *, '{db_label}' as source FROM {db_prefix}.everything"
                params = []

                if type:
                    sql += " WHERE type = ?"
                    params.append(type)

                sql += " ORDER BY updated DESC LIMIT ?"
                params.append(limit)

                rows = conn.execute(sql, params).fetchall()
                for row in rows:
                    results.append(self._row_to_dict(row))
        finally:
            conn.close()

        return results[:limit]

    def status(self) -> Dict:
        """Gibt den Status des Systems zurück."""
        info = {
            "home": str(self.home),
            "data_dir": str(self.data_dir),
            "system_db": str(self.system_db_path),
            "user_db": str(self.user_db_path),
        }

        # Counts
        conn = self._conn("user")
        try:
            for db_prefix, db_label in [("main", "user"), ("other", "system")]:
                count = conn.execute(
                    f"SELECT COUNT(*) FROM {db_prefix}.everything"
                ).fetchone()[0]
                info[f"{db_label}_entries"] = count

                # Nach Typ
                rows = conn.execute(
                    f"SELECT type, COUNT(*) as cnt FROM {db_prefix}.everything GROUP BY type"
                ).fetchall()
                info[f"{db_label}_types"] = {row["type"]: row["cnt"] for row in rows}
        finally:
            conn.close()

        # Blob-Halde
        blob_count = len(list(self.blob_dir.glob("*")))
        blob_size = sum(f.stat().st_size for f in self.blob_dir.glob("*") if f.is_file())
        info["blobs"] = {"count": blob_count, "size_mb": round(blob_size / 1_000_000, 1)}

        return info

    # ------------------------------------------------------------------
    # Config
    # ------------------------------------------------------------------

    def _load_config(self) -> Dict:
        """Laedt config.json aus dem Home-Ordner."""
        config_path = self.home / "config.json"
        if config_path.exists():
            try:
                return json.loads(config_path.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError):
                pass

        # Default-Config erstellen
        default = {
            "mode": "selective",
            "_modes": {
                "selective": "Nur .absorber/ wird absorbiert (Default)",
                "always_absorb": "Alles im Home-Ordner wird absorbiert",
                "observe_only": "Nichts absorbieren, nur beobachten"
            }
        }
        config_path.write_text(
            json.dumps(default, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
        return default

    def _save_config(self) -> None:
        """Schreibt self.config zurueck nach config.json im Home-Ordner."""
        config_path = self.home / "config.json"
        config_path.write_text(
            json.dumps(self.config, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )

    # ------------------------------------------------------------------
    # Cross-Source federated index (observe-sources)
    # ------------------------------------------------------------------
    #
    # Read-only adapters (siehe sources.py) die Text aus fremden
    # Wissensquellen in die eigene FTS-Suche einspeisen -- foederiert,
    # nicht absorbiert: Originale bleiben, wo sie sind, es wird nur
    # indexiert. Konfiguration liegt in config.json unter
    # "observe_sources" (source_id -> {kind, ...}); Scan-Fortschritt
    # (fuer inkrementelles Tailing grosser JSONL-Transkripte) liegt in
    # einer separaten Laufzeitdatei in data_dir, nicht in der DB.

    # ------------------------------------------------------------------
    # Cloud credential alert
    # ------------------------------------------------------------------

    @staticmethod
    def _is_cloud_path(path) -> bool:
        """True if `path` lies inside the cloud-synced root."""
        if not path:
            return False
        try:
            Path(path).resolve().relative_to(CLOUD_ALERT_ROOT.resolve())
            return True
        except (ValueError, OSError):
            return False

    @staticmethod
    def record_cloud_alerts(findings) -> int:
        """Appends new (path, family) findings to the alert file.

        `findings` is an iterable of (source_path, families). Only the path
        and the family name are ever written -- never the value and never
        the surrounding text, because the alert file itself lives in the
        cloud folder it is warning about.

        Idempotent: a finding already listed is not appended again, so a
        nightly refresh does not grow the file without new information.
        Returns the number of newly written lines.
        """
        pairs = sorted({(str(p), f) for p, fams in findings for f in fams})
        if not pairs:
            return 0

        path = CLOUD_ALERT_FILE
        try:
            existing = path.read_text(encoding="utf-8") if path.exists() else ""
        except OSError:
            existing = ""

        new = [(p, f) for p, f in pairs if f"`{p}` | {f}" not in existing]
        if not new:
            return 0

        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            with open(path, "a", encoding="utf-8") as fh:
                if not existing:
                    fh.write(
                        "# SICHERHEITSBEFUND: Token-Signatur in "
                        "Cloud-Dokument\n\n"
                        "Zugangsdaten gehoeren nicht in einen Cloud-Ordner. "
                        "Jede Zeile ist ein Fund beim Indizieren.\n"
                        "Es wird **nur der Fundort und die Muster-Familie** "
                        "notiert -- nie der Wert, nie der umgebende Text.\n\n"
                        "| Datum | Quellpfad | Muster-Familie |\n"
                        "|---|---|---|\n")
                stamp = datetime.now().isoformat(timespec="seconds")
                for p, f in new:
                    fh.write(f"| {stamp} | `{p}` | {f} |\n")
        except OSError:
            return 0
        return len(new)

    def _observe_source_state_path(self) -> Path:
        return self.data_dir / "observe_sources_state.json"

    def _load_observe_source_state(self) -> Dict:
        path = self._observe_source_state_path()
        if path.exists():
            try:
                return json.loads(path.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError):
                pass
        return {}

    def _save_observe_source_state(self, state: Dict) -> None:
        path = self._observe_source_state_path()
        path.write_text(
            json.dumps(state, indent=2, ensure_ascii=False),
            encoding="utf-8",
        )

    def observe_source_add(self, source_id: str, kind: str, **params) -> Dict:
        """Registriert eine neue observe-source (nur Konfiguration, kein Scan).

        Args:
            source_id: eindeutiger Schluessel (wird auch Namenspraefix
                der Eintraege: 'observed/<source_id>/...')
            kind: einer von sources.VALID_KINDS (markdown_dir,
                remember_files, sqlite_table, agent_transcripts)
            **params: kind-spezifische Konfiguration (path, glob,
                patterns, db_path, table, columns, format, ...) --
                siehe sources.py Modul-Docstring.
        """
        import sources
        if kind not in sources.VALID_KINDS:
            raise ValueError(
                f"Unbekannte observe-source kind: {kind!r} "
                f"(gueltig: {', '.join(sources.VALID_KINDS)})"
            )
        entry = {"kind": kind, "enabled": True, **params}
        self.config.setdefault("observe_sources", {})[source_id] = entry
        self._save_config()
        return entry

    def observe_source_remove(self, source_id: str) -> bool:
        """Entfernt eine observe-source-Konfiguration (Config, nicht die
        bereits indexierten Eintraege -- die bleiben bis delete()/consolidate()).
        """
        sources_cfg = self.config.get("observe_sources", {})
        if source_id not in sources_cfg:
            return False
        del sources_cfg[source_id]
        self._save_config()

        state = self._load_observe_source_state()
        if source_id in state:
            del state[source_id]
            self._save_observe_source_state(state)
        return True

    def observe_source_list(self) -> Dict:
        """Gibt die konfigurierten observe-sources zurueck (source_id -> config)."""
        return dict(self.config.get("observe_sources", {}))

    def observe_sources(self, source_id: Optional[str] = None) -> Dict:
        """Fuehrt die konfigurierten observe-source-Adapter aus (foederierte
        Cross-Source-Indexierung). Rein lesend gegenueber den Quellen: nur
        der eigene FTS-Index wird aktualisiert, Originale werden nie
        veraendert, verschoben oder geloescht.

        Args:
            source_id: nur diese eine Quelle aktualisieren (Default: alle)

        Returns:
            {source_id: {"kind": ..., "indexed": n, "skipped": n}} bzw.
            {"error": ...} pro Quelle bei Fehlern.
        """
        import sources

        configs = self.config.get("observe_sources", {})
        if source_id is not None:
            if source_id not in configs:
                return {"error": f"Unbekannte observe-source: {source_id}"}
            configs = {source_id: configs[source_id]}

        all_state = self._load_observe_source_state()
        stats: Dict[str, Dict] = {}

        for sid, cfg in configs.items():
            if not cfg.get("enabled", True):
                stats[sid] = {"skipped_disabled": True}
                continue

            kind = cfg.get("kind")
            file_state = all_state.setdefault(sid, {})
            indexed = 0
            skipped = 0
            # One connection for the whole source instead of three per item
            # (get + put's own + put's return-get). Each of those opened a
            # connection, ATTACHed the sibling DB and committed, which caps
            # throughput at ~20 items/s -- and a full agent-transcript
            # archive is six figures of items. Same upsert as put(), same
            # FTS (maintained by AFTER INSERT/UPDATE triggers), just batched.
            #
            # Looking the existing entry up in main (user.db) alone -- rather
            # than main-then-other, as get() does -- is safe here: an observed
            # entry is always written with target='user', so it can only ever
            # live in user.db.
            conn = self._conn("user")
            cloud_findings = []
            try:
                for item in sources.scan(sid, cfg, state=file_state):
                    if item.redacted:
                        src = (item.meta.get("source_ref") or {}).get("path")
                        if self._is_cloud_path(src):
                            cloud_findings.append((src, item.redacted))
                    row = conn.execute(
                        "SELECT id, meta FROM main.everything WHERE name = ?",
                        (item.name,)).fetchone()
                    if row is not None:
                        try:
                            prev_meta = json.loads(row["meta"] or "{}")
                        except (json.JSONDecodeError, TypeError):
                            prev_meta = {}
                        if prev_meta.get("source_fingerprint") == item.fingerprint:
                            skipped += 1
                            continue

                    meta = dict(item.meta)
                    meta.update({
                        "source_id": sid,
                        "source_kind": kind,
                        "source_fingerprint": item.fingerprint,
                        "observed": True,
                    })
                    meta_json = json.dumps(meta, ensure_ascii=False)
                    now = self._now()
                    if row is not None:
                        conn.execute("""
                            UPDATE main.everything
                            SET content = ?, type = ?, tags = ?, meta = ?,
                                pinned = ?, updated = ?
                            WHERE name = ?
                        """, (item.content, "observed", item.tags, meta_json,
                              0, now, item.name))
                    else:
                        conn.execute("""
                            INSERT INTO main.everything
                                (name, content, type, tags, meta, pinned,
                                 created, updated)
                            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                        """, (item.name, item.content, "observed", item.tags,
                              meta_json, 0, now, now))
                    indexed += 1
                    # Bounded transactions: a crash mid-scan keeps what was
                    # already committed, and the offset state is only saved
                    # after the source finishes, so the rest is simply
                    # re-read next time.
                    if indexed % 2000 == 0:
                        conn.commit()
                conn.commit()
                stats[sid] = {"kind": kind, "indexed": indexed, "skipped": skipped}
                if cloud_findings:
                    written = self.record_cloud_alerts(cloud_findings)
                    stats[sid]["cloud_alerts"] = len(cloud_findings)
                    stats[sid]["cloud_alerts_new"] = written
                    # Console warning: this is a security finding, not a
                    # statistic. stderr so it stays visible even when a
                    # sync run's stdout is piped into a log.
                    print(f"WARNUNG [{sid}]: Token-Signatur in "
                          f"{len(cloud_findings)} Cloud-Dokument(en) gefunden "
                          f"-- {written} neu in {CLOUD_ALERT_FILE}",
                          file=sys.stderr)
            except Exception as e:
                # Eine kaputte Quellen-Konfiguration darf den Refresh der
                # anderen Quellen nicht abreissen (gleiches Prinzip wie
                # sync()'s Fehlerbehandlung pro Datei).
                try:
                    # Keep what was already scanned; the per-file offset
                    # state only advances for files that finished, so the
                    # unread remainder is picked up on the next refresh.
                    conn.commit()
                except sqlite3.Error:
                    pass
                stats[sid] = {
                    "error": f"{e.__class__.__name__}: {e}",
                    "indexed": indexed, "skipped": skipped,
                }
            finally:
                conn.close()

        self._save_observe_source_state(all_state)
        return stats

    # ------------------------------------------------------------------
    # Hilfsfunktionen
    # ------------------------------------------------------------------

    @staticmethod
    def _is_internal(rel: str) -> bool:
        """True if a path relative to home points into an internal runtime dir
        or is an internal top-level file (config.json).

        Shared skip logic for observe() and sync(). Compares whole path
        segments, not string prefixes — sibling names like
        '.absorber-notes.txt' are NOT skipped.
        """
        parts = str(rel).replace("\\", "/").split("/")
        if any(p.lower() in INTERNAL_SKIP_PREFIXES for p in parts):
            return True
        if len(parts) == 1 and parts[0] in INTERNAL_SKIP_FILES:
            return True
        # Credential-bearing filenames (.npmrc, *.pem, auth.json, ...)
        return bool(_is_excluded_path and _is_excluded_path(rel))

    @staticmethod
    def _safe_filename(raw, fallback: str) -> str:
        """Reduziert einen Dateinamen aus meta-Daten auf seinen Basisnamen.

        meta-JSON ist über put() frei setzbar und damit nicht vertrauenswürdig:
        absolute Pfade, '..' und Laufwerksangaben würden materialize() sonst
        außerhalb des Zielordners schreiben lassen (Path Traversal).
        """
        name = str(raw).replace("\\", "/").rsplit("/", 1)[-1]
        name = name.rsplit(":", 1)[-1].strip()
        if name in ("", ".", ".."):
            return fallback
        return name

    def _row_to_dict(self, row: sqlite3.Row) -> Dict:
        """Konvertiert eine DB-Row in ein Dict."""
        d = dict(row)
        # meta als JSON parsen
        if "meta" in d and isinstance(d["meta"], str):
            try:
                d["meta"] = json.loads(d["meta"])
            except (json.JSONDecodeError, TypeError):
                d["meta"] = {}
        return d

    def _extract_code(self, content: str) -> Optional[str]:
        """Extrahiert den ersten ```python Code-Block aus Markdown."""
        pattern = r'```python\s*\n(.*?)```'
        match = re.search(pattern, content, re.DOTALL)
        if match:
            return match.group(1).strip()
        return None

    def _build_runner(self, code: str, input_data: Dict) -> str:
        """Baut ein ausfuehrbares Runner-Script."""
        input_json = json.dumps(input_data, ensure_ascii=False)
        return f'''# -*- coding: utf-8 -*-
# Auto-generated Gardener Runner
import json
import sys

input = json.loads({repr(input_json)})

{code}

if 'execute' in dir():
    result = execute(input)
    if result is not None:
        print(json.dumps(result, ensure_ascii=False, indent=2))
'''

    def _hash_file(self, file_path: Path) -> str:
        """Berechnet SHA256 einer Datei."""
        h = hashlib.sha256()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                h.update(chunk)
        return h.hexdigest()[:16]

    def _guess_mimetype(self, suffix: str) -> str:
        """Einfache Mimetype-Erkennung."""
        types = {
            '.pdf': 'application/pdf',
            '.doc': 'application/msword',
            '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
            '.xls': 'application/vnd.ms-excel',
            '.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
            '.png': 'image/png',
            '.jpg': 'image/jpeg',
            '.jpeg': 'image/jpeg',
            '.gif': 'image/gif',
            '.mp4': 'video/mp4',
            '.mp3': 'audio/mpeg',
            '.zip': 'application/zip',
        }
        return types.get(suffix.lower(), 'application/octet-stream')


# ---------------------------------------------------------------------------
# CLI (minimal)
# ---------------------------------------------------------------------------

def _parse_find_args(args: List[str]):
    """Trennt Optionen von Suchwörtern für `gardener find`.

    Das CLI kommt ohne argparse aus; ohne diese Trennung landete '--source'
    schlicht im Suchstring. Unterstützt '--flag wert' und '--flag=wert';
    alles ohne führende '--' gilt als Suchwort.

    Returns:
        (opts, words, error) -- error ist None, wenn alles verstanden wurde
    """
    opts = {"source": None, "type": None, "limit": 20}
    words: List[str] = []
    i = 0
    while i < len(args):
        arg = args[i]
        if arg.startswith("--"):
            if "=" in arg:
                key, value = arg[2:].split("=", 1)
                step = 1
            else:
                key, value, step = arg[2:], None, 2
                if i + 1 < len(args):
                    value = args[i + 1]
            if key not in opts:
                return opts, words, f"Unbekannte Option: {arg}"
            if value is None or value == "":
                return opts, words, f"Option --{key} braucht einen Wert."
            if key == "limit":
                try:
                    opts["limit"] = int(value)
                except ValueError:
                    return opts, words, f"Ungültiger Wert für --limit: {value}"
                if opts["limit"] < 1:
                    return opts, words, "--limit muss mindestens 1 sein."
            else:
                opts[key] = value
            i += step
        else:
            words.append(arg)
            i += 1
    return opts, words, None


def main():
    """Minimales CLI für Gardener."""
    from i18n import t
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    af = Gardener()

    if len(sys.argv) < 2:
        print(t("help.title"))
        print()
        s = af.status()
        print(f"  {t('help.home')}:      {s['home']}")
        print(f"  {t('help.data')}:     {s['data_dir']}")
        print(f"  {t('help.system_db')}: {s['system_entries']} {t('help.entries')}")
        print(f"  {t('help.user_db')}:   {s['user_entries']} {t('help.entries')}")
        print(f"  {t('help.heap')}:     {s['blobs']['count']} {t('help.files')} ({s['blobs']['size_mb']} MB)")
        print()
        print(t("help.commands"))
        commands = [
            ("gardener find <query>", "cmd.find"),
            ("gardener gui [--port N] [--no-browser]", "cmd.gui"),
            ("gardener get <name>", "cmd.get"),
            ("gardener put <name> <text>", "cmd.put"),
            ("gardener run <name>", "cmd.run"),
            ("gardener absorb <datei>", "cmd.absorb"),
            ("gardener materialize <name>", "cmd.materialize"),
            ("gardener sync", "cmd.sync"),
            ("gardener observe", "cmd.observe"),
            ("gardener observe-source add <id> <kind> [k=v...]", "cmd.observe_source_add"),
            ("gardener observe-source list", "cmd.observe_source_list"),
            ("gardener observe-source remove <id>", "cmd.observe_source_remove"),
            ("gardener observe-source refresh [id]", "cmd.observe_source_refresh"),
            ("gardener memo <text>", "cmd.memo"),
            ("gardener lesson <titel> [text]", "cmd.lesson"),
            ("gardener recall <query>", "cmd.recall"),
            ("gardener consolidate", "cmd.consolidate"),
            ("gardener session-end <text>", "cmd.session_end"),
            ("gardener tasks [status]", "cmd.tasks"),
            ("gardener task <name> [text]", "cmd.task"),
            ("gardener done <name>", "cmd.done"),
            ("gardener list [typ]", "cmd.list"),
            ("gardener delete <name>", "cmd.delete"),
            ("gardener status", "cmd.status"),
        ]
        for usage, label_key in commands:
            print(f"  {usage:<32} {t(label_key)}")
        print()
        print(f"  {t('help.find_filters')}")
        for flag, label_key in [
            ("--source <id>[,<id>]", "help.find_source"),
            ("--type <typ>", "help.find_type"),
            ("--limit <n>", "help.find_limit"),
        ]:
            print(f"    {flag:<22} {t(label_key)}")
        print(f"    {t('help.find_listing')}")
        print(f"    {t('help.find_legacy')}")
        return

    cmd = sys.argv[1]

    if cmd == "find":
        opts, words, err = _parse_find_args(sys.argv[2:])
        if err:
            print(f"  {err}")
            print("  Nutzung: gardener find [--source <id>[,<id>]] [--type <typ>] "
                  "[--limit <n>] <query>")
            return
        query = " ".join(words)
        if not query and not opts["source"]:
            print("  Nutzung: gardener find [--source <id>[,<id>]] [--type <typ>] "
                  "[--limit <n>] <query>")
            print("  Ohne Suchbegriff braucht es mindestens --source "
                  "(dann wird die Quelle aufgelistet).")
            return

        results = af.find(query, type=opts["type"], limit=opts["limit"],
                          source=opts["source"])
        for r in results:
            src = r.get("source", "?")
            print(f"  [{r['type']:10s}] {r['name']:30s} ({src})")
        if not results:
            aktiv = [f"{k}={v}" for k, v in
                     (("--source", opts["source"]), ("--type", opts["type"])) if v]
            zusatz = f" (Filter aktiv: {', '.join(aktiv)})" if aktiv else ""
            print(f"  Keine Ergebnisse.{zusatz}")

    elif cmd == "gui":
        # Such-GUI für menschliche Nutzer (lokaler Webserver, read-only)
        import search_gui
        port = search_gui.DEFAULT_PORT
        open_browser = True
        args = sys.argv[2:]
        i = 0
        while i < len(args):
            if args[i] == "--port" and i + 1 < len(args):
                try:
                    port = int(args[i + 1])
                except ValueError:
                    print(f"  Ungültiger Port: {args[i + 1]}")
                    return
                i += 2
            elif args[i] == "--no-browser":
                open_browser = False
                i += 1
            else:
                print(f"  Unbekannte Option: {args[i]}")
                return
        search_gui.run(port=port, open_browser=open_browser)

    elif cmd == "get":
        name = sys.argv[2] if len(sys.argv) > 2 else ""
        entry = af.get(name)
        if entry:
            print(f"Name:    {entry['name']}")
            print(f"Typ:     {entry['type']}")
            print(f"Tags:    {entry['tags']}")
            print(f"Source:  {entry.get('source', '?')}")
            print(f"Updated: {entry['updated']}")
            print(f"---")
            print(entry['content'][:2000])
        else:
            print(f"  Nicht gefunden: {name}")

    elif cmd == "put":
        name = sys.argv[2] if len(sys.argv) > 2 else "unnamed"
        content = " ".join(sys.argv[3:]) if len(sys.argv) > 3 else ""
        entry = af.put(name, content=content)
        print(f"  [OK] {entry['name']} ({entry['type']})")

    elif cmd == "run":
        name = sys.argv[2] if len(sys.argv) > 2 else ""
        ok, output = af.run(name)
        print(f"  [{'OK' if ok else 'FEHLER'}] {output}")

    elif cmd == "absorb":
        path = sys.argv[2] if len(sys.argv) > 2 else ""
        try:
            entry = af.absorb(path)
        except (FileNotFoundError, IsADirectoryError, PermissionError) as e:
            print(f"  [FEHLER] Datei nicht lesbar: {path} ({e.__class__.__name__})")
            return
        print(f"  [OK] {entry['name']} absorbiert ({entry['meta'].get('storage', '?')})")

    elif cmd == "materialize":
        name = sys.argv[2] if len(sys.argv) > 2 else ""
        result = af.materialize(name)
        if result:
            print(f"  [OK] Materialisiert: {result}")
        else:
            print(f"  Nicht gefunden: {name}")

    elif cmd == "observe":
        observed = af.observe()
        print(f"  {len(observed)} Dateien beobachtet")
        for o in observed[:10]:
            print(f"    {o['name']}")

    elif cmd == "sync":
        result = af.sync()
        print(f"  Modus: {result['mode']}")
        print(f"  Absorbiert: {result['absorbed']}")
        print(f"  Beobachtet: {result['observed']}")

    elif cmd == "observe-source":
        sub = sys.argv[2] if len(sys.argv) > 2 else ""

        if sub == "add":
            if len(sys.argv) < 5:
                print("  Nutzung: gardener observe-source add <id> <kind> [key=value ...]")
            else:
                source_id, kind = sys.argv[3], sys.argv[4]
                params = {}
                for kv in sys.argv[5:]:
                    if "=" in kv:
                        k, v = kv.split("=", 1)
                        params[k] = v
                try:
                    entry = af.observe_source_add(source_id, kind, **params)
                    print(f"  [OK] '{source_id}' ({entry['kind']}) registriert")
                except ValueError as e:
                    print(f"  [FEHLER] {e}")

        elif sub == "list":
            configured = af.observe_source_list()
            for sid, cfg in configured.items():
                status = "on " if cfg.get("enabled", True) else "off"
                print(f"  [{status}] {sid:20s} {cfg.get('kind', '?')}")
            if not configured:
                print("  Keine observe-sources konfiguriert.")

        elif sub == "remove":
            source_id = sys.argv[3] if len(sys.argv) > 3 else ""
            if af.observe_source_remove(source_id):
                print(f"  [OK] '{source_id}' entfernt")
            else:
                print(f"  Nicht gefunden: {source_id}")

        elif sub == "refresh":
            source_id = sys.argv[3] if len(sys.argv) > 3 else None
            result = af.observe_sources(source_id)
            for sid, s in result.items():
                if sid == "error":
                    print(f"  [FEHLER] {s}")
                elif "error" in s:
                    print(f"  [FEHLER] {sid}: {s['error']}")
                elif s.get("skipped_disabled"):
                    print(f"  [--] {sid}: deaktiviert")
                else:
                    print(f"  [OK] {sid} ({s['kind']}): "
                          f"{s['indexed']} indexiert, {s['skipped']} unveraendert")
            if not result:
                print("  Keine observe-sources konfiguriert.")

        else:
            print("  Nutzung: gardener observe-source <add|list|remove|refresh> ...")

    elif cmd == "list":
        type_filter = sys.argv[2] if len(sys.argv) > 2 else None
        entries = af.list(type=type_filter)
        for e in entries:
            src = e.get("source", "?")
            print(f"  [{e['type']:10s}] {e['name']:40s} ({src})")
        if not entries:
            print("  Keine Einträge.")

    elif cmd == "delete":
        name = sys.argv[2] if len(sys.argv) > 2 else ""
        if af.delete(name):
            print(f"  [OK] '{name}' gelöscht")
        else:
            print(f"  Nicht gefunden: {name}")

    elif cmd == "status":
        s = af.status()
        print(json.dumps(s, indent=2, ensure_ascii=False))

    elif cmd == "memo":
        text = " ".join(sys.argv[2:]) if len(sys.argv) > 2 else ""
        entry = af.memo(text)
        print(f"  [OK] Memo gespeichert: {entry['name']}")

    elif cmd == "lesson":
        title = sys.argv[2] if len(sys.argv) > 2 else ""
        content = " ".join(sys.argv[3:]) if len(sys.argv) > 3 else ""
        entry = af.lesson(title, content)
        print(f"  [OK] Lesson: {entry['name']}")

    elif cmd == "recall":
        query = " ".join(sys.argv[2:]) if len(sys.argv) > 2 else ""
        results = af.recall(query)
        for r in results:
            w = r.get("meta", {}).get("weight", 0)
            bar = "*" * int(w * 5) + " " * (5 - int(w * 5))
            print(f"  [{bar}] {r['name']:40s} ({r['type']})")
        if not results:
            print("  Keine Erinnerungen gefunden.")

    elif cmd == "consolidate":
        stats = af.consolidate()
        print(f"  Decay:     {stats['decayed']} Einträge verblasst")
        print(f"  Vergessen: {stats['forgotten']} Einträge gelöscht")
        print(f"  Behalten:  {stats['kept']} Einträge stabil")

    elif cmd == "session-end":
        summary = " ".join(sys.argv[2:]) if len(sys.argv) > 2 else ""
        entry = af.session_end(summary)
        print(f"  [OK] Session gespeichert: {entry['name']}")

    elif cmd == "tasks":
        status_filter = sys.argv[2] if len(sys.argv) > 2 else None
        tasks = af.tasks(status_filter)
        if not tasks:
            print("  Keine Tasks.")
        for tk in tasks:
            m = tk.get("meta", {})
            st = m.get("status", "?")
            pr = m.get("priority", "normal")
            due = m.get("due", "")
            marker = "[x]" if st == "done" else "[ ]"
            print(f"  {marker} {tk['name']:30s} ({pr}) {due}")

    elif cmd == "task":
        name = sys.argv[2] if len(sys.argv) > 2 else "unnamed"
        content = " ".join(sys.argv[3:]) if len(sys.argv) > 3 else ""
        entry = af.task(name, content=content)
        print(f"  [OK] Task '{entry['name']}' erstellt")

    elif cmd == "done":
        name = sys.argv[2] if len(sys.argv) > 2 else ""
        entry = af.task_done(name)
        if entry:
            print(f"  [OK] Task '{name}' erledigt")
        else:
            print(f"  Task nicht gefunden: {name}")

    else:
        print(f"  Unbekannter Befehl: {cmd}")


if __name__ == "__main__":
    main()