from __future__ import annotations

import hashlib
import json
import logging
import os
import random
import sys
import time
from core_memory.provider_config import resolve_embedding_config, normalize_provider, default_embedding_model as provider_default_embedding_model
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from .lifecycle import enqueue_semantic_rebuild
from .normalize import tokenize
from .vector_backend import create_vector_backend
from .visible_corpus import build_visible_corpus

logger = logging.getLogger(__name__)
_faiss_warning_emitted = False
_startup_check_done = False
_startup_keyless_hint_emitted = False

SEMANTIC_MODE_REQUIRED = "required"
SEMANTIC_MODE_DEGRADED_ALLOWED = "degraded_allowed"

VECTOR_BACKEND_LOCAL_FAISS = "local-faiss"
VECTOR_BACKEND_QDRANT = "qdrant"
VECTOR_BACKEND_PGVECTOR = "pgvector"
VECTOR_BACKEND_CHROMADB = "chromadb"
_EXTERNAL_VECTOR_BACKENDS = {VECTOR_BACKEND_QDRANT, VECTOR_BACKEND_PGVECTOR, VECTOR_BACKEND_CHROMADB}


def _normalize_vector_backend(value: str | None) -> str:
    # Default is qdrant (embedded, zero-ops). Use CORE_MEMORY_VECTOR_BACKEND=local-faiss to opt back.
    v = str(value or VECTOR_BACKEND_QDRANT).strip().lower().replace("_", "-")
    if v in {"", "auto", "qdrant"}:
        return VECTOR_BACKEND_QDRANT
    if v in {"local", "faiss", "local-faiss"}:
        return VECTOR_BACKEND_LOCAL_FAISS
    if v in {"pgvector", "postgres", "postgresql"}:
        return VECTOR_BACKEND_PGVECTOR
    if v in {"chromadb", "chroma"}:
        return VECTOR_BACKEND_CHROMADB
    return VECTOR_BACKEND_QDRANT


_faiss_deprecation_emitted = False


def _configured_vector_backend() -> str:
    global _faiss_deprecation_emitted
    result = _normalize_vector_backend(os.environ.get("CORE_MEMORY_VECTOR_BACKEND"))
    if result == VECTOR_BACKEND_LOCAL_FAISS and not _faiss_deprecation_emitted:
        _faiss_deprecation_emitted = True
        logger.warning(
            "core-memory: FAISS backend is deprecated; set CORE_MEMORY_VECTOR_BACKEND=qdrant. "
            "FAISS will be removed in the next major version."
        )
    return result


def _qdrant_external_embeddings_enabled() -> bool:
    """When True, embed Qdrant vectors with the configured external provider
    (e.g. OpenAI text-embedding-3-large) instead of Qdrant's built-in FastEmbed.

    FastEmbed's default model is small (~384-dim bge) and discriminates poorly on
    conversational text, capping retrieval recall regardless of downstream graph
    expansion. Opt in via CORE_MEMORY_QDRANT_EXTERNAL_EMBEDDINGS=1 with a provider
    key (OPENAI_API_KEY / GEMINI_API_KEY) configured. Default off preserves the
    zero-API-key FastEmbed behaviour.
    """
    val = str(os.environ.get("CORE_MEMORY_QDRANT_EXTERNAL_EMBEDDINGS") or "").strip().lower()
    return val in {"1", "true", "yes", "on"}


def _vector_collection_name(root: Path) -> str:
    import hashlib as _hashlib

    prefix = str(os.environ.get("CORE_MEMORY_VECTOR_COLLECTION") or "core_memory_beads").strip() or "core_memory_beads"
    suffix = _hashlib.sha1(str(root).encode("utf-8")).hexdigest()[:10]
    name = f"{prefix}_{suffix}"
    # Qdrant collections have fixed vector params (dimension/distance). FastEmbed
    # (~384-dim) and external providers (e.g. OpenAI text-embedding-3-large,
    # 3072-dim) are incompatible, so give the external-embedding mode its own
    # collection. This makes switching CORE_MEMORY_QDRANT_EXTERNAL_EMBEDDINGS on
    # (or changing the embedding model) target a fresh, correctly-dimensioned
    # collection instead of upserting into the existing FastEmbed one — no
    # dimension-mismatch failures, no manual collection drop, and switching back
    # is non-destructive. The model is folded in so a later model change (and its
    # new dimension) also lands in a distinct collection.
    if _qdrant_external_embeddings_enabled():
        model = str(os.environ.get("CORE_MEMORY_EMBEDDINGS_MODEL") or "ext").strip() or "ext"
        model_tag = _hashlib.sha1(model.encode("utf-8")).hexdigest()[:8]
        name = f"{name}_ext_{model_tag}"
    return name


def _create_external_backend(*, root: Path, backend: str, dimension: int):
    collection = _vector_collection_name(root)
    if backend == VECTOR_BACKEND_QDRANT:
        url = os.environ.get("CORE_MEMORY_QDRANT_URL") or None
        path = os.environ.get("CORE_MEMORY_QDRANT_PATH") or str(root / ".beads" / "qdrant")
        return create_vector_backend(
            "qdrant",
            collection_name=collection,
            url=url,
            path=None if url else path,
            dimensions=int(max(0, dimension)),
        )
    if backend == VECTOR_BACKEND_PGVECTOR:
        table_name = str(os.environ.get("CORE_MEMORY_PGVECTOR_TABLE") or collection).replace("-", "_")
        return create_vector_backend(
            "pgvector",
            table_name=table_name,
            dimensions=int(max(1, dimension)),
            dsn=os.environ.get("CORE_MEMORY_PG_DSN"),
        )
    if backend == VECTOR_BACKEND_CHROMADB:
        persist_dir = str(os.environ.get("CORE_MEMORY_CHROMADB_PERSIST_DIR") or (root / ".beads" / "semantic" / "chromadb"))
        return create_vector_backend(
            "chromadb",
            collection_name=collection,
            persist_directory=persist_dir,
        )
    raise ValueError(f"unsupported_external_backend:{backend}")


def _backend_deployment_profile(backend: str) -> dict[str, Any]:
    b = str(backend or "").strip().lower()
    if b.startswith("faiss"):
        return {
            "deployment_profile": "single_process_single_writer",
            "multi_worker_safe": False,
            "concurrency_warning": "FAISS/local index is development-oriented and not recommended for concurrent multi-worker writes.",
        }
    if b.startswith("chromadb") or b.startswith("chroma"):
        return {
            "deployment_profile": "single_process_single_writer",
            "multi_worker_safe": False,
            "concurrency_warning": "ChromaDB local persistence is best treated as single-writer unless externally coordinated.",
        }
    if b.startswith("qdrant") or b.startswith("pgvector") or b.startswith("postgres"):
        return {
            "deployment_profile": "distributed_safe",
            "multi_worker_safe": True,
            "concurrency_warning": "",
        }
    if b in {"", "lexical", "not_built"}:
        return {
            "deployment_profile": "lexical_only",
            "multi_worker_safe": True,
            "concurrency_warning": "No semantic backend is currently active; query-based anchors may fail closed in required mode.",
        }
    return {
        "deployment_profile": "unknown",
        "multi_worker_safe": False,
        "concurrency_warning": "Unknown backend profile; verify multi-worker safety before production use.",
    }


KEYLESS_SEMANTIC_HINT = (
    "core-memory: no embedding provider configured — running in lexical-only mode.\n"
    "Set OPENAI_API_KEY, GEMINI_API_KEY, or GOOGLE_API_KEY for full semantic recall.\n"
    "Run `core-memory semantic doctor` to diagnose."
)


def _explicit_semantic_mode_value(mode: str | None = None) -> str:
    explicit = str(mode or "").strip().lower()
    if explicit:
        return explicit
    return str(os.environ.get("CORE_MEMORY_CANONICAL_SEMANTIC_MODE") or "").strip().lower()


def _normalize_semantic_mode(mode: str | None) -> str:
    explicit = str(mode or "").strip().lower()
    if explicit:
        m = explicit
    else:
        env_mode = str(os.environ.get("CORE_MEMORY_CANONICAL_SEMANTIC_MODE") or "").strip().lower()
        if env_mode:
            m = env_mode
        elif _has_semantic_provider_signal():
            _auto_configure_embedding_provider_from_keys()
            m = SEMANTIC_MODE_REQUIRED
        else:
            os.environ.setdefault("CORE_MEMORY_CANONICAL_SEMANTIC_MODE", SEMANTIC_MODE_DEGRADED_ALLOWED)
            m = SEMANTIC_MODE_DEGRADED_ALLOWED
    if m not in {SEMANTIC_MODE_REQUIRED, SEMANTIC_MODE_DEGRADED_ALLOWED}:
        return SEMANTIC_MODE_REQUIRED
    return m


def _semantic_provider_detection() -> dict[str, Any]:
    cfg = resolve_embedding_config()
    provider = str(cfg.provider or "").strip().lower() or None
    if provider:
        reason = "provider-neutral embedding config detected"
        if cfg.source == "OPENAI_API_KEY" and str(os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") or "").strip():
            reason = "OPENAI_API_KEY takes precedence over Gemini/Google keys when both are set"
        return {
            "provider": provider,
            "selected_provider": normalize_provider(provider),
            "configured_provider": cfg.provider,
            "source": cfg.source,
            "key": None,
            "base_url_configured": bool(cfg.base_url),
            "api_key_configured": bool(cfg.api_key),
            "reason": reason,
            "priority": [
                "CORE_MEMORY_EMBEDDINGS_PROVIDER",
                "CORE_MEMORY_EMBEDDINGS_BASE_URL",
                "CORE_MEMORY_EMBEDDINGS_API_KEY",
                "OPENAI_API_KEY",
                "OPENROUTER_API_KEY",
                "GEMINI_API_KEY",
                "GOOGLE_API_KEY",
            ],
        }
    return {
        "provider": None,
        "selected_provider": None,
        "configured_provider": None,
        "source": None,
        "key": None,
        "reason": "no provider-neutral embedding config or supported API key detected",
        "priority": ["CORE_MEMORY_EMBEDDINGS_PROVIDER", "CORE_MEMORY_EMBEDDINGS_BASE_URL", "OPENAI_API_KEY", "GEMINI_API_KEY", "GOOGLE_API_KEY"],
    }


def _auto_detect_embedding_provider_from_keys() -> str:
    detected = _semantic_provider_detection()
    return str(detected.get("provider") or "")


def _auto_configure_embedding_provider_from_keys() -> str:
    configured = str(os.environ.get("CORE_MEMORY_EMBEDDINGS_PROVIDER") or "").strip().lower()
    if configured:
        return configured
    detected = str(_semantic_provider_detection().get("provider") or "")
    if detected:
        os.environ.setdefault("CORE_MEMORY_EMBEDDINGS_PROVIDER", detected)
    return detected


def _default_embedding_model(provider: str) -> str:
    return provider_default_embedding_model(provider) or "gemini-embedding-001"


def _emit_keyless_semantic_hint_once() -> None:
    global _startup_keyless_hint_emitted
    if _startup_keyless_hint_emitted:
        return
    _startup_keyless_hint_emitted = True
    print(KEYLESS_SEMANTIC_HINT, file=sys.stderr)


def _has_semantic_provider_signal() -> bool:
    configured_provider = str(os.environ.get("CORE_MEMORY_EMBEDDINGS_PROVIDER") or "").strip().lower()
    if configured_provider:
        return True
    if _auto_detect_embedding_provider_from_keys():
        return True
    explicit_backend = str(os.environ.get("CORE_MEMORY_VECTOR_BACKEND") or "").strip()
    if explicit_backend and _normalize_vector_backend(explicit_backend) in _EXTERNAL_VECTOR_BACKENDS:
        return True
    return False


def _check_semantic_mode_startup() -> None:
    """Run-once startup check for semantic mode configuration.

    In required mode: raises RuntimeError if no embedding provider is usable.
    In degraded_allowed mode: logs a single startup warning, then stays silent.
    """
    global _startup_check_done
    if _startup_check_done:
        return
    _startup_check_done = True

    explicit_mode = _explicit_semantic_mode_value()
    provider_detection = _semantic_provider_detection()
    mode = _normalize_semantic_mode(os.environ.get("CORE_MEMORY_CANONICAL_SEMANTIC_MODE"))

    if mode == SEMANTIC_MODE_DEGRADED_ALLOWED:
        if not explicit_mode and not provider_detection.get("provider"):
            _emit_keyless_semantic_hint_once()
        else:
            logger.warning(
                "core-memory: semantic mode is 'degraded_allowed' — lexical-only retrieval; "
                "deterministic within and across restarts via fixed-seed hashing. "
                "Install an embedding provider for full semantic recall: pip install core-memory[semantic]"
            )
        return

    # required mode: check if an embedding provider is actually available
    configured_provider = _auto_configure_embedding_provider_from_keys() if not explicit_mode else str(os.environ.get("CORE_MEMORY_EMBEDDINGS_PROVIDER") or "").strip().lower()
    embed_cfg = resolve_embedding_config()
    has_provider = bool(
        configured_provider in {"hash", "openai", "openai-compatible", "gemini", "google"}
        or embed_cfg.api_key
        or embed_cfg.base_url
        or os.environ.get("OPENAI_API_KEY")
        or os.environ.get("GEMINI_API_KEY")
        or os.environ.get("GOOGLE_API_KEY")
    )
    has_faiss = False
    try:
        import faiss  # type: ignore  # noqa: F401
        has_faiss = True
    except ImportError:
        pass

    vector_backend = _configured_vector_backend()
    # Qdrant with FastEmbed provides its own embeddings — no external API key required.
    has_qdrant_fastembed = vector_backend == VECTOR_BACKEND_QDRANT
    # FAISS is a vector index, not an embedding provider — it still needs an external API key.
    has_external = vector_backend in _EXTERNAL_VECTOR_BACKENDS and not has_qdrant_fastembed

    if not has_provider and not has_external and not has_qdrant_fastembed:
        raise RuntimeError(
            "core-memory: semantic mode is 'required' (default) but no embedding provider is configured. "
            "Either:\n"
            "  1. Install semantic extras: pip install core-memory[semantic]  (and set OPENAI_API_KEY or GEMINI_API_KEY)\n"
            "  2. Configure an external vector backend: CORE_MEMORY_VECTOR_BACKEND=qdrant|pgvector|chromadb\n"
            "  3. Opt into lexical-only mode: CORE_MEMORY_CANONICAL_SEMANTIC_MODE=degraded_allowed"
        )


def _env_int(name: str, default: int, *, min_value: int = 0, max_value: int | None = None) -> int:
    raw = str(os.environ.get(name) or "").strip()
    try:
        out = int(raw) if raw else int(default)
    except Exception:
        out = int(default)
    if out < min_value:
        out = min_value
    if max_value is not None and out > max_value:
        out = max_value
    return int(out)


def _env_bool(name: str, default: bool) -> bool:
    raw = str(os.environ.get(name) or "").strip().lower()
    if not raw:
        return bool(default)
    if raw in {"1", "true", "yes", "on"}:
        return True
    if raw in {"0", "false", "no", "off"}:
        return False
    return bool(default)


def _semantic_build_on_read_enabled() -> bool:
    return _env_bool("CORE_MEMORY_SEMANTIC_BUILD_ON_READ", True)


def _embedding_error_status(exc: Exception) -> int | None:
    try:
        if isinstance(exc, urllib.error.HTTPError):
            return int(getattr(exc, "code", 0) or 0) or None
    except Exception:
        pass
    status = getattr(exc, "status_code", None)
    if isinstance(status, int) and status > 0:
        return status
    try:
        resp = getattr(exc, "response", None)
        if resp is not None:
            code = getattr(resp, "status_code", None)
            if isinstance(code, int) and code > 0:
                return code
    except Exception:
        pass
    return None


def _embedding_error_token(exc: Exception) -> str:
    status = _embedding_error_status(exc)
    if status is not None:
        return f"http_{status}"
    if isinstance(exc, ModuleNotFoundError):
        missing = getattr(exc, "name", None)
        if missing:
            return f"missing_module:{_sanitize_token(str(missing))}"
    name = str(exc.__class__.__name__ or "error").strip().lower()
    return _sanitize_token(name)


def _sanitize_token(value: str) -> str:
    out = []
    for ch in str(value or ""):
        if ch.isalnum() or ch in {"_", "-", "."}:
            out.append(ch.lower())
        elif ch in {" ", ":", "/", "\\", "|"}:
            out.append("_")
    compact = "".join(out).strip("_")
    return compact or "error"


def _embedding_retry_after_seconds(exc: Exception) -> float | None:
    try:
        if isinstance(exc, urllib.error.HTTPError):
            raw = str(exc.headers.get("Retry-After") or "").strip()  # type: ignore[arg-type]
            if raw:
                return max(0.0, float(raw))
    except Exception:
        pass
    try:
        resp = getattr(exc, "response", None)
        headers = getattr(resp, "headers", None)
        if headers is not None:
            raw = str(headers.get("Retry-After") or "").strip()
            if raw:
                return max(0.0, float(raw))
    except Exception:
        pass
    return None


def _embedding_retryable(exc: Exception) -> bool:
    status = _embedding_error_status(exc)
    if status in {408, 409, 425, 429, 500, 502, 503, 504}:
        return True
    msg = str(exc or "").lower()
    if any(x in msg for x in {"rate limit", "too many requests", "timeout", "temporarily unavailable"}):
        return True
    return False


def _embedding_retry_sleep(attempt: int, retry_after_seconds: float | None = None) -> None:
    base_ms = _env_int("CORE_MEMORY_EMBEDDINGS_RETRY_BASE_MS", 400, min_value=50, max_value=5000)
    jitter_ms = _env_int("CORE_MEMORY_EMBEDDINGS_RETRY_JITTER_MS", 250, min_value=0, max_value=2000)
    if retry_after_seconds is not None and retry_after_seconds >= 0:
        sleep_s = float(retry_after_seconds)
    else:
        exp = min(6, max(0, int(attempt)))
        sleep_s = float((base_ms * (2**exp)) / 1000.0)
        if jitter_ms > 0:
            sleep_s += random.uniform(0.0, float(jitter_ms) / 1000.0)
    if sleep_s > 0:
        time.sleep(sleep_s)


def semantic_doctor(root: Path) -> dict[str, Any]:
    """Operational diagnostics for canonical semantic mode."""
    manifest_file, faiss_file, rows_file, *_ = _paths(root)
    explicit_mode = _explicit_semantic_mode_value()
    provider_detected = _semantic_provider_detection()
    if explicit_mode and not os.environ.get("CORE_MEMORY_EMBEDDINGS_PROVIDER"):
        provider_detected = {
            **provider_detected,
            "selected_provider": None,
            "skipped_auto_detection": True,
            "reason": "CORE_MEMORY_CANONICAL_SEMANTIC_MODE is explicitly set; provider key auto-detection is disabled",
        }
    mode = _normalize_semantic_mode(os.environ.get("CORE_MEMORY_CANONICAL_SEMANTIC_MODE"))
    degraded_enabled = mode == SEMANTIC_MODE_DEGRADED_ALLOWED

    manifest: dict[str, Any] = {}
    if manifest_file.exists():
        try:
            manifest = json.loads(manifest_file.read_text(encoding="utf-8"))
        except Exception:
            manifest = {}

    backend = str(manifest.get("backend") or "")
    if provider_detected.get("skipped_auto_detection"):
        detected_provider = None
    else:
        detected_provider = provider_detected.get("provider") or provider_detected.get("selected_provider")
    provider = str(
        manifest.get("provider")
        or os.environ.get("CORE_MEMORY_EMBEDDINGS_PROVIDER")
        or detected_provider
        or ""
    )
    rows_count = len(_read_rows(rows_file)) if rows_file.exists() else 0

    normalized_backend = _normalize_vector_backend(backend)
    ext_backend = normalized_backend in _EXTERNAL_VECTOR_BACKENDS
    connectivity_ok = True
    connectivity_error = ""
    if ext_backend:
        connectivity_ok, connectivity_error = _external_backend_connectivity(root=root, backend=normalized_backend, manifest=manifest)

    usable_backend = bool(
        (backend.startswith("faiss") and faiss_file.exists() and rows_count > 0)
        or (ext_backend and rows_count > 0 and connectivity_ok)
    )
    profile = _backend_deployment_profile(backend or "not_built")
    concurrency_warning = str(profile.get("concurrency_warning") or "")
    if (not usable_backend) and mode == SEMANTIC_MODE_DEGRADED_ALLOWED and str(backend or "not_built") in {"", "not_built", "lexical"}:
        # In degraded-allowed lexical mode this is expected, not an operational warning.
        concurrency_warning = ""

    if usable_backend:
        next_step = "Semantic backend is ready for canonical query-based anchor lookup."
    elif mode == SEMANTIC_MODE_REQUIRED:
        next_step = "Install semantic extras and run: core-memory graph semantic-build"
    else:
        next_step = "Degraded lexical mode is enabled; install semantic extras + run semantic-build for canonical mode."

    return {
        "ok": True,
        "mode": mode,
        "degraded_mode_enabled": degraded_enabled,
        "manifest_path": str(manifest_file),
        "manifest_exists": manifest_file.exists(),
        "backend": backend or "not_built",
        "provider": provider or "unknown",
        "provider_detected": provider_detected,
        "degraded": degraded_enabled,
        "rows_count": int(rows_count),
        "faiss_index_exists": faiss_file.exists(),
        "connectivity_checked": bool(ext_backend),
        "connectivity_ok": bool(connectivity_ok if ext_backend else True),
        "connectivity_error": str(connectivity_error or ""),
        "usable_backend": usable_backend,
        "deployment_profile": str(profile.get("deployment_profile") or "unknown"),
        "multi_worker_safe": bool(profile.get("multi_worker_safe")),
        "concurrency_warning": concurrency_warning,
        "recommended_production_backends": ["qdrant", "pgvector"],
        "next_step": next_step,
    }


def _external_backend_connectivity(*, root: Path, backend: str, manifest: dict[str, Any]) -> tuple[bool, str]:
    b = _normalize_vector_backend(backend)
    if b == VECTOR_BACKEND_QDRANT:
        base = str(os.environ.get("CORE_MEMORY_QDRANT_URL") or "http://localhost:6333").strip().rstrip("/")
        if not base:
            return False, "missing_qdrant_url"
        url = f"{base}/collections"
        try:
            req = urllib.request.Request(url, method="GET")
            with urllib.request.urlopen(req, timeout=3) as resp:  # nosec - configured endpoint
                code = int(getattr(resp, "status", 200) or 200)
            return (200 <= code < 500), ""
        except Exception as exc:
            return False, f"qdrant_unreachable:{exc}"

    if b == VECTOR_BACKEND_PGVECTOR:
        dsn = str(os.environ.get("CORE_MEMORY_PG_DSN") or "").strip()
        if not dsn:
            return False, "missing_pg_dsn"
        try:
            import psycopg  # type: ignore
        except Exception as exc:
            return False, f"pgvector_driver_missing:{exc}"
        try:
            conn = psycopg.connect(dsn, connect_timeout=3)
            try:
                cur = conn.execute("SELECT 1")
                _ = cur.fetchone()
            finally:
                conn.close()
            return True, ""
        except Exception as exc:
            return False, f"pgvector_unreachable:{exc}"

    if b == VECTOR_BACKEND_CHROMADB:
        try:
            dim = int(manifest.get("dimension") or 256)
        except Exception:
            dim = 256
        try:
            backend_obj = _create_external_backend(root=root, backend=b, dimension=max(1, dim))
            _ = backend_obj.count()
            return True, ""
        except Exception as exc:
            return False, f"chromadb_unavailable:{exc}"

    return True, ""


def semantic_unavailable_payload(*, query: str, warnings: list[str] | None = None, provider: str | None = None, detail: dict[str, Any] | None = None) -> dict[str, Any]:
    ws = list(warnings or [])
    if "semantic_backend_unavailable" not in ws:
        ws.append("semantic_backend_unavailable")
    err = {
        "code": "semantic_backend_unavailable",
        "message": "Semantic backend unavailable for anchor lookup",
    }
    if isinstance(detail, dict) and detail:
        err["detail"] = dict(detail)
    return {
        "ok": False,
        "degraded": False,
        "backend": "unavailable",
        "provider": provider,
        "query": query,
        "warnings": ws,
        "error": err,
        "results": [],
    }


def _semantic_backend_requires_real_index(*, mode: str, provider: str, vector_backend: str, texts: list[str]) -> bool:
    if _normalize_semantic_mode(mode) != SEMANTIC_MODE_REQUIRED:
        return False
    if not texts:
        return False
    if normalize_provider(provider) == "hash":
        return False
    if _normalize_vector_backend(vector_backend) in _EXTERNAL_VECTOR_BACKENDS:
        return True
    return True


def _exception_message(exc: Exception) -> str:
    msg = str(exc or "").strip()
    if msg:
        return msg
    cause = getattr(exc, "__cause__", None) or getattr(exc, "__context__", None)
    if isinstance(cause, Exception):
        cause_msg = _exception_message(cause)
        if cause_msg:
            return f"{exc.__class__.__name__}: caused_by={cause_msg}"
    return repr(exc)


def _invalid_semantic_state_detail(
    *,
    provider: str,
    model: str,
    backend: str,
    vector_backend: str,
    dim: int,
    cause: str | None = None,
    last_build_error: str | None = None,
    last_build_error_type: str | None = None,
) -> dict[str, Any]:
    out = {
        "provider": str(provider or ""),
        "model": str(model or ""),
        "backend": str(backend or ""),
        "vector_backend": str(vector_backend or ""),
        "dimension": int(dim or 0),
    }
    if cause:
        out["cause"] = str(cause)
    if last_build_error:
        out["last_build_error"] = str(last_build_error)
    if last_build_error_type:
        out["last_build_error_type"] = str(last_build_error_type)
    return out


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _parse_iso(ts: str) -> datetime | None:
    s = str(ts or "").strip()
    if not s:
        return None
    try:
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        return datetime.fromisoformat(s)
    except Exception:
        return None


def _paths(root: Path) -> tuple[Path, Path, Path, Path, Path]:
    sem = root / ".beads" / "semantic"
    return (
        sem / "manifest.json",
        sem / "index.faiss",
        sem / "rows.jsonl",
        sem / "build.lock",
        sem / "rebuild-queue.json",
    )


def _read_queue(path: Path) -> dict[str, Any]:
    default = {"queued": False, "queued_at": None, "epoch": 0, "mode": "delta"}
    if not path.exists():
        return dict(default)
    try:
        q = json.loads(path.read_text(encoding="utf-8"))
        if not isinstance(q, dict):
            return dict(default)
        out = dict(default)
        out.update(q)
        mode = str(out.get("mode") or "delta").strip().lower()
        out["mode"] = mode if mode in {"delta", "reconcile"} else "delta"
        return out
    except Exception:
        return dict(default)


def _write_queue(path: Path, q: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(q, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp.replace(path)


def _row_metadata(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "status": str(row.get("status") or ""),
        "session_id": str(row.get("session_id") or ""),
        "source_surface": str(row.get("source_surface") or ""),
        "created_at": str(row.get("created_at") or ""),
        # Delta upserts must preserve the same Qdrant payload contract as full
        # builds. The visible corpus already contains retrieval-eligible rows;
        # keep this flag present so filtered Qdrant vector search can return
        # newly added/re-embedded points before the next full rebuild.
        "retrieval_eligible": bool(row.get("retrieval_eligible", True)),
    }


def _row_metadata_changed(prev: dict[str, Any], nxt: dict[str, Any]) -> bool:
    keys = ("status", "session_id", "source_surface", "created_at", "retrieval_eligible")
    for k in keys:
        if str(prev.get(k) or "") != str(nxt.get(k) or ""):
            return True
    return False


def _deterministic_token_hash(tok: str, dim: int) -> int:
    """Fixed-seed hash for lexical-only mode. Deterministic across restarts.

    Uses MD5 with a fixed salt — not cryptographic, just consistent.
    This replaces Python's hash() which is randomized per-process.
    """
    h = hashlib.md5(b"core_memory_v1:" + tok.encode("utf-8")).digest()
    return int.from_bytes(h[:4], "little") % dim


def _hash_vectors(texts: list[str], dim: int = 256):
    import numpy as np  # type: ignore

    vecs = np.zeros((len(texts), dim), dtype="float32")
    for i, t in enumerate(texts):
        for tok in tokenize(t):
            vecs[i, _deterministic_token_hash(tok, dim)] += 1.0
    norms = np.linalg.norm(vecs, axis=1, keepdims=True)
    norms[norms == 0] = 1.0
    return vecs / norms


def _provider_vectors(texts: list[str], provider: str, model: str):
    import numpy as np  # type: ignore

    if provider in {"openai", "openai-compatible"}:
        cfg = resolve_embedding_config()
        key = cfg.api_key or os.environ.get("OPENAI_API_KEY")
        base_url = (cfg.base_url or "https://api.openai.com/v1").rstrip("/")
        if not key and not base_url.startswith(("http://localhost", "http://127.0.0.1")):
            raise RuntimeError("missing_openai_compatible_api_key")
        import json as _json
        import urllib.request as _urlreq

        rows = []
        batch_size = _env_int("CORE_MEMORY_EMBEDDINGS_BATCH_SIZE", 64, min_value=1, max_value=2048)
        max_retries = _env_int("CORE_MEMORY_EMBEDDINGS_MAX_RETRIES", 4, min_value=0, max_value=12)
        for i in range(0, len(texts), batch_size):
            batch = list(texts[i : i + batch_size])
            for attempt in range(max_retries + 1):
                try:
                    payload = {"model": model, "input": batch}
                    data = _json.dumps(payload).encode("utf-8")
                    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {key or 'local'}"}
                    req = _urlreq.Request(base_url + "/embeddings", data=data, headers=headers, method="POST")
                    with _urlreq.urlopen(req, timeout=30) as resp:  # nosec - user/provider configured endpoint
                        body = _json.loads(resp.read().decode("utf-8"))
                    data_rows = sorted(list(body.get("data") or []), key=lambda item: int(item.get("index", 0) or 0))
                    vec_rows = [list(item.get("embedding") or []) for item in data_rows]
                    if len(vec_rows) != len(batch):
                        raise RuntimeError("openai_compatible_embedding_batch_mismatch")
                    rows.extend(vec_rows)
                    break
                except Exception as exc:
                    if attempt >= max_retries or (not _embedding_retryable(exc)):
                        code = _embedding_error_token(exc)
                        raise RuntimeError(f"openai_compatible_embedding_failed:{code}") from exc
                    _embedding_retry_sleep(attempt, _embedding_retry_after_seconds(exc))
        vecs = np.array(rows, dtype="float32")
        norms = np.linalg.norm(vecs, axis=1, keepdims=True)
        norms[norms == 0] = 1.0
        return vecs / norms

    if provider in {"gemini", "google"}:
        cfg = resolve_embedding_config()
        key = cfg.api_key or os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")
        if not key:
            raise RuntimeError("missing_gemini_api_key")

        import json as _json
        import urllib.request as _urlreq

        rows = []
        max_retries = _env_int("CORE_MEMORY_EMBEDDINGS_MAX_RETRIES", 4, min_value=0, max_value=12)
        for t in texts:
            for attempt in range(max_retries + 1):
                try:
                    url = f"https://generativelanguage.googleapis.com/v1beta/models/{model}:embedContent?key={key}"
                    payload = {"content": {"parts": [{"text": t}]}}
                    data = _json.dumps(payload).encode("utf-8")
                    req = _urlreq.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
                    with _urlreq.urlopen(req, timeout=30) as resp:  # nosec - trusted provider endpoint
                        body = _json.loads(resp.read().decode("utf-8"))
                    vals = (((body or {}).get("embedding") or {}).get("values") or [])
                    if not vals:
                        raise RuntimeError("gemini_embedding_empty")
                    rows.append(vals)
                    break
                except Exception as exc:
                    if attempt >= max_retries or (not _embedding_retryable(exc)):
                        code = _embedding_error_token(exc)
                        raise RuntimeError(f"gemini_embedding_failed:{code}") from exc
                    _embedding_retry_sleep(attempt, _embedding_retry_after_seconds(exc))

        vecs = np.array(rows, dtype="float32")
        norms = np.linalg.norm(vecs, axis=1, keepdims=True)
        norms[norms == 0] = 1.0
        return vecs / norms

    raise RuntimeError(f"unsupported_provider:{provider}")


def _vector_dim(vecs: Any, *, fallback: int = 0) -> int:
    try:
        if hasattr(vecs, "shape") and len(getattr(vecs, "shape")) >= 2:
            return int(vecs.shape[1])
        rows = list(vecs or [])
        if rows and isinstance(rows[0], (list, tuple)):
            return int(len(rows[0]))
    except Exception:
        pass
    return int(fallback)


def _vector_rows(vecs: Any) -> list[list[float]]:
    if hasattr(vecs, "tolist"):
        raw = vecs.tolist()
    else:
        raw = list(vecs or [])
    out: list[list[float]] = []
    for row in raw:
        if isinstance(row, (list, tuple)):
            out.append([float(x) for x in row])
    return out


def _embed_vectors(*, texts: list[str], provider: str, model: str, hash_dim: int = 256):
    p = normalize_provider(str(provider or "").strip().lower())
    if p == "hash":
        return _hash_vectors(texts, dim=max(1, int(hash_dim)))
    return _provider_vectors(texts, provider=p, model=model)


def _rows_from_corpus(corpus: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for r in corpus:
        txt = str(r.get("semantic_text") or "")
        rows.append(
            {
                "bead_id": str(r.get("bead_id") or ""),
                "status": str(r.get("status") or ""),
                "session_id": str(r.get("session_id") or ""),
                "source_surface": str(r.get("source_surface") or ""),
                "created_at": str(r.get("created_at") or ""),
                "retrieval_eligible": bool(r.get("retrieval_eligible", True)),
                "semantic_text": txt,
                "semantic_text_hash": hashlib.sha256(txt.encode("utf-8")).hexdigest(),
            }
        )
    return rows


def _fingerprint(rows: list[dict[str, Any]]) -> str:
    h = hashlib.sha256()
    for r in sorted(rows, key=lambda x: (str(x.get("bead_id") or ""), str(x.get("status") or ""))):
        h.update(str(r.get("bead_id") or "").encode("utf-8"))
        h.update(b"|")
        h.update(str(r.get("status") or "").encode("utf-8"))
        h.update(b"|")
        h.update(str(r.get("semantic_text_hash") or "").encode("utf-8"))
        h.update(b"\n")
    return h.hexdigest()


def _write_rows(rows_file: Path, rows: list[dict[str, Any]]) -> None:
    rows_file.parent.mkdir(parents=True, exist_ok=True)
    with open(rows_file, "w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")


def _read_rows(rows_file: Path) -> list[dict[str, Any]]:
    if not rows_file.exists():
        return []
    out: list[dict[str, Any]] = []
    for ln in rows_file.read_text(encoding="utf-8").splitlines():
        ln = ln.strip()
        if not ln:
            continue
        try:
            r = json.loads(ln)
            if isinstance(r, dict):
                out.append(r)
        except Exception:
            continue
    return out


def _read_manifest(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        m = json.loads(path.read_text(encoding="utf-8"))
        return m if isinstance(m, dict) else {}
    except Exception:
        return {}


def _write_manifest(path: Path, manifest: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp.replace(path)


def _lock_info(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        if isinstance(payload, dict):
            return payload
    except Exception:
        pass
    return {}


def _acquire_build_lock(path: Path, *, stale_seconds: int = 300) -> tuple[bool, dict[str, Any]]:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {"acquired_at": _now(), "pid": os.getpid()}
    data = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8")

    # Try create-once lock.
    try:
        fd = os.open(str(path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        try:
            os.write(fd, data)
        finally:
            os.close(fd)
        return True, payload
    except FileExistsError:
        pass
    except Exception:
        return False, {"reason": "lock_create_failed"}

    # Existing lock: reclaim if stale.
    info = _lock_info(path)
    acquired_at = _parse_iso(str(info.get("acquired_at") or ""))
    is_stale = False
    if acquired_at is not None:
        age = (datetime.now(timezone.utc) - acquired_at).total_seconds()
        is_stale = age > max(5, int(stale_seconds))

    if is_stale:
        try:
            path.unlink(missing_ok=True)
        except Exception:
            return False, {"reason": "stale_lock_unlink_failed", "lock": info}
        try:
            fd = os.open(str(path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            try:
                os.write(fd, data)
            finally:
                os.close(fd)
            payload["reclaimed_stale_lock"] = True
            return True, payload
        except Exception:
            return False, {"reason": "lock_recreate_failed", "lock": info}

    return False, {"reason": "lock_held", "lock": info}


def _release_build_lock(path: Path) -> None:
    try:
        path.unlink(missing_ok=True)
    except Exception:
        pass


def build_semantic_index(root: Path) -> dict:
    _check_semantic_mode_startup()
    manifest_file, faiss_file, rows_file, build_lock, _queue_file = _paths(root)

    acquired, lock_meta = _acquire_build_lock(build_lock)
    if not acquired:
        enqueue_semantic_rebuild(root)
        return {
            "ok": False,
            "retryable": True,
            "error": {
                "code": "semantic_build_lock_held",
                "message": "semantic build lock is already held",
                "detail": lock_meta,
            },
            "queued": True,
            "lock_path": str(build_lock),
        }

    try:
        corpus = build_visible_corpus(root)
        rows = _rows_from_corpus(corpus)
        texts = [str(r.get("semantic_text") or "") for r in rows]

        provider = (_auto_configure_embedding_provider_from_keys() or "gemini").strip().lower()
        model = (os.environ.get("CORE_MEMORY_EMBEDDINGS_MODEL") or _default_embedding_model(provider)).strip()
        vector_backend = _configured_vector_backend()

        backend = "lexical"
        dim = 0
        semantic_ready = False
        last_build_error_code = ""
        last_build_error = ""
        last_build_error_type = ""
        if vector_backend in _EXTERNAL_VECTOR_BACKENDS:
            try:
                if vector_backend == VECTOR_BACKEND_QDRANT and not _qdrant_external_embeddings_enabled():
                    # Qdrant+FastEmbed generates embeddings natively — bypass _embed_vectors.
                    # Pass dimension=0 so QdrantBackend skips VectorParams collection creation;
                    # upsert_texts() uses client.add() which creates a fastembed-configured collection.
                    # Pre-creating with VectorParams(size=1536) would produce incompatible params.
                    vb = _create_external_backend(root=root, backend=vector_backend, dimension=0)
                    bead_ids = [str(r.get("bead_id") or "") for r in rows]
                    metadatas = [
                        {
                            "status": c.get("status"),
                            "session_id": c.get("session_id"),
                            "source_surface": c.get("source_surface"),
                            "created_at": c.get("created_at"),
                            "retrieval_eligible": True,
                            "topics": list(c.get("tags") or []),
                        }
                        for c in corpus
                    ]
                    vb.upsert_texts(bead_ids=bead_ids, texts=texts, metadatas=metadatas)
                    # dim is managed by FastEmbed; use sentinel to skip external-dim validation
                    dim = 1
                    provider = "fastembed"  # Mark manifest so lookup uses native Qdrant query path
                elif vector_backend == VECTOR_BACKEND_QDRANT:
                    # Qdrant with EXTERNAL embeddings: embed with the provider (e.g.
                    # OpenAI text-embedding-3-large) and upsert pre-computed vectors,
                    # creating the collection with VectorParams(size=dim). The manifest
                    # records the real provider so the query path embeds the query the
                    # same way and searches by vector (not FastEmbed query_text).
                    vecs = _embed_vectors(texts=texts, provider=provider, model=model, hash_dim=256)
                    dim = _vector_dim(vecs, fallback=256 if texts else 0)
                    vb = _create_external_backend(root=root, backend=vector_backend, dimension=dim)
                    for row, c, emb in zip(rows, corpus, _vector_rows(vecs)):
                        vb.upsert(
                            bead_id=str(row.get("bead_id") or ""),
                            embedding=emb,
                            metadata={
                                "status": c.get("status"),
                                "session_id": c.get("session_id"),
                                "source_surface": c.get("source_surface"),
                                "created_at": c.get("created_at"),
                                "retrieval_eligible": True,
                                "topics": list(c.get("tags") or []),
                            },
                        )
                else:
                    vecs = _embed_vectors(texts=texts, provider=provider, model=model, hash_dim=256)
                    dim = _vector_dim(vecs, fallback=256 if texts else 0)
                    vb = _create_external_backend(root=root, backend=vector_backend, dimension=dim)
                    for row, emb in zip(rows, _vector_rows(vecs)):
                        vb.upsert(
                            bead_id=str(row.get("bead_id") or ""),
                            embedding=emb,
                            metadata={
                                "status": row.get("status"),
                                "session_id": row.get("session_id"),
                                "source_surface": row.get("source_surface"),
                                "created_at": row.get("created_at"),
                            },
                        )
                backend = vector_backend
                semantic_ready = True
                # no local faiss artifact in external backend mode
                if faiss_file.exists():
                    try:
                        faiss_file.unlink()
                    except Exception:
                        pass
            except Exception as exc:
                last_build_error_code = _embedding_error_token(exc)
                last_build_error = _exception_message(exc)
                last_build_error_type = exc.__class__.__name__
                logger.warning(
                    "core-memory: external vector backend '%s' unavailable (%s). Falling back to lexical.",
                    vector_backend,
                    exc,
                )
                backend = "lexical"
                dim = 0
                semantic_ready = False
        else:
            try:
                import faiss  # type: ignore

                vecs = _embed_vectors(texts=texts, provider=provider, model=model, hash_dim=256)
                dim = _vector_dim(vecs, fallback=256 if texts else 0)
                if normalize_provider(provider) == "hash":
                    backend = "faiss-hash"
                else:
                    backend = f"faiss-{provider}"

                idx = faiss.IndexFlatIP(dim or 256)
                if len(texts):
                    idx.add(vecs)
                faiss_file.parent.mkdir(parents=True, exist_ok=True)
                faiss.write_index(idx, str(faiss_file))
                semantic_ready = True
            except ImportError:
                global _faiss_warning_emitted
                last_build_error_code = "faiss_import_error"
                last_build_error = "faiss-cpu and/or numpy not installed"
                last_build_error_type = "ImportError"
                if not _faiss_warning_emitted:
                    mode = _normalize_semantic_mode(os.environ.get("CORE_MEMORY_CANONICAL_SEMANTIC_MODE"))
                    mode_hint = (
                        "query-based anchor lookup may fail closed in required mode"
                        if mode == SEMANTIC_MODE_REQUIRED
                        else "query-based lookup may run in degraded lexical mode"
                    )
                    logger.warning(
                        "core-memory: faiss-cpu and/or numpy not installed. %s. "
                        "Install with: pip install core-memory[semantic]",
                        mode_hint,
                    )
                    _faiss_warning_emitted = True
                backend = "lexical"
                dim = 0
                semantic_ready = False
            except Exception as exc:
                last_build_error_code = _embedding_error_token(exc)
                last_build_error = _exception_message(exc)
                last_build_error_type = exc.__class__.__name__
                logger.warning(
                    "core-memory: local-faiss semantic build failed (%s). Falling back to lexical.",
                    exc,
                )
                backend = "lexical"
                dim = 0
                semantic_ready = False

        if _semantic_backend_requires_real_index(
            mode=os.environ.get("CORE_MEMORY_CANONICAL_SEMANTIC_MODE") or SEMANTIC_MODE_REQUIRED,
            provider=provider,
            vector_backend=vector_backend,
            texts=texts,
        ) and (not semantic_ready or int(dim or 0) <= 0 or backend == "lexical"):
            fp = _fingerprint(rows)
            prev = _read_manifest(manifest_file)
            manifest = {
                "provider": provider,
                "model": model,
                "dimension": int(dim),
                "backend": backend,
                "vector_backend": vector_backend,
                "semantic_ready": False,
                "last_build_error_code": str(last_build_error_code or "semantic_build_invalid_state"),
                "last_build_error": str(last_build_error or ""),
                "backend_version": "v9-s2",
                "corpus_fingerprint": fp,
                "built_at": _now(),
                "row_count": len(rows),
                "dirty": False,
                "last_dirty_at": prev.get("last_dirty_at"),
                "last_dirty_reason": prev.get("last_dirty_reason"),
                "last_turn_id": prev.get("last_turn_id"),
                "last_flush_tx_id": prev.get("last_flush_tx_id"),
                "visible_statuses": ["open", "candidate", "promoted", "archived"],
            }
            _write_rows(rows_file, rows)
            _write_manifest(manifest_file, manifest)
            detail = _invalid_semantic_state_detail(
                provider=provider,
                model=model,
                backend=backend,
                vector_backend=vector_backend,
                dim=int(dim or 0),
                cause=last_build_error_code or "semantic_build_invalid_state",
                last_build_error=last_build_error,
                last_build_error_type=last_build_error_type,
            )
            return {
                "ok": False,
                "retryable": True,
                "error": {
                    "code": "semantic_build_invalid_state",
                    "message": "Required semantic build did not produce a usable semantic backend",
                    "detail": detail,
                },
                "manifest": str(manifest_file),
                "faiss": str(faiss_file),
                "rows": str(rows_file),
                "lock": {"path": str(build_lock), **dict(lock_meta or {})},
            }

        _write_rows(rows_file, rows)
        fp = _fingerprint(rows)
        prev = _read_manifest(manifest_file)
        manifest = {
            "provider": provider,
            "model": model,
            "dimension": int(dim),
            "backend": backend,
            "vector_backend": vector_backend,
            "semantic_ready": bool(semantic_ready),
            "last_build_error_code": str(last_build_error_code or ""),
            "last_build_error": str(last_build_error or ""),
            "backend_version": "v9-s2",
            "corpus_fingerprint": fp,
            "built_at": _now(),
            "row_count": len(rows),
            "dirty": False,
            "last_dirty_at": prev.get("last_dirty_at"),
            "last_dirty_reason": prev.get("last_dirty_reason"),
            "last_turn_id": prev.get("last_turn_id"),
            "last_flush_tx_id": prev.get("last_flush_tx_id"),
            "visible_statuses": ["open", "candidate", "promoted", "archived"],
        }
        _write_manifest(manifest_file, manifest)

        # consume queue epoch on successful build
        q = _read_queue(_queue_file)
        q["queued"] = False
        q["queued_at"] = None
        q["mode"] = "delta"
        _write_queue(_queue_file, q)

        return {
            "ok": True,
            "backend": backend,
            "entries": len(rows),
            "manifest": str(manifest_file),
            "faiss": str(faiss_file),
            "rows": str(rows_file),
            "lock": {"path": str(build_lock), **dict(lock_meta or {})},
        }
    finally:
        _release_build_lock(build_lock)


def apply_semantic_delta(root: Path) -> dict[str, Any]:
    """Apply queued semantic index deltas from current visible corpus.

    Contract: write-path mutation updates only, no retrieval-path index mutation.
    """
    manifest_file, faiss_file, rows_file, build_lock, queue_file = _paths(root)
    q = _read_queue(queue_file)
    if not bool(q.get("queued")):
        return {"ok": True, "ran": False, "reason": "not_queued", "mode": str(q.get("mode") or "delta")}

    mode = str(q.get("mode") or "delta").strip().lower()
    if mode == "reconcile":
        return {
            "ok": False,
            "retryable": False,
            "error": {"code": "semantic_delta_requires_reconcile", "message": "queue mode is reconcile"},
            "mode": mode,
        }

    acquired, lock_meta = _acquire_build_lock(build_lock)
    if not acquired:
        enqueue_semantic_rebuild(root)
        return {
            "ok": False,
            "retryable": True,
            "error": {
                "code": "semantic_build_lock_held",
                "message": "semantic build lock is already held",
                "detail": lock_meta,
            },
            "queued": True,
            "lock_path": str(build_lock),
            "mode": mode,
        }

    try:
        vector_backend = _configured_vector_backend()
        if vector_backend not in _EXTERNAL_VECTOR_BACKENDS:
            return {
                "ok": False,
                "retryable": False,
                "error": {
                    "code": "delta_backend_not_supported",
                    "message": "semantic delta updates require external vector backend",
                    "detail": {"vector_backend": vector_backend},
                },
                "mode": mode,
            }

        manifest = _read_manifest(manifest_file)
        provider = (_auto_configure_embedding_provider_from_keys() or "gemini").strip().lower()
        model = (os.environ.get("CORE_MEMORY_EMBEDDINGS_MODEL") or _default_embedding_model(provider)).strip()

        prev_rows = _read_rows(rows_file)
        prev_by_id = {str(r.get("bead_id") or ""): dict(r) for r in prev_rows if str(r.get("bead_id") or "")}

        visible_rows = _rows_from_corpus(build_visible_corpus(root))
        next_by_id = {str(r.get("bead_id") or ""): dict(r) for r in visible_rows if str(r.get("bead_id") or "")}

        prev_ids = set(prev_by_id.keys())
        next_ids = set(next_by_id.keys())
        delete_ids = sorted(prev_ids - next_ids)

        embed_ids: list[str] = []
        metadata_only_ids: list[str] = []
        for bid in sorted(next_ids):
            nxt = next_by_id[bid]
            prev = prev_by_id.get(bid)
            if prev is None:
                embed_ids.append(bid)
                continue
            if str(prev.get("semantic_text_hash") or "") != str(nxt.get("semantic_text_hash") or ""):
                embed_ids.append(bid)
                continue
            if _row_metadata_changed(prev, nxt):
                metadata_only_ids.append(bid)

        # ensure no overlap after classification
        embed_id_set = set(embed_ids)
        metadata_only_ids = [bid for bid in metadata_only_ids if bid not in embed_id_set]

        dim_hint = int(manifest.get("dimension") or 256)
        dim = max(1, dim_hint)
        vb = None
        embedded = 0
        metadata_updates = 0
        deleted = 0
        embed_error: str = ""

        if embed_ids:
            texts = [str((next_by_id.get(bid) or {}).get("semantic_text") or "") for bid in embed_ids]
            vecs = _embed_vectors(texts=texts, provider=provider, model=model, hash_dim=dim)
            dim = max(1, _vector_dim(vecs, fallback=dim))
            vb = _create_external_backend(root=root, backend=vector_backend, dimension=dim)
            vec_rows = _vector_rows(vecs)
            if len(vec_rows) != len(embed_ids):
                raise RuntimeError("semantic_delta_embedding_count_mismatch")
            for bid, emb in zip(embed_ids, vec_rows):
                row = next_by_id.get(bid) or {}
                vb.upsert(bead_id=bid, embedding=emb, metadata=_row_metadata(row))
                embedded += 1

        if vb is None:
            vb = _create_external_backend(root=root, backend=vector_backend, dimension=dim)

        for bid in metadata_only_ids:
            row = next_by_id.get(bid) or {}
            try:
                if hasattr(vb, "update_metadata"):
                    vb.update_metadata(bid, _row_metadata(row))  # type: ignore[attr-defined]
                    metadata_updates += 1
                else:
                    embed_ids.append(bid)
            except Exception as exc:
                embed_error = f"metadata_update_failed:{_embedding_error_token(exc)}"
                raise RuntimeError(embed_error) from exc

        for bid in delete_ids:
            vb.delete(bid)
            deleted += 1

        # persist canonical row projection for future delta checks
        next_rows = list(next_by_id.values())
        next_rows.sort(key=lambda r: (str(r.get("created_at") or ""), str(r.get("bead_id") or "")), reverse=True)
        _write_rows(rows_file, next_rows)

        fp = _fingerprint(next_rows)
        out_manifest = {
            "provider": provider,
            "model": model,
            "dimension": int(dim),
            "backend": vector_backend,
            "vector_backend": vector_backend,
            "backend_version": "v10-delta",
            "corpus_fingerprint": fp,
            "built_at": str(manifest.get("built_at") or _now()),
            "last_delta_at": _now(),
            "row_count": len(next_rows),
            "dirty": False,
            "last_dirty_at": manifest.get("last_dirty_at"),
            "last_dirty_reason": manifest.get("last_dirty_reason"),
            "last_turn_id": manifest.get("last_turn_id"),
            "last_flush_tx_id": manifest.get("last_flush_tx_id"),
            "visible_statuses": ["open", "candidate", "promoted", "archived"],
            "last_delta_counts": {
                "embedded": int(embedded),
                "metadata_only": int(metadata_updates),
                "deleted": int(deleted),
                "total_visible": int(len(next_rows)),
            },
        }
        _write_manifest(manifest_file, out_manifest)

        q2 = _read_queue(queue_file)
        q2["queued"] = False
        q2["queued_at"] = None
        q2["mode"] = "delta"
        _write_queue(queue_file, q2)

        return {
            "ok": True,
            "ran": True,
            "mode": "delta",
            "backend": vector_backend,
            "entries": len(next_rows),
            "embedded": embedded,
            "metadata_only": metadata_updates,
            "deleted": deleted,
            "manifest": str(manifest_file),
            "rows": str(rows_file),
            "lock": {"path": str(build_lock), **dict(lock_meta or {})},
        }
    finally:
        _release_build_lock(build_lock)


def semantic_lookup(root: Path, query: str, k: int = 8, mode: str | None = None) -> dict:
    _check_semantic_mode_startup()
    manifest_file, faiss_file, rows_file, _build_lock, queue_file = _paths(root)
    warnings: list[str] = []
    mode_n = _normalize_semantic_mode(mode)
    build_on_read = _semantic_build_on_read_enabled()

    def warn_once(tag: str) -> None:
        if tag not in warnings:
            warnings.append(tag)

    if not manifest_file.exists() or not rows_file.exists():
        warn_once("semantic_index_missing_artifacts")
        enqueue_semantic_rebuild(root)
        if build_on_read:
            built = build_semantic_index(root)
            if not built.get("ok"):
                code = str(((built.get("error") or {}).get("code") or "")).strip()
                if code == "semantic_build_lock_held":
                    warn_once("semantic_build_lock_held")
                else:
                    warn_once("semantic_index_build_failed")
        else:
            warn_once("semantic_build_on_read_disabled")

    manifest = _read_manifest(manifest_file)
    rows = _read_rows(rows_file)

    # Provider/model mismatch -> hard rebuild before querying.
    req_provider = (_auto_configure_embedding_provider_from_keys() or "gemini").strip().lower()
    req_model = (os.environ.get("CORE_MEMORY_EMBEDDINGS_MODEL") or _default_embedding_model(req_provider)).strip()
    req_vector_backend = _configured_vector_backend()
    # Qdrant+FastEmbed manages its own embedding model — never rebuild due to external provider mismatch.
    _manifest_is_fastembed = str(manifest.get("provider") or "").strip().lower() == "fastembed"
    _req_is_qdrant_fastembed = req_vector_backend == VECTOR_BACKEND_QDRANT
    if not (_manifest_is_fastembed and _req_is_qdrant_fastembed) and (
        (manifest.get("provider") and (str(manifest.get("provider")) != req_provider or str(manifest.get("model")) != req_model))
        or (manifest.get("vector_backend") and _normalize_vector_backend(str(manifest.get("vector_backend"))) != req_vector_backend)
    ):
        warn_once("semantic_index_config_mismatch")
        enqueue_semantic_rebuild(root)
        if build_on_read:
            rebuilt = build_semantic_index(root)
            if rebuilt.get("ok"):
                manifest = _read_manifest(manifest_file)
                rows = _read_rows(rows_file)
                warn_once("semantic_index_rebuilt_config_mismatch")
            else:
                warn_once("semantic_index_rebuild_config_mismatch_failed")
        else:
            warn_once("semantic_build_on_read_disabled")

    # Dirty/fingerprint mismatch -> serve stale + enqueue rebuild when possible.
    current_fp = _fingerprint(_rows_from_corpus(build_visible_corpus(root)))
    dirty = bool(manifest.get("dirty")) or (str(manifest.get("corpus_fingerprint") or "") != current_fp)
    stale_age_ms: int | None = None
    if dirty:
        dt = _parse_iso(str(manifest.get("last_dirty_at") or ""))
        if dt is not None:
            stale_age_ms = int((datetime.now(timezone.utc) - dt).total_seconds() * 1000)
    if dirty:
        warn_once("semantic_index_stale")
        enqueue_semantic_rebuild(root)

        # background_stale contract: never rebuild synchronously on hot query path
        # when a usable index already exists.
        mode = str(os.getenv("CORE_MEMORY_SEMANTIC_REBUILD_MODE", "background_stale") or "background_stale").strip().lower()
        if mode in {"eager", "sync"} and build_on_read:
            q = _read_queue(queue_file)
            if bool(q.get("queued")):
                rebuilt = build_semantic_index(root)
                if rebuilt.get("ok"):
                    warn_once("semantic_index_rebuilt_sync")
                    manifest = _read_manifest(manifest_file)
                    rows = _read_rows(rows_file)
                    dirty = False
                else:
                    warn_once("semantic_rebuild_sync_failed")
        elif mode in {"eager", "sync"}:
            warn_once("semantic_build_on_read_disabled")

    backend = str(manifest.get("backend") or "lexical")
    semantic_ready = bool(manifest.get("semantic_ready", backend != "lexical"))

    def lexical_rank() -> list[dict[str, Any]]:
        q_tokens = set(tokenize(query))
        scored: list[dict[str, Any]] = []
        for r in rows:
            txt = str(r.get("semantic_text") or "")
            tks = set(tokenize(txt))
            ov = len(q_tokens.intersection(tks)) if q_tokens else 0
            scored.append(
                {
                    "bead_id": r.get("bead_id"),
                    "score": float(ov),
                    "status": r.get("status"),
                    "anchor_reason": "retrieved",
                }
            )
        scored = sorted(scored, key=lambda x: (x.get("score", 0.0), str(x.get("bead_id") or "")), reverse=True)
        return scored[: max(1, int(k))]

    # If no usable rows yet, do sync build then lexical fallback response.
    if not rows:
        enqueue_semantic_rebuild(root)
        if build_on_read:
            rebuilt = build_semantic_index(root)
            if rebuilt.get("ok"):
                manifest = _read_manifest(manifest_file)
                rows = _read_rows(rows_file)
            else:
                warn_once("semantic_index_rebuild_no_rows_failed")
        else:
            warn_once("semantic_build_on_read_disabled")
            warn_once("semantic_index_rows_missing")

    if _normalize_vector_backend(backend) in _EXTERNAL_VECTOR_BACKENDS and rows:
        try:
            backend_n = _normalize_vector_backend(backend)
            manifest_provider = str(manifest.get("provider") or req_provider)
            if backend_n == VECTOR_BACKEND_QDRANT and manifest_provider == "fastembed":
                # FastEmbed index: use Qdrant's native query_text path — no external API key required.
                # dimension=0 preserves the FastEmbed sentinel so QdrantBackend skips VectorParams
                # and does not recreate (or wipe) the existing FastEmbed collection.
                vb = _create_external_backend(root=root, backend=backend_n, dimension=0)
                raw = vb.hybrid_search(query, k=max(1, int(k)))
            else:
                qv = _embed_vectors(
                    texts=[query],
                    provider=manifest_provider,
                    model=str(manifest.get("model") or req_model),
                    hash_dim=int(manifest.get("dimension") or 256),
                )
                q_rows = _vector_rows(qv)
                if not q_rows:
                    raise RuntimeError("query_embedding_empty")
                vb = _create_external_backend(
                    root=root,
                    backend=backend_n,
                    dimension=max(1, _vector_dim(qv, fallback=int(manifest.get("dimension") or 256))),
                )
                raw = vb.search(q_rows[0], k=max(1, int(k)))

            row_by_id = {str(r.get("bead_id") or ""): r for r in rows}
            out = []
            for item in raw:
                bead_id = str((item or {}).get("bead_id") or "")
                if not bead_id:
                    continue
                row = row_by_id.get(bead_id, {})
                meta = (item or {}).get("metadata") or {}
                out.append(
                    {
                        "bead_id": bead_id,
                        "score": float((item or {}).get("score") or 0.0),
                        "status": (meta.get("status") if isinstance(meta, dict) else None) or row.get("status"),
                        "anchor_reason": "retrieved",
                    }
                )

            return {
                "ok": True,
                "degraded": False,
                "backend": backend_n,
                "provider": manifest.get("provider"),
                "query": query,
                "warnings": warnings,
                "stale_age_ms": stale_age_ms,
                "results": out,
            }
        except Exception as exc:
            logger.debug("core-memory: external semantic lookup failed, using lexical fallback: %s", exc)
            warn_once("semantic_backend_query_failed_lexical_fallback")
            warn_once(f"semantic_backend_query_error:{_embedding_error_token(exc)}")

    if backend.startswith("faiss") and faiss_file.exists() and rows:
        try:
            import faiss  # type: ignore

            idx = faiss.read_index(str(faiss_file))
            dim = idx.d

            if backend == "faiss-hash":
                qv = _embed_vectors(texts=[query], provider="hash", model=str(manifest.get("model") or req_model), hash_dim=dim)
            elif backend == "faiss-openai":
                qv = _embed_vectors(texts=[query], provider="openai", model=str(manifest.get("model") or req_model), hash_dim=dim)
            elif backend == "faiss-gemini":
                qv = _embed_vectors(texts=[query], provider="gemini", model=str(manifest.get("model") or req_model), hash_dim=dim)
            else:
                raise RuntimeError(f"unsupported_backend:{backend}")

            D, I = idx.search(qv, max(1, int(k)))
            out = []
            for dist, row_idx in zip(D[0].tolist(), I[0].tolist()):
                if row_idx < 0 or row_idx >= len(rows):
                    continue
                r = rows[row_idx]
                out.append(
                    {
                        "bead_id": r.get("bead_id"),
                        "score": float(dist),
                        "status": r.get("status"),
                        "anchor_reason": "retrieved",
                    }
                )
            return {
                "ok": True,
                "degraded": False,
                "backend": backend,
                "provider": manifest.get("provider"),
                "query": query,
                "warnings": warnings,
                "stale_age_ms": stale_age_ms,
                "results": out,
            }
        except ImportError:
            global _faiss_warning_emitted
            if not _faiss_warning_emitted:
                logger.warning(
                    "core-memory: faiss-cpu/numpy not available for semantic lookup. "
                    "Install with: pip install core-memory[semantic]"
                )
                _faiss_warning_emitted = True
            warn_once("semantic_backend_query_failed_lexical_fallback")
            warn_once("semantic_backend_query_error:faiss_import_error")
        except Exception as exc:
            logger.debug("core-memory: semantic lookup failed, using lexical fallback: %s", exc)
            warn_once("semantic_backend_query_failed_lexical_fallback")
            warn_once(f"semantic_backend_query_error:{_embedding_error_token(exc)}")

    if mode_n == SEMANTIC_MODE_REQUIRED:
        detail = None
        if not semantic_ready or (str(manifest.get("provider") or req_provider) != "hash" and backend == "lexical"):
            detail = {
                "reason": "build_failed_or_invalid",
                "provider": str(manifest.get("provider") or req_provider),
                "model": str(manifest.get("model") or req_model),
                "backend": backend,
                "vector_backend": str(manifest.get("vector_backend") or req_vector_backend),
                "dimension": int(manifest.get("dimension") or 0),
                "last_build_error_code": str(manifest.get("last_build_error_code") or ""),
            }
        return semantic_unavailable_payload(
            query=query,
            warnings=warnings,
            provider=str(manifest.get("provider") or req_provider),
            detail=detail,
        )

    return {
        "ok": True,
        "degraded": True,
        "backend": "lexical",
        "provider": manifest.get("provider") or req_provider,
        "query": query,
        "warnings": list(warnings) + ["semantic_backend_unavailable_degraded"],
        "stale_age_ms": stale_age_ms,
        "results": lexical_rank(),
    }
