"""StashBase sidecar daemon.

Owns the **single** Milvus Lite DB at ``<kb_root>/.stashbase/store/milvus.db``
with **one** collection (``vectors_openai_1536``). V1 ships a single
fixed embedder (OpenAI), so the whole KB lives in that one collection —
there's no embedder switching, no per-provider collection pool, and no
active/archive distinction. Each space (a folder directly under the KB
root) is **bound** so the daemon knows it exists; all spaces share the
collection.

The Node side (server/mfs-daemon.ts) spawns this script once and talks
to it over stdin/stdout in line-delimited JSON.

Protocol
--------
Each request is one JSON object on a single stdin line:

    {"id": 7, "op": "<name>", "args": {...}}

Each response is one JSON object on a single stdout line:

    {"id": 7, "ok": true,  "result": ...}
    {"id": 7, "ok": false, "error": "..."}

The daemon also emits unsolicited progress events
(``{"event": "ready" | "starting" | "error", ...}``) — Node treats
events as informational and matches results back to requests by ``id``.

Supported ops
-------------
- ``bind_space {space, provider, api_key?, model?, dimension?}``
                        — register that ``space`` exists. The first bind
                          carrying an ``api_key`` builds the embedder +
                          collection; later binds reuse them. Idempotent;
                          safe to call after a daemon respawn.
- ``unbind_space {space}``
                        — forget ``space``. Existing rows stay; the
                          space can be re-bound later.
- ``upsert {path, content, ext, file_hash?}``
                        — chunk + embed + insert/replace one file.
                          ``path`` is **kbRoot-relative** (e.g.
                          ``cs183b/lecture-01.md``); the space is the
                          first path segment.
- ``delete {path}``     — drop rows for one file.
- ``delete_prefix {prefix}``
                        — drop rows for files under a folder.
- ``rename {old, new, content, ext, file_hash}``
                        — move a file's rows (fast-path reuses vectors
                          when the hash matches; else re-embed).
- ``rename_prefix {old, new, files}``
                        — folder rename: bulk version.
- ``search {query, space?, top_k}``
                        — hybrid search (dense + BM25 + RRF) in the
                          collection, optionally scoped to one space via
                          a ``source like "<space>/%"`` filter.
- ``status {space?}``   — name-only diff of disk vs index. ``space``
                          omitted means the whole library.
- ``scan_diff {space?}`` — content-hash diff. ``space`` omitted = whole
                          library.
- ``list {space?}``     — ``{path: file_hash}`` of every indexed file.
                          Scoped by ``space`` prefix when given.
- ``close_store``       — release Milvus Lite's flock so the server can
                          delete or move the DB file.
- ``set_rules {excluded_dirs?, max_indexable_bytes?, include_extensions?}``
                        — receive indexing rules from Node (single source
                          of truth there); built-in constants are only the
                          fallback for an old Node. Echoes effective rules.

Paths
-----
``path`` / ``prefix`` / ``old`` / ``new`` in every op are
**kbRoot-relative POSIX** (``cs183b/lecture-01.md``). The first path
segment is the space name; the daemon uses it to look up the bound
collection. The Node side translates between space-relative (its native
representation) and kbRoot-relative at the indexer boundary.
"""

from __future__ import annotations

import argparse
import datetime
import json
import os
import sys
import time
import traceback
from collections import defaultdict
from pathlib import Path
from typing import Any

# Content + chunk-id hashing. BLAKE3 is ~3-5x SHA256 and streaming-friendly
# — matters for reconcile over large PDFs / videos. The Node side
# (indexer.mfs.ts) and mfs's Scanner (patched below) must use the SAME
# algorithm, or scan_diff would see every file as modified.
from blake3 import blake3

# Defer all heavy imports until after stdout is unbuffered + greeting
# is printed, so the Node side can tell quickly whether Python even
# launched.
print(json.dumps({"event": "starting", "pid": os.getpid()}), flush=True)


# ---------------------------------------------------------------- embedder
#
# V1 ships a single embedder: OpenAI `text-embedding-3-small` (1536d).
# There is no embedder switching and no local fallback — the whole KB
# uses one collection. Built lazily on the first bind that carries an
# API key; the daemon may have zero embedders loaded at idle.

def make_embedder(provider: str = "openai", *, model=None, api_key=None, dimension=None):
    """Build the OpenAI embedding provider satisfying MFS's protocol
    (`.embed(texts) -> list[list[float]]`, `.dimension`, `.model_name`).

    Rolled in-house — see `_OpenAIEmbedder`. ``provider`` is accepted for
    protocol compatibility but must be ``openai`` (V1 has no other).
    """
    if provider != "openai":
        raise ValueError(f"unsupported embedder provider {provider!r}; V1 is openai-only")
    if not api_key:
        raise ValueError("openai embedder requires api_key")
    return _OpenAIEmbedder(
        model=model or "text-embedding-3-small",
        api_key=api_key,
        dimension=dimension,
    )


class _OpenAIEmbedder:
    """OpenAI embedding provider, rolled in-house.

    Rolled separately from `mfs.embedder.get_provider('openai')` so we can
    (a) cap the OpenAI client timeout at 60s instead of the SDK default
    of 10 minutes, and (b) wrap retries around transient errors without
    monkey-patching MFS internals.

    Satisfies MFS's `EmbeddingProvider` protocol: `.embed(texts)`,
    `.dimension`, `.model_name`.
    """

    _NATIVE_DIMS = {
        "text-embedding-3-small": 1536,
        "text-embedding-3-large": 3072,
        "text-embedding-ada-002": 1536,
    }

    # OpenAI's embedding endpoint enforces a request-level token ceiling
    # (300k for text-embedding-3* at time of writing). Keep a margin and
    # batch locally so one big space cannot take the daemon down.
    _MAX_REQUEST_TOKENS = 250_000
    _MAX_BATCH_ITEMS = 128

    def __init__(self, *, model: str, api_key: str, dimension: int | None = None,
                 timeout: float = 60.0, max_retries: int = 3, base_delay: float = 1.5) -> None:
        import openai
        self._openai = openai
        self._client = openai.OpenAI(api_key=api_key, timeout=timeout)
        self.model_name = model
        self.dimension = dimension or self._NATIVE_DIMS.get(model, 1536)
        self._max_retries = max(1, max_retries)
        self._base_delay = base_delay

    def embed(self, texts: list[str]) -> list[list[float]]:
        if not texts:
            return []
        out: list[list[float]] = []
        batch: list[str] = []
        batch_tokens = 0
        for text in texts:
            est = self._estimate_tokens(text)
            if est > self._MAX_REQUEST_TOKENS:
                raise ValueError(
                    f"single embedding input is too large "
                    f"(estimated {est:,} tokens > {self._MAX_REQUEST_TOKENS:,})",
                )
            if batch and (
                batch_tokens + est > self._MAX_REQUEST_TOKENS
                or len(batch) >= self._MAX_BATCH_ITEMS
            ):
                out.extend(self._embed_batch(batch))
                batch = []
                batch_tokens = 0
            batch.append(text)
            batch_tokens += est
        if batch:
            out.extend(self._embed_batch(batch))
        return out

    @staticmethod
    def _estimate_tokens(text: str) -> int:
        # Conservative language-agnostic approximation. English averages
        # ~4 chars/token; CJK is denser, so 3 chars/token leaves margin.
        return max(1, (len(text) + 2) // 3)

    def _embed_batch(self, texts: list[str]) -> list[list[float]]:
        transient = (
            self._openai.APITimeoutError,
            self._openai.APIConnectionError,
            self._openai.RateLimitError,
            self._openai.InternalServerError,
        )
        native = self._NATIVE_DIMS.get(self.model_name)
        kwargs: dict = {"model": self.model_name, "input": texts}
        if native is not None and self.dimension != native:
            kwargs["dimensions"] = self.dimension
        last_err: Exception | None = None
        for attempt in range(self._max_retries):
            try:
                resp = self._client.embeddings.create(**kwargs)
                return [d.embedding for d in resp.data]
            except transient as err:
                last_err = err
                if attempt == self._max_retries - 1:
                    raise
                delay = self._base_delay * (2 ** attempt)
                print(
                    f"[stashbase] openai embed attempt {attempt + 1}/{self._max_retries} "
                    f"failed ({type(err).__name__}); retrying in {delay:.1f}s",
                    file=sys.stderr,
                )
                time.sleep(delay)
        raise RuntimeError(f"embed retry loop exhausted: {last_err}")


# ---------------------------------------------------------------- chunking

def _chunk(path_rel: str, content: str, ext: str):
    """Route to MFS's chunker. Returns list of MFS ``Chunk``.

    HTML is fed in as markdown-shaped plaintext by the Node side (see
    ``server/html.ts``), so we always pass ``.md`` to MFS regardless
    of the on-disk extension — the chunker doesn't know HTML and would
    otherwise fall back to dumb char splits.
    """
    from mfs.ingest.chunker import chunk_file
    effective_ext = ".md" if ext in (".html", ".htm") else ext
    return chunk_file(Path(path_rel), content, effective_ext)


# ---------------------------------------------------------------- store

def _patch_inverted_index_skip() -> None:
    """Drop INVERTED scalar indexes that Milvus Lite refuses.

    MFS adds INVERTED indexes on ``source`` / ``parent_dir`` /
    ``content_type`` / ``is_dir``. Recent pymilvus + Milvus Lite reject
    ``add_index`` calls without ``metric_type``; INVERTED is a scalar
    index with no meaningful metric, so we monkey-patch ``add_index`` to
    swallow them. Affected fields fall back to table-scan filtering,
    which on a single-user KB is comfortably under 10ms. Idempotent —
    flagged via a sentinel attribute on the patched function.
    """
    try:
        from pymilvus.milvus_client.index import IndexParams  # type: ignore
    except ImportError:
        return
    if getattr(IndexParams.add_index, "__stashbase_patched__", False):
        return
    original = IndexParams.add_index

    def _add_index(self, field_name, index_type=None, index_name="", **kwargs):
        if index_type == "INVERTED" and not kwargs.get("metric_type"):
            return self
        return original(self, field_name=field_name, index_type=index_type,
                        index_name=index_name, **kwargs)

    _add_index.__stashbase_patched__ = True  # type: ignore[attr-defined]
    IndexParams.add_index = _add_index


def _patch_scanner_blake3() -> None:
    """Make MFS's ``Scanner.compute_file_hash`` use BLAKE3, not SHA256.

    ``scan_diff`` compares a stored ``file_hash`` (which the Node side now
    computes with BLAKE3) against ``scanner.compute_file_hash(path)``
    recomputed from disk. The two MUST use the same algorithm or every
    file would forever look "modified". MFS upstream hard-codes SHA256
    (``mfs/ingest/scanner.py``), so we override the method to stream the
    raw bytes through BLAKE3 instead — same 64-hex-char output width, so
    the stored-hash column is unaffected. Idempotent via a sentinel.

    Migration: existing rows carry SHA256 hashes; after this patch the
    next *full* content reconcile (manual ``/api/sync``) sees them as
    mismatched and re-embeds, replacing them with BLAKE3. The boot-time
    name-only sync leaves them untouched, so queries keep working on the
    old vectors until then. See build-map 04-indexing.
    """
    try:
        from mfs.ingest.scanner import Scanner  # type: ignore
    except ImportError:
        return
    if getattr(Scanner.compute_file_hash, "__stashbase_blake3__", False):
        return

    def compute_file_hash(self, path) -> str:  # noqa: ANN001
        h = blake3()
        with open(path, "rb") as fh:
            for chunk in iter(lambda: fh.read(1 << 16), b""):
                h.update(chunk)
        return h.hexdigest()

    compute_file_hash.__stashbase_blake3__ = True  # type: ignore[attr-defined]
    Scanner.compute_file_hash = compute_file_hash


## Snapshot file format version. Bumped when the on-disk layout or
## column schema changes incompatibly — the importer hard-rejects any
## version it doesn't recognise to avoid silent miscoercion.
##
## v3: snapshot is now a pure **embedding cache** — the Parquet holds
## only `{text_hash, dense_vector}` (the one artifact that's expensive
## to recompute), and the human-readable descriptor (embedder identity,
## counts, timestamp) lives in a sibling `snapshot.meta.json` written by
## the Node side. Import no longer bulk-inserts chunk rows; instead the
## normal ingestion pipeline re-chunks the (co-located) source files and
## reuses a cached vector whenever a chunk's `BLAKE3(chunk_text)` hits
## the cache, only embedding the misses. v2 (self-contained chunk rows)
## snapshots are rejected — re-export with this build.
SNAPSHOT_VERSION = 3

INDEX_EXCLUDED_DIRS = {
    ".cache",
    ".git",
    ".hg",
    ".next",
    ".nuxt",
    ".output",
    ".parcel-cache",
    ".pnpm-store",
    ".stashbase",
    ".svelte-kit",
    ".turbo",
    ".venv",
    ".vite",
    ".yarn",
    "__pycache__",
    "bower_components",
    "build",
    "coverage",
    "DerivedData",
    "dist",
    "node_modules",
    "out",
    "target",
    "vendor",
}
MAX_INDEXABLE_BYTES = 2 * 1024 * 1024

# Mutable copy of the indexing rules. Node is the single source of truth
# for format/admission knowledge (server/indexable.ts, server/format.ts)
# and pushes its values via the `set_rules` op right after every spawn;
# the constants above are only the fallback for an old Node that doesn't
# know the op. Scanners are created fresh per call (_make_scanner), so a
# rules update takes effect on the next scan with no cache invalidation.
_RULES = {
    "excluded_dirs": set(INDEX_EXCLUDED_DIRS),
    "max_indexable_bytes": MAX_INDEXABLE_BYTES,
    "include_extensions": [".html", ".htm"],
}


def op_set_rules(svc: StashbaseStore, args: dict) -> dict:
    """Receive indexing rules from Node (the single source of truth —
    see data-layer §8.6 I7). Only supplied keys are updated; the reply
    echoes the effective rules so Node can verify what landed."""
    if isinstance(args.get("excluded_dirs"), list):
        _RULES["excluded_dirs"] = {str(d) for d in args["excluded_dirs"] if d}
    mib = args.get("max_indexable_bytes")
    if isinstance(mib, int) and mib > 0:
        _RULES["max_indexable_bytes"] = mib
    if isinstance(args.get("include_extensions"), list):
        _RULES["include_extensions"] = [
            str(e) for e in args["include_extensions"] if str(e).startswith(".")
        ]
    return {
        "excluded_dirs": sorted(_RULES["excluded_dirs"]),
        "max_indexable_bytes": _RULES["max_indexable_bytes"],
        "include_extensions": _RULES["include_extensions"],
    }


def _collection_name(dim: int) -> str:
    """The single collection's name. Encodes dim so a `text-embedding-3-
    large` (3072) config wouldn't collide with the default small (1536);
    the default keeps the historical `vectors_openai_1536` name so
    already-indexed KBs aren't orphaned."""
    return f"vectors_openai_{dim}"


class StashbaseStore:
    """Holds the kb-root-anchored DB and the **single** ``MilvusStore``
    every space shares. V1 has one fixed embedder (OpenAI), so the whole
    KB lives in one collection — no per-provider pool, no active/archive
    distinction.

    Lifecycle:
        1. ``__init__`` records the kb-root and the resolved
           ``milvus.db`` path. No daemon-side I/O yet.
        2. ``bind_space(space, ...)`` — first bind carrying an API key
           creates the embedder + collection; later binds reuse them and
           just register the space. Spaces still get bound so the
           "must bind before writing" contract holds and ``scan_diff`` /
           ``status`` know which spaces exist.
        3. ``store_for_path(path)`` / ``stores()`` — return the one
           ``(embedder, store)``; raise if nothing's bound yet.

    A daemon respawn loses the bindings; the Node side re-issues
    ``bind_space`` for every known space on reconnect.
    """

    def __init__(self, kb_root: str) -> None:
        self._kb_root: Path = Path(kb_root).resolve()
        self._db_path: Path = self._kb_root / ".stashbase" / "store" / "milvus.db"
        legacy_dir = self._kb_root / ".stashbase" / "mfs"
        if legacy_dir.exists() and not self._db_path.parent.exists():
            legacy_dir.rename(self._db_path.parent)
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        # The one embedder + store, created on the first bind with a key.
        self._embedder: Any = None
        self._store: Any = None
        self._dim: int = 0
        # Spaces Node has bound (kb-root-relative first path segment).
        # Membership only — they all share the single collection.
        self._bound: set[str] = set()
        # space -> {BLAKE3(chunk_text): dense_vector}. Transient embedding
        # cache loaded from a snapshot just before a space is (re)indexed;
        # `op_upsert` consults it to skip embedding chunks whose text is
        # unchanged. Cleared by Node once the import-time sync finishes.
        self._vector_cache: dict[str, dict[str, list[float]]] = {}

    @property
    def kb_root(self) -> Path:
        return self._kb_root

    def _ensure_store(self, embedder):
        """Open the single Milvus collection. Idempotent: reuses the
        cached store once created."""
        if self._store is not None:
            return self._store
        from mfs.store import MilvusStore
        from mfs.config import MilvusConfig
        os.environ["MFS_HOME"] = str(self._db_path.parent)
        dim = embedder.dimension
        config = MilvusConfig(uri=str(self._db_path), collection_name=_collection_name(dim))
        store = MilvusStore(config, dim)
        _patch_inverted_index_skip()
        try:
            store.connect()
        except Exception as err:
            # pymilvus wraps Milvus Lite's lock error generically, so we
            # walk the exception chain (both __cause__ and __context__)
            # and also pattern-match the wrapper message.
            chain = [err]
            cur = err
            seen = {id(err)}
            for _ in range(20):
                nxt = cur.__cause__ or cur.__context__
                if nxt is None or id(nxt) in seen:
                    break
                seen.add(id(nxt))
                chain.append(nxt)
                cur = nxt
            msg = str(err)
            is_lock = (
                'open local milvus failed' in msg.lower()
                or any('lock' in str(e).lower() or 'DataDirLocked' in type(e).__name__ for e in chain)
            )
            if is_lock:
                raise RuntimeError(
                    f"Milvus DB is locked by another process: {self._db_path}\n"
                    f"  Most likely a stale stashbase_daemon from a previous run.\n"
                    f"  Fix: pkill -f stashbase_daemon, then retry."
                ) from err
            raise
        # Milvus Lite leaves freshly-created (and re-opened) collections
        # in the "released" state — queries fail with code=101 until we
        # explicitly load. MFS's ensure_collection doesn't do this for us.
        try:
            store.client.load_collection(config.collection_name)
        except Exception as err:
            print(f"[stashbase] load_collection warn: {err}", file=sys.stderr)
        self._embedder = embedder
        self._store = store
        self._dim = dim
        return store

    def bind_space(self, space: str, provider: str, *, api_key=None, model=None, dimension=None) -> dict:
        # First bind with a key builds the embedder + collection; later
        # binds reuse them. Without a key the space is still registered
        # but the collection isn't created — indexing stays disabled until
        # the user supplies an OpenAI key (graceful no-key degrade).
        if self._store is None and api_key:
            embedder = make_embedder(provider, model=model, api_key=api_key, dimension=dimension)
            self._ensure_store(embedder)
        self._bound.add(space)
        return {
            "space": space,
            "provider": "openai",
            "model": getattr(self._embedder, "model_name", None),
            "dim": self._dim,
            "collection": _collection_name(self._dim) if self._dim else None,
        }

    def unbind_space(self, space: str) -> dict:
        had = space in self._bound
        self._bound.discard(space)
        self._vector_cache.pop(space, None)
        return {"space": space, "was_bound": had}

    def set_vector_cache(self, space: str, cache: dict[str, list[float]]) -> None:
        self._vector_cache[space] = cache

    def vector_cache_for(self, space: str) -> dict[str, list[float]] | None:
        return self._vector_cache.get(space)

    def clear_vector_cache(self, space: str) -> bool:
        return self._vector_cache.pop(space, None) is not None

    def store_for_path(self, path: str):
        """Return ``(embedder, store)`` for ``path`` (kb-root-relative).
        Every space shares the one collection; the first path segment
        still has to be a bound space."""
        if "/" not in path:
            # Top-level file directly under kbRoot — not allowed; every
            # file must live inside a space.
            raise RuntimeError(f"path '{path}' is not inside a space (must be '<space>/...')")
        space = path.split("/", 1)[0]
        if space not in self._bound or self._store is None:
            raise RuntimeError(
                f"no bound space matches path '{path}'; call bind_space first "
                "(or set an OpenAI API key)",
            )
        return (self._embedder, self._store)

    def stores(self) -> list[tuple[str, Any, Any]]:
        """The single ``(provider_key, embedder, store)`` as a list (or
        empty before the first keyed bind). Tuple shape kept so call
        sites can unpack uniformly."""
        if self._store is None:
            return []
        return [(f"openai_{self._dim}", self._embedder, self._store)]

    def bound_spaces(self) -> list[str]:
        """Spaces Node has registered, sorted. Used by whole-library
        disk walks."""
        return sorted(self._bound)

    def require_current(self):
        """Return ``(embedder, store, dim)``; raise if no embedder is
        bound yet (e.g. no OpenAI key set)."""
        if self._store is None:
            raise RuntimeError(
                "no embedder bound; call bind_space with an OpenAI API key first",
            )
        return self._embedder, self._store, self._dim

    def close_all(self) -> None:
        """Release Milvus Lite's flock. The next ``bind_space`` reopens."""
        if self._store is not None:
            try:
                self._store.close()
            except Exception:
                pass
        self._store = None
        self._embedder = None
        self._dim = 0
        self._bound.clear()


# ---------------------------------------------------------------- ops

def _hash_text(text: str) -> str:
    return blake3(text.encode("utf-8", errors="replace")).hexdigest()


def _embed_with_cache(svc: "StashbaseStore", path: str, embedder, texts: list[str]) -> list:
    """Embed ``texts``, reusing vectors from the space's snapshot cache
    (keyed by ``BLAKE3(chunk_text)``) where present. A snapshot is a pure
    embedding cache (see ``SNAPSHOT_VERSION`` v3): on import the normal
    pipeline re-chunks the source files and lands here, and only the
    chunks the cache doesn't cover get embedded. No cache → embed all.
    The dim guard drops a cached vector that doesn't match the current
    embedder so a stale / mismatched snapshot can't poison the store."""
    space = path.split("/", 1)[0]
    cache = svc.vector_cache_for(space)
    if not cache:
        return embedder.embed(texts)
    dim = embedder.dimension
    out: list = [None] * len(texts)
    misses: list[int] = []
    for i, t in enumerate(texts):
        v = cache.get(_hash_text(t))
        if v is not None and len(v) == dim:
            out[i] = list(v)
        else:
            misses.append(i)
    if misses:
        embedded = embedder.embed([texts[i] for i in misses])
        for j, i in enumerate(misses):
            out[i] = embedded[j]
    return out


def _require(args: dict, *keys: str) -> None:
    missing = [k for k in keys if args.get(k) is None]
    if missing:
        raise ValueError(f"missing field(s): {', '.join(missing)}")


def op_bind_space(svc: StashbaseStore, args: dict) -> dict:
    """Register ``space`` → ``provider`` mapping. Creates the
    collection if first use; idempotent."""
    _require(args, "space", "provider")
    return svc.bind_space(
        args["space"],
        args["provider"],
        api_key=args.get("api_key"),
        model=args.get("model"),
        dimension=args.get("dimension"),
    )


def op_unbind_space(svc: StashbaseStore, args: dict) -> dict:
    _require(args, "space")
    return svc.unbind_space(args["space"])


def op_upsert(svc: StashbaseStore, args: dict) -> dict:
    """Replace all rows for ``path`` with freshly-embedded chunks.

    Args: ``path`` (kb-root-relative POSIX), ``content`` (raw text /
    pre-flattened HTML-as-markdown), ``ext``, optional ``file_hash``.
    Routes to the bound provider for the space implied by ``path``.
    """
    from mfs.store import ChunkRecord

    _require(args, "path", "content")
    path = args["path"]
    content = args["content"]
    if len(content.encode("utf-8", errors="replace")) > _RULES["max_indexable_bytes"]:
        raise ValueError(
            f"file is too large to index ({len(content):,} chars; "
            f"limit {_RULES['max_indexable_bytes']:,} bytes)",
        )
    ext = args.get("ext", ".md")
    # File-level metadata (user front-matter / HTML <meta> + the agent's
    # file-metadata.md sidecar), resolved Node-side. Used as the base for
    # every chunk's metadata; per-chunk keys (e.g. heading_text) win.
    file_metadata = args.get("metadata") or {}
    if not isinstance(file_metadata, dict):
        file_metadata = {}
    embedder, store = svc.store_for_path(path)
    chunks = _chunk(path, content, ext)
    file_hash = args.get("file_hash") or _hash_text(content)
    t0 = time.time()

    # Defensive: also wipe the same source from OTHER collections, so
    # if a user switched providers we don't accidentally retain stale
    # rows under the old collection that'd surface in search hits.
    for _pk, _emb, other in svc.stores():
        if other is store:
            continue
        try:
            other.delete_by_source(path)
        except Exception:
            pass
    store.delete_by_source(path)
    if not chunks:
        return {"chunks": 0, "embed_ms": 0, "total_ms": int((time.time() - t0) * 1000)}

    texts = [c.text for c in chunks]
    te0 = time.time()
    vectors = _embed_with_cache(svc, path, embedder, texts)
    embed_ms = int((time.time() - te0) * 1000)

    parent = "/".join(path.split("/")[:-1])
    records = []
    for i, (ch, vec) in enumerate(zip(chunks, vectors)):
        records.append(ChunkRecord(
            id=blake3(
                f"{path}:{ch.start_line}:{ch.end_line}:{_hash_text(ch.text)}".encode(),
            ).hexdigest()[:32],
            source=path,
            parent_dir=parent,
            chunk_index=i,
            start_line=ch.start_line,
            end_line=ch.end_line,
            chunk_text=ch.text,
            dense_vector=vec,
            content_type=ch.content_type,
            file_hash=file_hash,
            is_dir=False,
            embed_status="complete",
            metadata={**file_metadata, **(ch.metadata or {})},
            account_id="stashbase",
        ))
    store.insert_chunks(records)
    return {
        "chunks": len(records),
        "embed_ms": embed_ms,
        "total_ms": int((time.time() - t0) * 1000),
    }


def op_delete(svc: StashbaseStore, args: dict) -> dict:
    """Drop rows whose ``source`` equals ``path`` from every open
    collection — a file may have rows in any collection if the user
    switched providers."""
    _require(args, "path")
    path = args["path"]
    n = 0
    for _pk, _emb, store in svc.stores():
        try:
            n += int(store.delete_by_source(path))
        except Exception:
            pass
    return {"removed": n}


def op_rename(svc: StashbaseStore, args: dict) -> dict:
    """Rename a file's chunks across all collections.

    Fast path (no embedding) when the caller-supplied ``file_hash``
    matches every stored row's ``file_hash`` for the old source: copy
    each row with the new source / id / parent_dir, keep the cached
    ``dense_vector``, then drop the old rows. Saves the embedding round
    trip for true renames (huge PDFs, mass folder moves) — the common
    case once scan_diff pairs adds + deletes by hash.

    Falls back to delete-and-reinsert when the hash differs (content
    actually changed alongside the rename) or when the stored row data
    is incomplete.
    """
    from mfs.store import ChunkRecord

    _require(args, "old", "new", "content")
    old = args["old"]
    new = args["new"]
    arg_hash = args.get("file_hash")

    if arg_hash:
        try:
            copied = _try_rename_without_reembed(svc, old, new, arg_hash)
            if copied is not None:
                return {"chunks": copied, "embed_ms": 0, "fast_path": True}
        except Exception:
            # Any failure in the fast path falls back to the safe
            # re-embed path below — never leaves the store half-renamed.
            pass

    for _pk, _emb, store in svc.stores():
        try:
            store.delete_by_source(old)
        except Exception:
            pass
    return op_upsert(svc, {
        "path": new,
        "content": args["content"],
        "ext": args.get("ext", ".md"),
        "file_hash": arg_hash,
        "metadata": args.get("metadata") or {},
    })


def _try_rename_without_reembed(
    svc: StashbaseStore, old: str, new: str, expected_hash: str,
) -> int | None:
    """Copy each row of ``old`` to ``new`` while keeping cached vectors.

    Returns ``None`` (caller should fall back to re-embed) when:
      - no rows for the old source exist in any collection
      - any row's stored ``file_hash`` differs from ``expected_hash``
        (content drifted; we'd embed stale text otherwise)
      - any row lacks a ``dense_vector`` we can re-insert

    Otherwise returns the number of chunks moved. Old rows are dropped
    only after every collection's copy succeeded, so a partial failure
    leaves the index unchanged.
    """
    from mfs.store import ChunkRecord

    new_parent = "/".join(new.split("/")[:-1])
    fields = [
        "id", "source", "parent_dir", "chunk_index", "start_line", "end_line",
        "chunk_text", "dense_vector", "content_type", "file_hash",
        "is_dir", "embed_status", "metadata", "account_id",
    ]
    per_store_records: list[tuple[Any, list[ChunkRecord]]] = []
    total = 0
    for _pk, _emb, store in svc.stores():
        try:
            rows = store._query_all(
                f'source == "{old}"', output_fields=fields,
            )
        except Exception:
            return None
        if not rows:
            continue
        records: list[ChunkRecord] = []
        for r in rows:
            stored_hash = r.get("file_hash", "")
            if stored_hash and stored_hash != expected_hash:
                # Content drifted since we last embedded — re-embed.
                return None
            vec = r.get("dense_vector")
            if vec is None:
                return None
            new_id = blake3(
                f"{new}:{r['start_line']}:{r['end_line']}:"
                f"{_hash_text(r.get('chunk_text', ''))}".encode(),
            ).hexdigest()[:32]
            records.append(ChunkRecord(
                id=new_id,
                source=new,
                parent_dir=new_parent,
                chunk_index=int(r.get("chunk_index", 0)),
                start_line=int(r.get("start_line", 0)),
                end_line=int(r.get("end_line", 0)),
                chunk_text=r.get("chunk_text", ""),
                dense_vector=vec,
                content_type=r.get("content_type", ""),
                file_hash=expected_hash,
                is_dir=bool(r.get("is_dir", False)),
                embed_status=r.get("embed_status", "complete"),
                metadata=r.get("metadata", {}) or {},
                account_id=r.get("account_id", "stashbase"),
            ))
        per_store_records.append((store, records))
        total += len(records)

    if total == 0:
        return None

    # All rows validated — commit the copy + drop the old source.
    for store, records in per_store_records:
        store.insert_chunks(records)
    for _pk, _emb, store in svc.stores():
        try:
            store.delete_by_source(old)
        except Exception:
            pass
    return total


def op_rename_prefix(svc: StashbaseStore, args: dict) -> dict:
    """Folder rename — move every file under ``old`` to ``new``.

    Mirrors the single-file ``op_rename`` fast path per file: when a
    file's caller-supplied ``file_hash`` still matches every stored
    row's hash, copy the rows with the new source / id and reuse the
    cached ``dense_vector`` (no embedding); only fall back to
    delete-and-reinsert for files whose content actually drifted or
    whose rows lack vectors. A whole-folder move of unchanged files
    becomes ~free instead of a full re-embed of every file.

    The fast path is skipped when one prefix nests inside the other
    (``a`` → ``a/b`` or vice versa): the trailing orphan sweep of
    ``old_prefix`` could otherwise delete the freshly-written new rows.
    That case takes the original wipe-then-reembed route.
    """
    _require(args, "old", "new")
    old_prefix = args["old"].rstrip("/") + "/"
    new_prefix = args["new"].rstrip("/") + "/"
    files = args.get("files", [])
    nested = new_prefix.startswith(old_prefix) or old_prefix.startswith(new_prefix)

    total = 0
    fast_files = 0

    if nested:
        # Safe original path: clear the old prefix first, then re-embed.
        for _pk, _emb, store in svc.stores():
            try:
                store.delete_by_prefix(old_prefix)
            except Exception:
                pass
        for f in files:
            res = op_upsert(svc, {
                "path": f["path"], "content": f["content"], "ext": f.get("ext", ".md"),
                "file_hash": f.get("file_hash"),
            })
            total += int(res.get("chunks", 0))
        return {"files": len(files), "chunks": total, "fast_path_files": 0}

    for f in files:
        new_path = f["path"]
        rel = new_path[len(new_prefix):] if new_path.startswith(new_prefix) else new_path.split("/")[-1]
        old_path = old_prefix + rel
        arg_hash = f.get("file_hash")
        copied = None
        if arg_hash:
            try:
                copied = _try_rename_without_reembed(svc, old_path, new_path, arg_hash)
            except Exception:
                # Any fast-path failure drops to the safe re-embed below —
                # never leaves the store half-renamed for this file.
                copied = None
        if copied is not None:
            total += copied
            fast_files += 1
            continue
        # Per-file fallback: drop the stale old rows, then re-embed.
        for _pk, _emb, store in svc.stores():
            try:
                store.delete_by_source(old_path)
            except Exception:
                pass
        res = op_upsert(svc, {
            "path": new_path, "content": f["content"], "ext": f.get("ext", ".md"),
            "file_hash": arg_hash,
        })
        total += int(res.get("chunks", 0))

    # Sweep any rows still under the old prefix — old files that weren't in
    # the move list (e.g. excluded by reserved-file rules upstream).
    for _pk, _emb, store in svc.stores():
        try:
            store.delete_by_prefix(old_prefix)
        except Exception:
            pass
    return {"files": len(files), "chunks": total, "fast_path_files": fast_files}


def op_delete_prefix(svc: StashbaseStore, args: dict) -> dict:
    """Drop every chunk row whose source starts with ``prefix/`` from
    every collection."""
    _require(args, "prefix")
    prefix = args["prefix"].rstrip("/") + "/"
    removed = 0
    for _pk, _emb, store in svc.stores():
        try:
            removed += int(store.delete_by_prefix(prefix))
        except Exception:
            pass
    return {"removed": removed}


def op_search(svc: StashbaseStore, args: dict) -> dict:
    """Hybrid search in the single collection, optionally scoped to one
    ``space``. MFS's ``hybrid_search`` already does dense + BM25 + RRF
    inside the collection, so its order is the final order — no
    second-pass fusion. ``top_k`` bounded to [1, 200]."""
    _require(args, "query")
    query = args["query"].strip()
    space = args.get("space")
    explicit_prefix = args.get("path_prefix")
    top_k_raw = int(args.get("top_k", 8))
    top_k = max(1, min(200, top_k_raw))
    if not query:
        return {"hits": []}
    stores = svc.stores()
    if not stores:
        return {"hits": []}
    _pk, embedder, store = stores[0]

    # Path filter: MFS's _make_filter applies `source like "<prefix>%"`.
    # `path_prefix` wins when provided — it's more specific than the
    # space-derived prefix (e.g. "cs183b/transcripts/" vs "cs183b/").
    # Otherwise fall back to space-only scoping; both omitted = whole
    # library.
    if explicit_prefix:
        path_filter = explicit_prefix if explicit_prefix.endswith("/") else explicit_prefix.rstrip("/") + "/"
    elif space:
        path_filter = space.rstrip("/") + "/"
    else:
        path_filter = None

    try:
        if store.is_empty():
            return {"hits": []}
        qvec = embedder.embed([query])[0]
        hits = store.hybrid_search(qvec, query, path_filter=path_filter, top_k=top_k)
    except Exception as exc:
        sys.stderr.write(f"[stashbase] search store failed: {exc}\n")
        return {"hits": []}

    out = []
    for h in hits:
        if h.is_dir:
            continue
        out.append({
            "path": h.source,
            "chunk_index": h.chunk_index,
            "chunk_text": h.chunk_text,
            "start_line": h.start_line,
            "end_line": h.end_line,
            "content_type": h.content_type,
            "score": h.score,
            "metadata": h.metadata or {},
        })
    return {"hits": out}


def op_list(svc: StashbaseStore, args: dict) -> dict:
    """Return ``{path: file_hash}`` for every file with rows across the
    **active** collections, optionally scoped to one ``space``. A file
    in multiple active collections collapses to one entry (last write
    wins — shouldn't matter in practice, files live in exactly one
    collection in the steady state).

    Archive collections are excluded: if an old collection still has
    rows for a file the reconcile path would see ``hash matches`` and
    skip re-embedding under the current embedder, leaving the file
    unsearchable. See build-map 04-indexing #03."""
    space = args.get("space")
    prefix = (space.rstrip("/") + "/") if space else ""
    out: dict[str, str] = {}
    for _pk, _emb, store in svc.stores():
        try:
            files = store.get_indexed_files(prefix)
        except Exception:
            continue
        for src, fh in files.items():
            out[src] = fh
    return {"files": out}


def _is_reserved_metadata_path(rel_local: str) -> bool:
    """True for hidden sidecar metadata plus the legacy space-root file.

    Do not match by basename alone: a user note at
    ``research/file-metadata.md`` is normal content and should be indexed.
    """
    parts = [p for p in rel_local.split("/") if p]
    if len(parts) >= 2 and parts[0] == ".stashbase" and parts[1] == "file-metadata.md":
        return True
    return len(parts) == 1 and parts[0] == "file-metadata.md"


def _make_scanner():
    """Configure an MFS Scanner for our space layout.

    Two tweaks over MFS defaults:
      - `.html` / `.htm` aren't in `INDEXED_EXTENSIONS`, inject via
        `IndexingConfig.include_extensions`
      - generated/dependency/source-control dirs are skipped so a root
        pointed at a code checkout does not flood the indexer
      - `compute_file_hash` is patched to BLAKE3 (see _patch_scanner_blake3)
    """
    from mfs.config import Config, IndexingConfig
    from mfs.ingest.scanner import Scanner
    _patch_scanner_blake3()
    config = Config(indexing=IndexingConfig(include_extensions=list(_RULES["include_extensions"])))
    extra = []
    for name in sorted(_RULES["excluded_dirs"]):
        extra.append(name)
        extra.append(f"{name}/")
    return Scanner(config, extra_excludes=extra)


def _walk_disk(root: Path, rel_prefix: str = "") -> dict:
    """Walk ``root`` returning ``{rel_path: FileInfo}`` for indexable
    files. ``rel_path`` is prefixed with ``rel_prefix`` so callers can
    return kb-root-relative paths even when scanning a single space's
    subdir.

    Filters out anything inside a ``<stem>_files/`` bundle dir and any
    0-byte note — same rules as the sidebar's tree walk.
    """
    scanner = _make_scanner()
    raw = []
    for f in scanner.scan([root]):
        try:
            rel_local = str(f.path.relative_to(root)).replace(os.sep, "/")
        except ValueError:
            continue
        full_rel = f"{rel_prefix}/{rel_local}" if rel_prefix else rel_local
        raw.append((full_rel, rel_local, f))

    # Note-stem detection runs against the local-relative path; only
    # `.md` / `.html` files in the SAME directory can produce a bundle.
    note_stems = set()
    for _full, rel_local, _f in raw:
        base = rel_local.rsplit("/", 1)[-1]
        for ext in (".md", ".markdown", ".html", ".htm"):
            if base.lower().endswith(ext):
                parent = rel_local[: -len(base)]
                stem = base[: -len(ext)]
                note_stems.add(parent + stem)
                break

    on_disk = {}
    for full_rel, rel_local, f in raw:
        if _has_excluded_segment(rel_local):
            continue
        if _under_bundle(rel_local, note_stems):
            continue
        # Reserved agent metadata files must never be indexed (their YAML
        # / 目录 prose would surface as bogus hits). Mirrors the Node-side
        # guard in `indexer.mfs.ts:upsertFile`. See build-map 02-storage.
        if _is_reserved_metadata_path(rel_local):
            continue
        try:
            size = f.path.stat().st_size
            if size == 0 or size > _RULES["max_indexable_bytes"]:
                continue
        except OSError:
            continue
        on_disk[full_rel] = f
    return on_disk


def _has_excluded_segment(rel: str) -> bool:
    return any(seg in _RULES["excluded_dirs"] for seg in rel.split("/") if seg)


def _under_bundle(rel: str, note_stems: set) -> bool:
    """True if ``rel`` lives inside a ``<stem>_files/`` bundle whose
    sibling ``<stem>.{md,html}`` we know about."""
    segments = rel.split("/")
    for i, seg in enumerate(segments[:-1]):
        if not seg.endswith("_files"):
            continue
        stem = seg[: -len("_files")]
        parent = "/".join(segments[:i])
        candidate = (parent + "/" + stem) if parent else stem
        if candidate in note_stems:
            return True
    return False


def _walk_for_scope(svc: StashbaseStore, space: str | None) -> dict:
    """Pick the right disk walk for ``status`` / ``scan_diff``:
    - ``space`` given → walk just ``<kb_root>/<space>`` with the space
      name as rel-prefix so returned paths are kb-root-relative.
    - ``space`` omitted → walk every bound space; skip unbound
      directories so we don't count files no collection is responsible
      for.
    """
    if space is not None:
        root = svc.kb_root / space
        return _walk_disk(root, rel_prefix=space)
    out: dict = {}
    for sp in svc.bound_spaces():
        root = svc.kb_root / sp
        out.update(_walk_disk(root, rel_prefix=sp))
    return out


def op_scan_diff(svc: StashbaseStore, args: dict) -> dict:
    """Content-hash diff: catches external edits the name-set diff misses.

    ``args.space`` optional; whole library if omitted.

    Pairs deleted+added entries with matching content hash and reports
    them as ``renamed`` instead so the Node syncIndex can route them
    through ``op_rename`` (which skips re-embedding when the content is
    unchanged — see fast-path in ``op_rename``). Only 1:1 hash matches
    are treated as renames; ambiguous N:M cases stay in
    deleted/added so the user-visible diff doesn't silently mis-attribute
    moves.
    """
    scanner = _make_scanner()
    space = args.get("space")
    on_disk = _walk_for_scope(svc, space)
    # Aggregate indexed files across **active** collections, scoped if
    # needed. Archive collections are excluded so a stale row from the
    # previous embedder doesn't mark a file as "indexed" — the
    # reconcile loop would then skip re-embedding it under the current
    # embedder and the file would stay unsearchable.
    indexed: dict[str, str] = {}
    prefix = (space.rstrip("/") + "/") if space else ""
    for _pk, _emb, store in svc.stores():
        try:
            for src, fh in store.get_indexed_files(prefix).items():
                indexed[src] = fh
        except Exception:
            continue

    added, modified, unchanged = [], [], []
    added_hashes: dict[str, str] = {}
    for rel, f in on_disk.items():
        if rel not in indexed:
            added.append(rel)
            try:
                added_hashes[rel] = scanner.compute_file_hash(f.path)
            except OSError:
                # If we can't hash a fresh file, it stays in `added`
                # without a hash entry — rename detection just won't
                # pair it. Reconcile will index it normally.
                pass
            continue
        try:
            disk_hash = scanner.compute_file_hash(f.path)
        except OSError:
            unchanged.append(rel)
            continue
        if disk_hash != indexed[rel]:
            modified.append(rel)
        else:
            unchanged.append(rel)
    deleted = [rel for rel in indexed if rel not in on_disk]

    # Pair adds and deletes that share a content hash. Only handle 1:1
    # matches — if two added files share a hash with two deleted files,
    # we can't tell who renamed to who, so leave them in the buckets.
    renamed: list[dict[str, str]] = []
    if added_hashes and deleted:
        deleted_hashes: dict[str, str] = {}
        for rel in deleted:
            deleted_hashes[rel] = indexed[rel]
        added_by_hash: dict[str, list[str]] = {}
        for rel, h in added_hashes.items():
            added_by_hash.setdefault(h, []).append(rel)
        deleted_by_hash: dict[str, list[str]] = {}
        for rel, h in deleted_hashes.items():
            deleted_by_hash.setdefault(h, []).append(rel)
        consumed_added: set[str] = set()
        consumed_deleted: set[str] = set()
        for h, adds in added_by_hash.items():
            dels = deleted_by_hash.get(h, [])
            if len(adds) == 1 and len(dels) == 1:
                renamed.append({"old": dels[0], "new": adds[0], "file_hash": h})
                consumed_added.add(adds[0])
                consumed_deleted.add(dels[0])
        if consumed_added:
            added = [a for a in added if a not in consumed_added]
        if consumed_deleted:
            deleted = [d for d in deleted if d not in consumed_deleted]

    return {
        "added": added,
        "modified": modified,
        "deleted": deleted,
        "renamed": renamed,
        "unchanged_count": len(unchanged),
    }


def op_status(svc: StashbaseStore, args: dict) -> dict:
    """Name-only diff. ``args.space`` optional; whole library if omitted.

    Active-store-only (mirrors ``op_scan_diff``): archive rows must not
    inflate the "indexed" set or the UI would report green when files
    actually still need to be re-embedded under the current embedder."""
    space = args.get("space")
    on_disk = set(_walk_for_scope(svc, space).keys())
    prefix = (space.rstrip("/") + "/") if space else ""
    indexed: set[str] = set()
    for _pk, _emb, store in svc.stores():
        try:
            indexed.update(store.get_indexed_files(prefix).keys())
        except Exception:
            continue

    pending = sorted(on_disk - indexed)
    orphaned_count = len(indexed - on_disk)
    orphaned = sorted(indexed - on_disk)

    return {
        "total": len(on_disk),
        "indexed": len(on_disk & indexed),
        "pending_count": len(pending),
        "pending": pending,
        "orphaned_count": orphaned_count,
        "orphaned": orphaned,
        "up_to_date": len(pending) == 0 and orphaned_count == 0,
    }


def op_export_space(svc: StashbaseStore, args: dict) -> dict:
    """Dump the space's embeddings to a pure vector cache at ``out_path``.

    A snapshot (v3) is just ``{text_hash, dense_vector}`` — the only
    artifact that's expensive to recompute. Chunk text, line ranges and
    metadata are deliberately NOT stored: they're re-derivable from the
    source files the snapshot travels with (it lives at
    ``<space>/.stashbase/snapshot.parquet``). The human-readable
    descriptor (embedder identity, counts, timestamp) is written
    separately by the Node side as a sibling ``snapshot.meta.json``.

    Only the space's currently-bound (active) collection is exported —
    one embedder, so the content-addressed ``text_hash`` key is
    unambiguous. ``text_hash = BLAKE3(chunk_text)`` is exactly what
    ``op_upsert`` recomputes on import, so identical text reuses its
    vector regardless of which file / line it now lives at, and
    identical chunks collapse to a single row (natural dedup).
    """
    import pyarrow as pa
    import pyarrow.parquet as pq

    _require(args, "space", "out_path")
    space = args["space"].strip().strip("/")
    out_path = Path(args["out_path"]).expanduser().resolve()
    out_path.parent.mkdir(parents=True, exist_ok=True)

    if space not in svc.bound_spaces():
        raise RuntimeError(f"space '{space}' is not bound; call bind_space first")
    embedder, store, dim = svc.require_current()
    provider = "openai"

    rows = store._query_all(
        f'source like "{space}/%"',
        output_fields=["chunk_text", "dense_vector"],
    )

    # Content-address by BLAKE3(chunk_text); identical chunks dedup to one.
    by_hash: dict[str, list[float]] = {}
    for row in rows:
        vec = row.get("dense_vector")
        if vec is None:
            continue
        h = _hash_text(row["chunk_text"])
        if h not in by_hash:
            by_hash[h] = list(vec)

    columns = {
        "text_hash": list(by_hash.keys()),
        "dense_vector": list(by_hash.values()),
    }
    schema = pa.schema([
        pa.field("text_hash", pa.string()),
        pa.field("dense_vector", pa.list_(pa.float32())),
    ])
    table = pa.table(columns, schema=schema)
    pq.write_table(table, out_path, compression="zstd")

    return {
        "path": str(out_path),
        "vectors": len(by_hash),
        "chunks": len(rows),
        "version": SNAPSHOT_VERSION,
        "embedder": {
            "provider": provider,
            "model": getattr(embedder, "model_name", None),
            "dim": dim,
        },
    }


def op_load_vector_cache(svc: StashbaseStore, args: dict) -> dict:
    """Load a v3 snapshot's ``{text_hash, dense_vector}`` into the space's
    transient embedding cache. The subsequent (Node-driven) re-index
    re-chunks the source files and lands in ``op_upsert``, which reuses a
    cached vector for any chunk whose ``BLAKE3(chunk_text)`` hits — only
    the misses get embedded.

    Validates the snapshot's declared embedder (passed by Node from
    ``snapshot.meta.json``) against the space's bound embedder. A
    mismatch (different provider / model / dim → different vectors) means
    the cache is unusable: nothing is loaded and ``mismatch`` is returned
    so the caller surfaces a "switch embedder / re-export" warning and
    falls back to a full embed. The per-vector dim guard is a final
    backstop against a corrupt / wrong-dim file.
    """
    import pyarrow.parquet as pq

    _require(args, "space", "in_path")
    space = args["space"].strip().strip("/")
    in_path = Path(args["in_path"]).expanduser().resolve()
    if not in_path.exists():
        raise RuntimeError(f"snapshot not found: {in_path}")

    if space not in svc.bound_spaces():
        raise RuntimeError(f"space '{space}' is not bound; call bind_space first")
    embedder, _store, bound_dim = svc.require_current()
    provider = "openai"
    bound_model = getattr(embedder, "model_name", None)

    declared = args.get("embedder") or {}
    if declared and (
        declared.get("provider") != provider
        or int(declared.get("dim") or 0) != bound_dim
        or (declared.get("model") and declared.get("model") != bound_model)
    ):
        return {
            "loaded": 0,
            "mismatch": True,
            "expected": {"provider": provider, "model": bound_model, "dim": bound_dim},
            "got": declared,
        }

    table = pq.read_table(in_path, columns=["text_hash", "dense_vector"])
    hashes = table.column("text_hash").to_pylist()
    vecs = table.column("dense_vector").to_pylist()
    cache: dict[str, list[float]] = {}
    for h, v in zip(hashes, vecs):
        if h and v is not None and len(v) == bound_dim:
            cache[h] = list(v)
    svc.set_vector_cache(space, cache)
    return {"loaded": len(cache), "mismatch": False}


def op_clear_vector_cache(svc: StashbaseStore, args: dict) -> dict:
    """Drop the space's transient embedding cache once the import-time
    re-index has drained. Always safe to call (no-op if none loaded)."""
    _require(args, "space")
    space = args["space"].strip().strip("/")
    return {"cleared": svc.clear_vector_cache(space)}


def op_close_store(svc: StashbaseStore, _args: dict) -> dict:
    """Release the Milvus Lite flock so the server can move / wipe the
    DB. Next ``bind_space`` reopens lazily."""
    svc.close_all()
    return {}


OPS = {
    "bind_space": op_bind_space,
    "unbind_space": op_unbind_space,
    "upsert": op_upsert,
    "delete": op_delete,
    "delete_prefix": op_delete_prefix,
    "rename": op_rename,
    "rename_prefix": op_rename_prefix,
    "search": op_search,
    "scan_diff": op_scan_diff,
    "status": op_status,
    "list": op_list,
    "export_space": op_export_space,
    "load_vector_cache": op_load_vector_cache,
    "clear_vector_cache": op_clear_vector_cache,
    "close_store": op_close_store,
    "set_rules": op_set_rules,
}


def _emit(obj: dict) -> None:
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()


def main() -> int:
    import atexit
    import signal

    parser = argparse.ArgumentParser(description="StashBase MFS sidecar daemon")
    parser.add_argument("--kb-root", required=True,
                        help="Absolute path of the StashBase library root; the daemon "
                             "owns one Milvus DB at <kb_root>/.stashbase/store/milvus.db")
    parsed, _unknown = parser.parse_known_args()

    # Single-instance guard (data-layer §8.6 I3): hold an exclusive flock
    # on a sidecar lock file for the whole process lifetime. Milvus Lite
    # has its own LOCK, but the loser of that race doesn't fail — it
    # half-opens (reads fine, writes silently lost). Failing fast here
    # turns "second daemon on the same kbRoot" into an explicit startup
    # error the Node side can surface. The flock dies with the process,
    # so a dirty exit can never strand it. Best-effort: fcntl missing
    # (Windows) or fs oddities skip the guard rather than block startup.
    daemon_lock = None  # noqa: F841 — held open for the process lifetime
    try:
        import fcntl
        lock_path = Path(parsed.kb_root) / ".stashbase" / "store" / "daemon.lock"
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        daemon_lock = open(lock_path, "w")
        try:
            fcntl.flock(daemon_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            daemon_lock.write(str(os.getpid()))
            daemon_lock.flush()
        except BlockingIOError:
            _emit({
                "event": "error",
                "phase": "daemon_lock",
                "error": f"another stashbase daemon already holds {lock_path} — "
                         "refusing to run a second instance against the same KB",
            })
            return 1
    except Exception:
        daemon_lock = None  # unguarded platforms: proceed as before

    try:
        svc = StashbaseStore(parsed.kb_root)
    except Exception as exc:
        _emit({"event": "error", "phase": "store_init", "error": str(exc)})
        return 1

    # Release every Milvus Lite flock cleanly on any exit path. Without
    # this, killing the Node parent leaves the locks held until the kernel
    # reaps the FDs, and the next StashBase launch gets a "DataDirLocked"
    # error from MilvusLite.
    def _cleanup_store(*_):
        try:
            svc.close_all()
        except Exception:
            pass
    atexit.register(_cleanup_store)
    for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
        try:
            signal.signal(sig, lambda *_: (_cleanup_store(), sys.exit(0)))
        except (ValueError, OSError):
            pass

    _emit({"event": "ready", "kb_root": str(svc.kb_root), "db": str(svc._db_path)})

    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
            req_id = req.get("id")
            op = req["op"]
            args = req.get("args", {}) or {}
        except (ValueError, KeyError) as exc:
            _emit({"id": None, "ok": False, "error": f"bad request: {exc}"})
            continue

        try:
            handler = OPS.get(op)
            if handler is None:
                _emit({"id": req_id, "ok": False, "error": f"unknown op: {op}", "op": op})
                continue
            result = handler(svc, args)
            _emit({"id": req_id, "ok": True, "result": result})
        except (KeyError, ValueError) as exc:
            sys.stderr.write(f"[stashbase] bad args for {op}: {exc}\n")
            _emit({"id": req_id, "ok": False, "error": f"bad args for {op}: {exc}", "op": op})
        except Exception as exc:
            sys.stderr.write(traceback.format_exc())
            sys.stderr.flush()
            _emit({"id": req_id, "ok": False, "error": str(exc), "op": op})

    return 0


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