"""
localVault.py — the runner's long-lived Obsidian vault host.
=============================================================
Started by the runner (src/vaults.ts, VaultHostProcess) with the venv python
and PYTHONPATH = the shared bundle, so the pure-Python engine
``shared.runtime.vault.engine.Vault`` is importable. One process serves every
vault registered on this computer.

Protocol: JSON lines on stdin / stdout.

  request   {"id": "v1", "cmd": "<cmd>", ...}
  response  {"id": "v1", "ok": true, "result": ...} | {"id": "v1", "ok": false, "error": "<code>"}
  event     {"event": "status", "vaultId", "status", "notes", "bytes", "indexRev", "error"?}
            {"event": "ready"}

Commands:
  register    {vault: {id, root, configDirName, include, exclude, indexPath}, cursorKey}
              → {notes, bytes, indexRev, capabilities}  (opens + incremental scan)
  unregister  {vaultId}
  op          {vaultId, op, args, folders, write, idempotency_key} → the engine envelope
  changed     {vaultId, paths | null}  (debounced fs.watch batch from Node; null = full)
  mirror_sync {vaultId, token, apiBase, budgetSeconds} → one cloud-copy sync round
              (localVaultMirror.py; storage: mirror vaults only)
  mirror_drop {vaultId}  (mirroring turned off: forget the ledger)
  ping

Authority is checked by the Node side BEFORE an op reaches this process
(capability, grant, read/write, route, revocation); this host enforces the
remaining boundary:
  - only public methods defined on the engine's Vault class can be called,
    never ``apply`` / ``resume_recovery`` / ``refresh`` / lifecycle methods;
  - the grant's ``folders`` always reach the engine, the caller's never (an op
    that cannot be folder-scoped is refused when the grant is folder-bounded);
  - a write method only proposes; the host applies THAT proposal in the same
    request (``args.dry_run`` returns the proposal instead), so a caller can
    never hand the engine forged bytes to apply. ``idempotency_key`` reaches
    the write method, and an applied result is cached per key: a retry
    returns it and never re-applies.

Change detection: Node feeds fs.watch batches (``changed``); a background
thread runs a stat-only reconciliation per vault every 10 minutes (plan §6.4),
which also covers watcher overflow, sleep and platforms without recursive
fs.watch. Scans of one vault are serialised and coalesced.

Never written to stdout/stderr: note content, op arguments, capability tokens.
"""
from __future__ import annotations

import importlib
import importlib.util
import inspect
import json
import os
import sqlite3
import sys
import threading
import time
from collections import OrderedDict
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from typing import Any

RECONCILE_SECONDS = 600
RECONCILE_TICK = 30
# Engine methods a caller may never name. ``apply`` / ``resume_recovery``
# take a caller-supplied proposal (a sandbox could write forged bytes); the
# others are host lifecycle. Writes are applied by the host itself (cmd_op).
_HOST_ONLY = frozenset({"apply", "resume_recovery", "refresh", "close", "capabilities_list", "shutdown"})
# Engine write methods: each returns a proposal, which the host applies in the
# same request (unless args.dry_run). Mirror of VAULT_WRITE_OPS in src/vaults.ts.
WRITE_OPS = frozenset({
    "create", "append", "prepend", "patch", "property_set", "property_remove", "move", "rename", "copy",
    "delete", "task", "daily_append", "daily_prepend", "unique", "template_insert", "bookmark",
    "canvas_write", "base_create", "upload",
})
IDEMPOTENCY_CACHE_MAX = 512
# Engine entry points, tried in order (engine.Vault: ``refresh(changed_paths)``
# and ``vault_info()``); older or newer facades may name them differently.
_SCAN_METHODS = ("refresh", "scan", "reindex", "sync")
_STATS_METHODS = ("vault_info", "stats", "info")

_out_lock = threading.Lock()


def _send(obj: dict) -> None:
    line = json.dumps(obj, ensure_ascii=False, default=str)
    with _out_lock:
        sys.stdout.write(line + "\n")
        sys.stdout.flush()


def _log(msg: str) -> None:
    # Metadata only (ids, counts, exception types) — never content.
    try:
        sys.stderr.write(f"[localVault] {msg}\n")
        sys.stderr.flush()
    except Exception:
        pass


# ── engine ──────────────────────────────────────────────────────────────

_engine_cls: Any = None
_engine_error = ""
_envelope: Any = None


_ISOLATED_PKG = "_melaya_vault_engine"


def _import_isolated() -> tuple[Any, Any]:
    """Load shared/runtime/vault as a standalone package.

    The engine only uses relative imports, so it does not need ``shared`` or
    ``shared.runtime``, whose ``__init__`` modules import the whole agent stack
    (agentscope, every tool module: tens of seconds on a cold start). The vault
    host needs none of that."""
    for base in sys.path:
        d = os.path.join(base or ".", "shared", "runtime", "vault")
        init = os.path.join(d, "__init__.py")
        if not (os.path.isfile(init) and os.path.isfile(os.path.join(d, "engine.py"))):
            continue
        spec = importlib.util.spec_from_file_location(_ISOLATED_PKG, init, submodule_search_locations=[d])
        if spec is None or spec.loader is None:
            continue
        mod = importlib.util.module_from_spec(spec)
        sys.modules[_ISOLATED_PKG] = mod
        try:
            spec.loader.exec_module(mod)
            eng_mod = importlib.import_module(f"{_ISOLATED_PKG}.engine")
        except Exception:
            for k in [k for k in sys.modules if k == _ISOLATED_PKG or k.startswith(_ISOLATED_PKG + ".")]:
                sys.modules.pop(k, None)
            raise
        try:
            env_mod = importlib.import_module(f"{_ISOLATED_PKG}.envelope")
        except Exception:
            env_mod = None  # the host builds plain envelopes itself
        return eng_mod, env_mod
    raise ImportError("shared/runtime/vault not found on PYTHONPATH")


def _load_engine() -> Any:
    global _engine_cls, _engine_error, _envelope
    if _engine_cls is not None or _engine_error:
        return _engine_cls
    try:
        eng_mod, env_mod = _import_isolated()
    except Exception:
        try:  # regular package import (slower: runs shared/__init__)
            from shared.runtime.vault import engine as eng_mod  # type: ignore
            try:
                from shared.runtime.vault import envelope as env_mod  # type: ignore
            except Exception:
                env_mod = None
        except Exception as e:  # missing bundle / older bundle
            _engine_error = f"{type(e).__name__}"
            _log(f"vault engine unavailable: {_engine_error}")
            return None
    _envelope = env_mod
    _engine_cls = getattr(eng_mod, "Vault", None)
    if _engine_cls is None:
        _engine_error = "no Vault class"
        _log("vault engine unavailable: no Vault class")
    return _engine_cls


def _err(code: str, message: str) -> dict:
    if _envelope is not None:
        try:
            return _envelope.err(code, message)
        except Exception:
            pass
    return {"ok": False, "error": {"code": code, "message": message, "retry_after": None}}


def _ok(vault_id: str, content: Any) -> dict:
    if _envelope is not None:
        try:
            return _envelope.ok(vault_id, content=content)
        except Exception:
            pass
    return {"ok": True, "vault_id": vault_id, "content": content}


def engine_ops(cls: Any) -> list[str]:
    """Public methods DEFINED on the Vault class (and its own bases, not object)."""
    names: set[str] = set()
    for klass in getattr(cls, "__mro__", (cls,)):
        if klass is object:
            continue
        for name, member in vars(klass).items():
            if name.startswith("_") or name in _HOST_ONLY:
                continue
            fn = member.__func__ if isinstance(member, (staticmethod, classmethod)) else member
            if inspect.isfunction(fn):
                names.add(name)
    return sorted(names)


def _params(fn: Any) -> tuple[set[str], bool]:
    try:
        sig = inspect.signature(fn)
    except (TypeError, ValueError):
        return set(), True
    names = {p.name for p in sig.parameters.values()
             if p.kind in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY)}
    var_kw = any(p.kind == p.VAR_KEYWORD for p in sig.parameters.values())
    return names, var_kw


def _call_filtered(fn: Any, **kwargs: Any) -> Any:
    """Call with only the keyword arguments the callable accepts."""
    names, var_kw = _params(fn)
    return fn(**(kwargs if var_kw else {k: v for k, v in kwargs.items() if k in names}))


# ── vaults ──────────────────────────────────────────────────────────────

class HostedVault:
    def __init__(self, vid: str, root: str, index_path: str, engine: Any):
        self.id = vid
        self.root = root
        self.index_path = index_path
        self.engine = engine
        self.scan_lock = threading.Lock()
        self.pending_lock = threading.Lock()
        self.pending: set[str] | None = set()   # None = a full scan is queued
        self.queued = False
        self.last_full = 0.0
        self.last_status: tuple | None = None
        self.closed = False
        # Every engine call is counted, so close() waits for in-flight scans,
        # stats and ops instead of closing SQLite under them.
        self.cv = threading.Condition()
        self.inflight = 0
        # Single writer per vault, plus applied results by idempotency key:
        # a retried write returns the stored result and never re-applies.
        self.write_lock = threading.Lock()
        self.applied: OrderedDict = OrderedDict()


class VaultClosed(Exception):
    pass


@contextmanager
def _using(hv: HostedVault):
    with hv.cv:
        if hv.closed:
            raise VaultClosed()
        hv.inflight += 1
    try:
        yield hv.engine
    finally:
        with hv.cv:
            hv.inflight -= 1
            hv.cv.notify_all()


_vaults: dict[str, HostedVault] = {}
_vaults_lock = threading.Lock()
_req_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="vault-req")
_scan_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="vault-scan")


class ScanFailed(Exception):
    """The engine answered a scan with an error envelope (code only, no content)."""


def _scan_now(hv: HostedVault, paths: list[str] | None) -> None:
    """One incremental scan through the engine; serialised per vault."""
    with hv.scan_lock, _using(hv) as eng:
        for name in _SCAN_METHODS:
            fn = getattr(eng, name, None)
            if not callable(fn):
                continue
            names, var_kw = _params(fn)
            kwargs: dict[str, Any] = {}
            if paths is not None:
                if "changed_paths" in names or var_kw:
                    kwargs["changed_paths"] = paths
                elif "paths" in names:
                    kwargs["paths"] = paths
                # else: this method only does full scans
            r = fn(**kwargs)
            if isinstance(r, dict) and r.get("ok") is False:
                code = (r.get("error") or {}).get("code") if isinstance(r.get("error"), dict) else None
                raise ScanFailed(str(code or "invalid"))
            break
        if paths is None:
            hv.last_full = time.time()


def _stats(hv: HostedVault) -> dict:
    """notes / bytes / indexRev, from the engine when it reports them, else
    from the index database (read-only), else zeros."""
    eng = hv.engine
    for name in _STATS_METHODS:
        fn = getattr(eng, name, None)
        if not callable(fn):
            continue
        try:
            r = _call_filtered(fn)
        except Exception:
            continue
        if isinstance(r, dict) and r.get("ok") is False:
            continue
        d = r.get("content") if isinstance(r, dict) and isinstance(r.get("content"), dict) else r
        if isinstance(d, dict) and ("notes" in d or "notes_count" in d):
            rev = r.get("index_rev") if isinstance(r, dict) and r.get("index_rev") is not None else d.get("index_rev", 0)
            return {
                "notes": int(d.get("notes", d.get("notes_count", 0)) or 0),
                "bytes": int(d.get("size", d.get("bytes", 0)) or 0),
                "indexRev": int(rev or 0),
            }
    try:
        con = sqlite3.connect(f"file:{hv.index_path}?mode=ro", uri=True, timeout=5)
        try:
            notes = con.execute("SELECT COUNT(*) FROM notes").fetchone()[0]
            size = con.execute("SELECT COALESCE(SUM(size), 0) FROM manifest").fetchone()[0]
            row = con.execute("SELECT value FROM meta WHERE key='index_rev'").fetchone()
            return {"notes": int(notes or 0), "bytes": int(size or 0), "indexRev": int(row[0]) if row else 0}
        finally:
            con.close()
    except Exception:
        return {"notes": 0, "bytes": 0, "indexRev": 0}


def _emit_status(hv: HostedVault, status: str = "ready", error: str | None = None, force: bool = False) -> dict:
    st: dict = {}
    if status == "ready":
        try:
            with _using(hv):
                st = _stats(hv)
        except VaultClosed:
            return {}
    key = (status, st.get("notes"), st.get("bytes"), st.get("indexRev"), error)
    if force or key != hv.last_status:
        hv.last_status = key
        ev = {"event": "status", "vaultId": hv.id, "status": status, **st}
        if error:
            ev["error"] = error[:300]
        _send(ev)
    return st


def _queue_scan(hv: HostedVault, paths: list[str] | None) -> None:
    """Coalesce into the vault's pending set; one drain worker per vault."""
    with hv.pending_lock:
        if paths is None or hv.pending is None:
            hv.pending = None
        else:
            hv.pending.update(p for p in paths if isinstance(p, str) and p)
            if len(hv.pending) > 5000:
                hv.pending = None
        if hv.queued:
            return
        hv.queued = True
    _scan_pool.submit(_drain, hv)


def _drain(hv: HostedVault) -> None:
    while True:
        with hv.pending_lock:
            batch = hv.pending
            if batch is not None and not batch:
                hv.queued = False
                return
            hv.pending = set()
        if hv.closed:
            with hv.pending_lock:
                hv.queued = False
            return
        try:
            _scan_now(hv, None if batch is None else sorted(batch))
            _emit_status(hv)
        except VaultClosed:
            continue
        except Exception as e:
            why = str(e) if isinstance(e, ScanFailed) else type(e).__name__
            _log(f"scan failed for {hv.id}: {why}")
            _emit_status(hv, "error", f"index refresh failed ({why})")


def _reconcile_loop() -> None:
    while True:
        time.sleep(RECONCILE_TICK)
        now = time.time()
        with _vaults_lock:
            due = [hv for hv in _vaults.values() if now - hv.last_full >= RECONCILE_SECONDS]
        for hv in due:
            hv.last_full = now  # do not re-queue while the scan runs
            _queue_scan(hv, None)


def _capabilities(eng: Any) -> dict:
    """Capability report stored by the server (agents.vaults.capabilities)."""
    caps: dict[str, Any] = {"ops": engine_ops(type(eng)), "source": "device", "app_bridge": False}
    try:
        lst = getattr(eng, "capabilities_list", None)
        if callable(lst):
            caps["features"] = list(lst())
        rep = getattr(eng, "capabilities", None)
        val = rep() if callable(rep) else None
        if isinstance(val, dict) and val.get("ok") is not False:
            if isinstance(val.get("report"), dict):
                caps["report"] = val["report"]
            if "features" not in caps and isinstance(val.get("items"), list):
                caps["features"] = list(val["items"])
    except Exception:
        pass
    return caps


def _close(hv: HostedVault) -> None:
    with hv.cv:
        hv.closed = True
        hv.cv.wait_for(lambda: hv.inflight == 0, timeout=120)
    with hv.scan_lock:
        fn = getattr(hv.engine, "close", None)
        if callable(fn):
            try:
                fn()
            except Exception:
                pass


# ── commands ────────────────────────────────────────────────────────────

class HostError(Exception):
    """An RPC-level failure (the Node side maps it); ``code`` is short and fixed."""

    def __init__(self, code: str):
        super().__init__(code)
        self.code = code


def cmd_register(msg: dict) -> dict:
    cls = _load_engine()
    if cls is None:
        raise HostError("vault engine unavailable in this runner's runtime; update the runner")
    v = msg.get("vault") or {}
    vid = str(v.get("id") or "")
    root = str(v.get("root") or "")
    index_path = str(v.get("indexPath") or "")
    if not vid.startswith("vlt_") or not root or not index_path:
        raise HostError("invalid")
    if not os.path.isdir(root):
        raise HostError("not_a_folder")
    os.makedirs(os.path.dirname(index_path), exist_ok=True)
    key_hex = str(msg.get("cursorKey") or "")
    try:
        cursor_key = bytes.fromhex(key_hex) if key_hex else os.urandom(32)
    except ValueError:
        cursor_key = os.urandom(32)
    with _vaults_lock:
        old = _vaults.pop(vid, None)
    if old is not None:
        _close(old)
    eng = _call_filtered(
        cls,
        root=root,
        vault_id=vid,
        label=str(v.get("label") or vid),
        index_path=index_path,
        config_dir_name=str(v.get("configDirName") or ".obsidian"),
        include_globs=[g for g in (v.get("include") or []) if isinstance(g, str)],
        exclude_globs=[g for g in (v.get("exclude") or []) if isinstance(g, str)],
        cursor_key=cursor_key,
    )
    hv = HostedVault(vid, root, index_path, eng)
    _send({"event": "status", "vaultId": vid, "status": "indexing"})
    try:
        _scan_now(hv, None)
    except Exception as e:
        _close(hv)
        why = str(e) if isinstance(e, ScanFailed) else type(e).__name__
        _log(f"initial index failed for {vid}: {why}")
        raise HostError(f"indexing failed ({why})")
    with _vaults_lock:
        _vaults[vid] = hv
    st = _emit_status(hv, force=True)
    return {**st, "capabilities": _capabilities(eng)}


def cmd_unregister(msg: dict) -> dict:
    vid = str(msg.get("vaultId") or "")
    with _vaults_lock:
        hv = _vaults.pop(vid, None)
    _drop_mirror(vid, delete=False)
    if hv is not None:
        _close(hv)
    return {"removed": hv is not None}


# ── cloud copy (localVaultMirror.py, plan §6.4 S4) ──────────────────────
# One Mirror per vault, one sync at a time per vault. The capability arrives
# with each request from Node (memory only) and is never written to disk.

_mirror_mod: Any = None
_mirrors: dict[str, tuple[Any, threading.Lock]] = {}
_mirrors_lock = threading.Lock()


def _mirror_module() -> Any:
    global _mirror_mod
    if _mirror_mod is None:
        path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "localVaultMirror.py")
        spec = importlib.util.spec_from_file_location("_melaya_vault_mirror", path)
        if spec is None or spec.loader is None:
            raise HostError("mirror module missing; update the runner")
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        _mirror_mod = mod
    return _mirror_mod


def _drop_mirror(vid: str, delete: bool) -> None:
    with _mirrors_lock:
        entry = _mirrors.pop(vid, None)
    if entry is not None:
        with entry[1]:
            entry[0].close()
    if delete:
        with _vaults_lock:
            hv = _vaults.get(vid)
        base = os.path.join(os.path.dirname(hv.index_path), "mirror.sqlite") if hv else None
        for suffix in ("", "-wal", "-shm"):
            if base:
                try:
                    os.remove(base + suffix)
                except FileNotFoundError:
                    pass


def cmd_mirror_sync(msg: dict) -> dict:
    """One sync round: {vaultId, token, apiBase, budgetSeconds} → status."""
    vid = str(msg.get("vaultId") or "")
    with _vaults_lock:
        hv = _vaults.get(vid)
    if hv is None:
        raise HostError("unknown_vault")
    token = str(msg.get("token") or "")
    base = str(msg.get("apiBase") or "")
    if not token or not base.startswith(("https://", "http://")):
        raise HostError("invalid")
    policy = getattr(hv.engine, "policy", None)
    if policy is None:
        return {"state": "error", "error": "this vault engine cannot mirror"}
    mod = _mirror_module()
    with _mirrors_lock:
        entry = _mirrors.get(vid)
        if entry is None:
            ledger = os.path.join(os.path.dirname(hv.index_path), "mirror.sqlite")
            entry = (mod.Mirror(vid, policy, hv.index_path, ledger, log=_log), threading.Lock())
            _mirrors[vid] = entry
    mirror, lock = entry
    if not lock.acquire(blocking=False):
        return {"state": "busy"}  # coalesced: the running round picks the changes up next
    try:
        result = mirror.sync(mod.HttpTransport(base, token), budget_s=float(msg.get("budgetSeconds") or 600))
    finally:
        lock.release()
    if result.get("rescan"):
        _queue_scan(hv, [p for p in result["rescan"] if isinstance(p, str)])
    return result


def cmd_mirror_drop(msg: dict) -> dict:
    """Mirroring was turned off: forget the ledger (the copy is purged server-side)."""
    _drop_mirror(str(msg.get("vaultId") or ""), delete=True)
    return {"dropped": True}


def cmd_changed(msg: dict) -> dict:
    vid = str(msg.get("vaultId") or "")
    with _vaults_lock:
        hv = _vaults.get(vid)
    if hv is None:
        raise HostError("unknown_vault")
    paths = msg.get("paths")
    _queue_scan(hv, None if not isinstance(paths, list) else [p for p in paths if isinstance(p, str)][:5000])
    return {"queued": True}


def _engine_call(hv: HostedVault, fn: Any, op: str, args: dict) -> dict:
    """Call one engine method; always an envelope."""
    try:
        with _using(hv):
            result = fn(**args)
    except VaultClosed:
        raise HostError("unknown_vault")
    except TypeError as e:
        # Wrong / missing arguments from the model: the message names parameters only.
        return _err("invalid", f"bad arguments for {op}: {str(e)[:200]}")
    except Exception as e:
        env = getattr(e, "envelope", None)
        if callable(env):
            try:
                return env()
            except Exception:
                pass
        _log(f"op {op} failed on {hv.id}: {type(e).__name__}")
        return _err("invalid", f"{op} failed ({type(e).__name__})")
    if isinstance(result, dict) and isinstance(result.get("ok"), bool):
        return result
    return _ok(hv.id, result)


def cmd_op(msg: dict) -> dict:
    """Dispatch one engine op; always returns an envelope (never raises for
    engine failures — only an unknown vault is an RPC error, so Node can
    re-register it after a host restart).

    Writes: the engine's write method only PROPOSES a change; the host applies
    that exact proposal in the same request (``apply`` itself is never callable
    by a caller), unless ``args.dry_run`` asks for the proposal only."""
    vid = str(msg.get("vaultId") or "")
    with _vaults_lock:
        hv = _vaults.get(vid)
    if hv is None:
        raise HostError("unknown_vault")
    cls = _load_engine()
    op = str(msg.get("op") or "")
    if cls is None:
        return _err("unsupported", "The vault engine is not available in this runner's runtime.")
    if op in _HOST_ONLY:
        return _err("forbidden", f"{op[:64]} cannot be called directly.")
    if op not in engine_ops(type(hv.engine)):
        return _err("unsupported", f"Unknown vault operation: {op[:64]}")
    fn = getattr(hv.engine, op, None)
    if not callable(fn):
        return _err("unsupported", f"Unknown vault operation: {op[:64]}")
    raw = msg.get("args")
    args = {k: v for k, v in (raw.items() if isinstance(raw, dict) else [])
            if isinstance(k, str) and not k.startswith("_")}
    # Folder bounds come from the grant only, never from the model's arguments.
    args.pop("folders", None)
    args.pop("idempotency_key", None)
    dry_run = bool(args.pop("dry_run", False)) if op in WRITE_OPS else False
    # An empty grant folder list means the whole vault, which the engine spells
    # ``folders=None`` (an empty list there would match nothing).
    folders = [f for f in (msg.get("folders") or []) if isinstance(f, str)] or None
    names, var_kw = _params(fn)
    if "folders" in names or var_kw:
        args["folders"] = folders
    elif folders:
        return _err("unsupported", f"{op} cannot be limited to the folders this run may see.")
    idem = msg.get("idempotency_key")
    idem = idem[:200] if isinstance(idem, str) and idem else None
    if idem and ("idempotency_key" in names or var_kw):
        args["idempotency_key"] = idem

    if op not in WRITE_OPS or dry_run:
        return _engine_call(hv, fn, op, args)

    apply_fn = getattr(hv.engine, "apply", None)
    if not callable(apply_fn):
        return _err("unsupported", "This vault engine cannot apply changes.")
    key = (op, idem) if idem else None
    with hv.write_lock:
        if key is not None and key in hv.applied:
            hv.applied.move_to_end(key)
            return hv.applied[key]
        proposed = _engine_call(hv, fn, op, args)
        if proposed.get("ok") is not True:
            return proposed
        proposal = proposed.get("proposal")
        if not isinstance(proposal, dict):
            # Nothing to apply (e.g. a no-op edit): the engine's answer stands.
            return proposed
        a_names, a_var_kw = _params(apply_fn)
        a_args: dict[str, Any] = {"proposal": proposal}
        if "folders" in a_names or a_var_kw:
            a_args["folders"] = folders
        result = _engine_call(hv, apply_fn, "apply", a_args)
        if key is not None and result.get("ok") is True:
            hv.applied[key] = result
            while len(hv.applied) > IDEMPOTENCY_CACHE_MAX:
                hv.applied.popitem(last=False)
    if result.get("ok") is True and isinstance(result.get("path"), str):
        _queue_scan(hv, [result["path"]])  # fresh counts even without a watcher
    return result


_COMMANDS = {
    "register": cmd_register,
    "unregister": cmd_unregister,
    "changed": cmd_changed,
    "op": cmd_op,
    "mirror_sync": cmd_mirror_sync,
    "mirror_drop": cmd_mirror_drop,
    "ping": lambda _m: {"pong": True, "engine": _load_engine() is not None},
}


def _handle(msg: dict) -> None:
    rid = msg.get("id")
    fn = _COMMANDS.get(str(msg.get("cmd") or ""))
    if fn is None:
        _send({"id": rid, "ok": False, "error": "unknown_command"})
        return
    try:
        _send({"id": rid, "ok": True, "result": fn(msg)})
    except HostError as e:
        _send({"id": rid, "ok": False, "error": e.code})
    except Exception as e:
        _log(f"{msg.get('cmd')} failed: {type(e).__name__}")
        _send({"id": rid, "ok": False, "error": f"host_error ({type(e).__name__})"})


def main() -> int:
    try:
        sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
        sys.stdin.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
    except Exception:
        pass
    threading.Thread(target=_reconcile_loop, name="vault-reconcile", daemon=True).start()
    _send({"event": "ready"})
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(msg, dict):
            continue
        if msg.get("cmd") == "ping":
            _handle(msg)
        else:
            _req_pool.submit(_handle, msg)
    # stdin closed: the runner is gone. Close indexes (checkpoints the WAL).
    with _vaults_lock:
        hosted = list(_vaults.values())
        _vaults.clear()
    for hv in hosted:
        _close(hv)
    return 0


if __name__ == "__main__":
    sys.exit(main())