feat: initial commit — xiaoheideplog MCP (Milvus + Azure OpenAI, 4-tool RAG API)

Made-with: Cursor

.dockerignore

@@ -0,0 +1,7 @@
+.venv
+__pycache__
+*.egg-info
+.git
+.env
+*.md
+!README.md

.env.example

@@ -0,0 +1,18 @@
+# Copy to .env next to docker-compose.yml (do not commit .env).
+
+# Disk-backed data root for etcd / minio / milvus (host paths)
+DOCKER_VOLUME_DIRECTORY=/mnt/redis-disk/xiaoheideplog/volumes
+
+# Optional: publish Milvus gRPC on host (migration / debugging)
+MILVUS_PUBLISH_PORT=19530
+
+# MCP HTTP port on host
+MCP_PUBLISH_PORT=3101
+
+DEFAULT_PROJECT_PATH=/workspace
+
+# Azure OpenAI (embeddings)
+AZURE_OPENAI_ENDPOINT=https://YOUR_RESOURCE.openai.azure.com
+AZURE_OPENAI_API_KEY=
+AZURE_OPENAI_API_VERSION=2023-05-15
+AZURE_OPENAI_EMBEDDING_DEPLOYMENT=text-embedding-3-small

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+.venv/
+*.egg-info/
+.dist-info/
+.env

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.12-slim-bookworm
+
+WORKDIR /app
+
+ENV PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    MCP_TRANSPORT=streamable-http \
+    MCP_HOST=0.0.0.0 \
+    MCP_PORT=3101 \
+    DEFAULT_PROJECT_PATH=/workspace
+
+COPY pyproject.toml README.md ./
+COPY src ./src
+
+RUN pip install --upgrade pip && pip install .
+
+EXPOSE 3101
+
+HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
+  CMD python -c "from pymilvus import utility, connections; import os; connections.connect('xiaoheideplog', host=os.environ.get('MILVUS_HOST','127.0.0.1'), port=os.environ.get('MILVUS_PORT','19530')); utility.get_server_version(using='xiaoheideplog')" || exit 1
+
+CMD ["xiaoheideplog-mcp"]

README.md

@@ -0,0 +1,53 @@
+# xiaoheideplog
+
+Milvus + Azure OpenAI (`text-embedding-3-small`) project memory MCP server for Cursor. Transport: `stdio` or `streamable-http` (default in Docker on port **3101**, path `/mcp`).
+
+## Design principles
+
+1. **Long-term memory without dumping the repo into the model** — store summaries and decisions in Milvus; retrieve via MCP tools instead of pasting large files or logs into chat (token control).
+2. **Follow Milvus-style retrieval** — new collections include a **sparse BM25** field (via `milvus-model` MSMARCO stats) plus **dense** Azure embeddings; search uses **`hybrid_search` + `RRFRanker`** when that schema is present ([multi-vector / hybrid search](https://milvus.io/docs/v2.4.x/multi-vector-search.md), [reranking](https://milvus.io/docs/v2.4.x/reranking.md)). Older collections without `sparse_bm25` keep dense ANN only and add **client-side RRF** (dense order + lexical overlap) so accuracy does not rely on returning huge payloads.
+3. **Accuracy vs. payload size** — tool defaults keep bodies small; ranking uses hybrid/RRF rather than “more text = better”. `memory_semantic_search` responses include **`retrieval_mode`** describing which path ran.
+
+Environment: `HYBRID_PREFETCH` (default `96`) controls the hybrid candidate pool before `top_k`.
+
+## Layout
+
+- Data on disk: set `DOCKER_VOLUME_DIRECTORY` (default `/mnt/redis-disk/xiaoheideplog/volumes`) so etcd, MinIO, and Milvus use bind mounts under that path.
+- Environment: see [.env.example](.env.example).
+
+## Run with Docker Compose
+
+```bash
+mkdir -p /mnt/redis-disk/xiaoheideplog/volumes/{etcd,minio,milvus}
+cp .env.example .env
+# edit .env — Azure endpoint and key
+docker compose up -d --build
+```
+
+Point Cursor MCP at `http://<host>:3101/mcp` (HTTP transport).
+
+## Redis migration
+
+Install with the migrate extra, with Milvus reachable and Azure env vars set (same as server):
+
+```bash
+pip install -e ".[migrate]"
+export REDIS_URL=redis://127.0.0.1:6379/0
+export MILVUS_HOST=127.0.0.1
+export MILVUS_PORT=19530
+# AZURE_OPENAI_* ...
+xiaoheideplog-migrate-redis --dry-run
+xiaoheideplog-migrate-redis
+```
+
+## Cutover
+
+After migration checks, stop the old `cursor-project-memory` container and keep using port **3101** for the new stack.
+
+## Post-migration cleanup (optional)
+
+When Milvus is verified and the old Redis-backed MCP is no longer needed:
+
+- Stop and remove old containers (`cursor-memory-redis`, `cursor-project-memory`, etc.).
+- Remove the old Redis data directory on the host if it was only used for this feature.
+- `docker rmi` unused images (for example `cursor-project-memory:local`, dedicated `redis` images). Use `docker image prune` carefully so unrelated images are not removed.

docker-compose.yml

@@ -0,0 +1,88 @@
+# Milvus (etcd + minio + milvus) + xiaoheideplog MCP.
+# Data: bind to disk (set DOCKER_VOLUME_DIRECTORY e.g. /mnt/redis-disk/xiaoheideplog/volumes).
+# MCP: host port 3101 -> streamable-http /mcp
+
+services:
+  etcd:
+    container_name: xiaoheideplog-etcd
+    image: quay.io/coreos/etcd:v3.5.14
+    environment:
+      - ETCD_AUTO_COMPACTION_MODE=revision
+      - ETCD_AUTO_COMPACTION_RETENTION=1000
+      - ETCD_QUOTA_BACKEND_BYTES=4294967296
+      - ETCD_SNAPSHOT_COUNT=50000
+    volumes:
+      - ${DOCKER_VOLUME_DIRECTORY:-/mnt/redis-disk/xiaoheideplog/volumes}/etcd:/etcd
+    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
+    healthcheck:
+      test: ["CMD", "etcdctl", "endpoint", "health"]
+      interval: 30s
+      timeout: 20s
+      retries: 3
+
+  minio:
+    container_name: xiaoheideplog-minio
+    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
+    environment:
+      MINIO_ACCESS_KEY: minioadmin
+      MINIO_SECRET_KEY: minioadmin
+    volumes:
+      - ${DOCKER_VOLUME_DIRECTORY:-/mnt/redis-disk/xiaoheideplog/volumes}/minio:/minio_data
+    command: minio server /minio_data --console-address ":9001"
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3
+
+  milvus:
+    container_name: xiaoheideplog-milvus
+    image: milvusdb/milvus:v2.4.14
+    command: ["milvus", "run", "standalone"]
+    security_opt:
+      - seccomp:unconfined
+    environment:
+      MINIO_REGION: us-east-1
+      ETCD_ENDPOINTS: etcd:2379
+      MINIO_ADDRESS: minio:9000
+    volumes:
+      - ${DOCKER_VOLUME_DIRECTORY:-/mnt/redis-disk/xiaoheideplog/volumes}/milvus:/var/lib/milvus
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
+      interval: 30s
+      start_period: 90s
+      timeout: 20s
+      retries: 3
+    depends_on:
+      etcd:
+        condition: service_healthy
+      minio:
+        condition: service_healthy
+    ports:
+      - "${MILVUS_PUBLISH_PORT:-19530}:19530"
+
+  xiaoheideplog-mcp:
+    build: .
+    image: xiaoheideplog-mcp:local
+    container_name: xiaoheideplog-mcp
+    restart: unless-stopped
+    ports:
+      - "${MCP_PUBLISH_PORT:-3101}:3101"
+    environment:
+      MILVUS_HOST: milvus
+      MILVUS_PORT: "19530"
+      MCP_TRANSPORT: streamable-http
+      MCP_HOST: "0.0.0.0"
+      MCP_PORT: "3101"
+      DEFAULT_PROJECT_PATH: ${DEFAULT_PROJECT_PATH:-/workspace}
+      AZURE_OPENAI_ENDPOINT: ${AZURE_OPENAI_ENDPOINT}
+      AZURE_OPENAI_API_KEY: ${AZURE_OPENAI_API_KEY}
+      AZURE_OPENAI_API_VERSION: ${AZURE_OPENAI_API_VERSION:-2023-05-15}
+      AZURE_OPENAI_EMBEDDING_DEPLOYMENT: ${AZURE_OPENAI_EMBEDDING_DEPLOYMENT:-text-embedding-3-small}
+    depends_on:
+      milvus:
+        condition: service_healthy
+
+networks:
+  default:
+    name: xiaoheideplog-net

mcp.json.example

@@ -0,0 +1,8 @@
+{
+  "mcpServers": {
+    "xiaoheideplog": {
+      "url": "http://127.0.0.1:3101/mcp",
+      "transport": "http"
+    }
+  }
+}

pyproject.toml

@@ -0,0 +1,29 @@
+[build-system]
+requires = ["setuptools>=61.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "xiaoheideplog"
+version = "0.1.0"
+description = "MCP server: Milvus + Azure OpenAI embeddings for project memory"
+readme = "README.md"
+requires-python = ">=3.11"
+license = { text = "MIT" }
+dependencies = [
+  "mcp>=1.26.0",
+  "pymilvus>=2.4.0,<2.5",
+  "openai>=1.40.0",
+  "setuptools>=61,<70",
+  "milvus-model>=0.2.12",
+]
+
+[project.optional-dependencies]
+migrate = ["redis>=5.0.0"]
+dev = ["pytest>=8.0"]
+
+[project.scripts]
+xiaoheideplog-mcp = "xiaoheideplog.server:run"
+xiaoheideplog-migrate-redis = "xiaoheideplog.migrate_redis:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/xiaoheideplog/__init__.py

@@ -0,0 +1,3 @@
+"""xiaoheideplog: Milvus-backed project memory MCP."""
+
+__version__ = "0.1.0"

src/xiaoheideplog/__main__.py

@@ -0,0 +1,4 @@
+from xiaoheideplog.server import run
+
+if __name__ == "__main__":
+    run()

src/xiaoheideplog/bm25_sparse.py

@@ -0,0 +1,44 @@
+"""Milvus-style sparse BM25 using milvus-model (MSMARCO vocab via load()).
+
+See: Milvus hybrid search + RRFRanker; BM25 sparse vectors:
+https://milvus.io/docs/multi-vector-search.md
+"""
+
+from __future__ import annotations
+
+import logging
+from functools import lru_cache
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache(maxsize=1)
+def _bm25_ef() -> Any:
+    """Pretrained MSMARCO BM25 statistics — fixed vocab, safe for incremental inserts."""
+    try:
+        from milvus_model.sparse import BM25EmbeddingFunction
+    except ImportError as e:
+        raise ImportError(
+            "Install milvus-model: pip install milvus-model"
+        ) from e
+
+    ef = BM25EmbeddingFunction()
+    try:
+        ef.load()
+    except Exception as exc:
+        logger.warning("BM25 MSMARCO load failed (offline?): %s", exc)
+        raise
+    return ef
+
+
+def bm25_encode_document(text: str) -> Any:
+    """Single document row -> scipy csr (1 row) for Milvus SPARSE_FLOAT_VECTOR."""
+    ef = _bm25_ef()
+    return ef.encode_documents([text or " "])[0]
+
+
+def bm25_encode_query(text: str) -> Any:
+    """Query -> scipy csr for sparse AnnSearchRequest."""
+    ef = _bm25_ef()
+    return ef.encode_queries([text or " "])[0]

src/xiaoheideplog/embeddings.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+
+from openai import AzureOpenAI
+
+
+EMBED_DIM = 1536
+
+
+@lru_cache(maxsize=1)
+def _client() -> AzureOpenAI:
+    endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT", "").strip().rstrip("/")
+    key = os.environ.get("AZURE_OPENAI_API_KEY", "").strip()
+    api_version = os.environ.get("AZURE_OPENAI_API_VERSION", "2023-05-15").strip()
+    if not endpoint or not key:
+        raise ValueError("AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY are required")
+    return AzureOpenAI(
+        azure_endpoint=endpoint,
+        api_key=key,
+        api_version=api_version,
+    )
+
+
+def embedding_deployment() -> str:
+    return os.environ.get("AZURE_OPENAI_EMBEDDING_DEPLOYMENT", "text-embedding-3-small").strip()
+
+
+def embed_text(text: str) -> list[float]:
+    """Single text -> embedding vector (dim 1536 for text-embedding-3-small)."""
+    t = (text or "").strip()
+    if not t:
+        t = " "
+    resp = _client().embeddings.create(
+        model=embedding_deployment(),
+        input=t[:32000],
+    )
+    vec = resp.data[0].embedding
+    if len(vec) != EMBED_DIM:
+        raise ValueError(f"unexpected embedding dim {len(vec)}, expected {EMBED_DIM}")
+    return vec
+
+
+def embedding_health() -> bool:
+    try:
+        embed_text("ping")
+        return True
+    except Exception:
+        return False

src/xiaoheideplog/migrate_redis.py

@@ -0,0 +1,99 @@
+"""One-shot migration: Redis pm:* keys -> Milvus (requires [migrate] extra: redis)."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+
+import redis
+
+from xiaoheideplog.storage import (
+    CATEGORIES,
+    MilvusMemoryStore,
+    parse_pm_mem_key,
+)
+
+
+def _hash_to_path(r: redis.Redis) -> dict[str, str]:
+    rows = r.hgetall("pm:projects")
+    out: dict[str, str] = {}
+    for path, h in rows.items():
+        out[h] = path
+    return out
+
+
+def migrate(redis_url: str, dry_run: bool) -> int:
+    r = redis.Redis.from_url(redis_url, decode_responses=True)
+    if not r.ping():
+        print("redis ping failed", file=sys.stderr)
+        return 1
+
+    default_path = os.environ.get("DEFAULT_PROJECT_PATH", "/workspace").strip()
+    os.environ.setdefault("DEFAULT_PROJECT_PATH", default_path)
+
+    h2p = _hash_to_path(r)
+    keys = list(r.scan_iter(match="pm:*:mem:*", count=512))
+    print(f"found {len(keys)} memory keys", file=sys.stderr)
+
+    if dry_run:
+        for k in keys[:20]:
+            print("dry-run:", k)
+        print("dry-run complete; omit --dry-run to write Milvus")
+        return 0
+
+    store = MilvusMemoryStore(default_project_path=default_path)
+    n = 0
+    for redis_key in keys:
+        parsed = parse_pm_mem_key(redis_key)
+        if not parsed:
+            print("skip unparsable key:", redis_key, file=sys.stderr)
+            continue
+        proj_hash, mem_key = parsed
+        data = r.hgetall(redis_key)
+        if not data:
+            continue
+        cat = data.get("category", "")
+        if cat not in CATEGORIES:
+            print("skip bad category:", redis_key, cat, file=sys.stderr)
+            continue
+        project_path = h2p.get(proj_hash) or ""
+        tags_raw = data.get("tags", "")
+        tags = [t.strip() for t in tags_raw.split(",") if t.strip()]
+        store.upsert_from_migration(
+            project_hash=proj_hash,
+            project_path=project_path or "__orphan__",
+            memory_key=mem_key,
+            content=data.get("content", ""),
+            category=cat,
+            tags=tags,
+            created_at=data.get("created_at", ""),
+            updated_at=data.get("updated_at", ""),
+        )
+        n += 1
+        if n % 50 == 0:
+            print(f"migrated {n} ...", file=sys.stderr)
+
+    store.flush()
+    print(f"done: {n} rows upserted", file=sys.stderr)
+    return 0
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="Migrate Redis project memory to Milvus")
+    p.add_argument(
+        "--redis-url",
+        default=os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0"),
+        help="Redis URL (source)",
+    )
+    p.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Only list keys, do not write Milvus",
+    )
+    args = p.parse_args()
+    raise SystemExit(migrate(args.redis_url, args.dry_run))
+
+
+if __name__ == "__main__":
+    main()

src/xiaoheideplog/rerank.py

@@ -0,0 +1,43 @@
+"""Client-side RRF fusion when Milvus hybrid_search is unavailable (legacy collections)."""
+
+from __future__ import annotations
+
+
+def _rrf_score(rank: int, k: int) -> float:
+    return 1.0 / float(k + rank + 1)
+
+
+def rrf_merge_two_rankings(
+    dense_order: list[str],
+    secondary_order: list[str],
+    *,
+    k: int = 60,
+) -> list[str]:
+    """Reciprocal Rank Fusion (same family as Milvus RRFRanker)."""
+    scores: dict[str, float] = {}
+    seen: set[str] = set()
+    for i, rid in enumerate(dense_order):
+        scores[rid] = scores.get(rid, 0.0) + _rrf_score(i, k)
+        seen.add(rid)
+    for i, rid in enumerate(secondary_order):
+        scores[rid] = scores.get(rid, 0.0) + _rrf_score(i, k)
+        seen.add(rid)
+    ordered = sorted(seen, key=lambda r: scores.get(r, 0.0), reverse=True)
+    return ordered
+
+
+def lexical_rank_by_token_overlap(
+    query: str,
+    items: list[tuple[str, str]],
+) -> list[str]:
+    """items: (id, body_text). Higher overlap rank first."""
+    qt = set((query or "").lower().split())
+    if not qt:
+        return [i[0] for i in items]
+    scored: list[tuple[float, str]] = []
+    for rid, body in items:
+        dt = set((body or "").lower().split())
+        overlap = len(qt & dt) / float(len(qt))
+        scored.append((overlap, rid))
+    scored.sort(key=lambda x: (-x[0], x[1]))
+    return [rid for _, rid in scored]

src/xiaoheideplog/server.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import os
+
+from mcp.server.fastmcp import FastMCP
+
+from xiaoheideplog.embeddings import embedding_health
+from xiaoheideplog.storage import CATEGORIES, MilvusMemoryStore
+
+SERVER_NAME = "xiaoheideplog"
+
+store = MilvusMemoryStore()
+
+_host = os.getenv("MCP_HOST", "0.0.0.0")
+_port = int(os.getenv("MCP_PORT", "3101"))
+mcp = FastMCP(SERVER_NAME, host=_host, port=_port)
+
+
+def _parse_tags(tags: str | None) -> list[str]:
+    if not tags:
+        return []
+    return [t.strip() for t in tags.split(",") if t.strip()]
+
+
+def _cap_body(body: str, max_chars: int) -> tuple[str, bool]:
+    if max_chars > 0 and len(body) > max_chars:
+        return body[:max_chars] + "\n…[truncated]", True
+    return body, False
+
+
+@mcp.tool()
+def memory_search(
+    query: str,
+    top_k: int = 8,
+    category: str = "",
+    tags: str = "",
+    body_max_chars: int = 2000,
+) -> dict:
+    """Hybrid semantic+lexical search across ALL memories (global scope, no project filter).
+    Use this as the primary way to retrieve relevant context before answering.
+    Returns top_k most relevant results ranked by hybrid RRF score.
+    category filter: file_summary | decision | architecture | note | progress (optional).
+    body_max_chars=0 means no truncation."""
+    try:
+        raw = store.search(
+            query=query.strip(),
+            top_k=max(1, min(int(top_k), 128)),
+            category=category.strip() or None,
+            tags=_parse_tags(tags),
+        )
+        cap = int(body_max_chars)
+        results = []
+        for m in raw["results"]:
+            body = m.get("content") or ""
+            body, truncated = _cap_body(body, cap)
+            row = {
+                "key": m.get("key", ""),
+                "category": m.get("category", ""),
+                "tags": m.get("tags", []),
+                "body": body,
+                "score": m.get("score"),
+                "updated_at": m.get("updated_at", ""),
+            }
+            if truncated:
+                row["body_truncated"] = True
+            results.append(row)
+        return {
+            "ok": True,
+            "count": len(results),
+            "retrieval_mode": raw.get("retrieval_mode", "unknown"),
+            "results": results,
+        }
+    except Exception as exc:
+        return {"ok": False, "error": str(exc)}
+
+
+@mcp.tool()
+def memory_write(
+    key: str,
+    body: str,
+    category: str,
+    tags: str = "",
+) -> dict:
+    """Write or update one memory (embedded and stored in Milvus).
+    key: unique identifier (e.g. 'src/api/auth.py' or 'decision/db-choice').
+    category: file_summary | decision | architecture | note | progress.
+    tags: comma-separated keywords for filtering."""
+    try:
+        saved = store.save(
+            key=key.strip(),
+            content=body,
+            category=category.strip(),
+            tags=_parse_tags(tags),
+        )
+        return {
+            "ok": True,
+            "key": saved.get("key"),
+            "category": saved.get("category"),
+            "tags": saved.get("tags"),
+            "updated_at": saved.get("updated_at"),
+        }
+    except ValueError as exc:
+        return {"ok": False, "error": str(exc), "valid_categories": sorted(CATEGORIES)}
+
+
+@mcp.tool()
+def memory_delete(key: str) -> dict:
+    """Delete a memory by key (searches globally across all records)."""
+    try:
+        deleted_count = store.delete(key.strip())
+        if not deleted_count:
+            return {"ok": False, "error": f"not found: {key}"}
+        return {"ok": True, "deleted_key": key, "deleted_count": deleted_count}
+    except Exception as exc:
+        return {"ok": False, "error": str(exc)}
+
+
+@mcp.tool()
+def memory_service_status() -> dict:
+    """Check Milvus + Azure embedding connectivity and list valid categories."""
+    return {
+        "ok": True,
+        "server": SERVER_NAME,
+        "milvus_connected": bool(store.health_check()),
+        "azure_embedding_ok": bool(embedding_health()),
+        "total_records": store.count_all(),
+        "categories": sorted(CATEGORIES),
+    }
+
+
+def run() -> None:
+    transport = os.getenv("MCP_TRANSPORT", "stdio")
+    mcp.run(transport=transport)  # type: ignore[arg-type]
+
+
+if __name__ == "__main__":
+    run()

src/xiaoheideplog/storage.py

@@ -0,0 +1,683 @@
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+import re
+from datetime import datetime, timezone
+from typing import Any
+
+from pymilvus import (
+    AnnSearchRequest,
+    Collection,
+    CollectionSchema,
+    DataType,
+    FieldSchema,
+    RRFRanker,
+    connections,
+    utility,
+)
+
+from xiaoheideplog.embeddings import EMBED_DIM, embed_text
+from xiaoheideplog.rerank import lexical_rank_by_token_overlap, rrf_merge_two_rankings
+
+
+logger = logging.getLogger(__name__)
+
+CATEGORIES = {"file_summary", "decision", "architecture", "note", "progress"}
+
+CONN_ALIAS = "xiaoheideplog"
+COLLECTION_ENV = "MILVUS_COLLECTION"
+DEFAULT_COLLECTION = "xiaoheideplog_memories"
+SPARSE_FIELD = "sparse_bm25"
+
+
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
+
+
+def stable_row_id(project_hash: str, memory_key: str) -> str:
+    return hashlib.sha256(f"{project_hash}\n{memory_key}".encode("utf-8")).hexdigest()
+
+
+def _esc_expr_str(s: str) -> str:
+    return "'" + s.replace("\\", "\\\\").replace("'", "''") + "'"
+
+
+def _parse_tags_csv(raw: str) -> list[str]:
+    if not raw:
+        return []
+    return [t.strip() for t in raw.split(",") if t.strip()]
+
+
+def _embedding_input_text(
+    memory_key: str, content: str, tags: list[str]
+) -> str:
+    parts = [memory_key, content, " ".join(tags)]
+    return "\n".join(p for p in parts if p).strip() or " "
+
+
+def _bm25_runtime_available() -> bool:
+    try:
+        from xiaoheideplog.bm25_sparse import bm25_encode_document
+
+        bm25_encode_document("healthcheck")
+        return True
+    except Exception as e:
+        logger.warning("BM25 / milvus-model not available: %s", e)
+        return False
+
+
+# Fixed global namespace used by the simplified API (no project_path isolation).
+GLOBAL_HASH = hashlib.sha256(b"global").hexdigest()[:16]
+GLOBAL_PATH = "__global__"
+
+
+class MilvusMemoryStore:
+    def __init__(self, default_project_path: str = GLOBAL_PATH) -> None:
+        self.default_project_path = default_project_path
+        host = os.environ.get("MILVUS_HOST", "127.0.0.1").strip()
+        port = os.environ.get("MILVUS_PORT", "19530").strip()
+        self.collection_name = os.environ.get(COLLECTION_ENV, DEFAULT_COLLECTION).strip()
+        self.hybrid_prefetch = max(
+            32, min(int(os.environ.get("HYBRID_PREFETCH", "96")), 512)
+        )
+        connections.connect(CONN_ALIAS, host=host, port=port)
+        self.milvus_hybrid: bool = False
+        self._ensure_collection()
+        self.collection.load()
+        self.milvus_hybrid = self._collection_has_sparse_field()
+
+    def _collection_has_sparse_field(self) -> bool:
+        for f in self.collection.schema.fields:
+            if f.name == SPARSE_FIELD:
+                return True
+        return False
+
+    def _ensure_collection(self) -> None:
+        if utility.has_collection(self.collection_name, using=CONN_ALIAS):
+            self.collection = Collection(self.collection_name, using=CONN_ALIAS)
+            return
+
+        want_sparse = _bm25_runtime_available()
+        fields: list[FieldSchema] = [
+            FieldSchema(
+                name="row_id",
+                dtype=DataType.VARCHAR,
+                is_primary=True,
+                max_length=64,
+            ),
+            FieldSchema(name="project_hash", dtype=DataType.VARCHAR, max_length=32),
+            FieldSchema(name="project_path", dtype=DataType.VARCHAR, max_length=4096),
+            FieldSchema(name="memory_key", dtype=DataType.VARCHAR, max_length=2048),
+            FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64),
+            FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
+            FieldSchema(name="tags_csv", dtype=DataType.VARCHAR, max_length=8192),
+            FieldSchema(name="created_at", dtype=DataType.VARCHAR, max_length=64),
+            FieldSchema(name="updated_at", dtype=DataType.VARCHAR, max_length=64),
+            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=EMBED_DIM),
+        ]
+        if want_sparse:
+            fields.insert(
+                -1,
+                FieldSchema(name=SPARSE_FIELD, dtype=DataType.SPARSE_FLOAT_VECTOR),
+            )
+
+        schema = CollectionSchema(
+            fields,
+            description="project memory: dense (Azure) + optional sparse BM25 (Milvus hybrid)",
+        )
+        self.collection = Collection(self.collection_name, schema, using=CONN_ALIAS)
+        self.collection.create_index(
+            field_name="embedding",
+            index_params={
+                "metric_type": "COSINE",
+                "index_type": "IVF_FLAT",
+                "params": {"nlist": 1024},
+            },
+        )
+        if want_sparse:
+            self.collection.create_index(
+                field_name=SPARSE_FIELD,
+                index_params={
+                    "metric_type": "IP",
+                    "index_type": "SPARSE_INVERTED_INDEX",
+                    "params": {"drop_ratio_build": 0.2},
+                },
+            )
+
+    def health_check(self) -> bool:
+        try:
+            return bool(utility.get_server_version(using=CONN_ALIAS))
+        except Exception:
+            return False
+
+    def _project_path(self, project_path: str | None) -> str:
+        path = (project_path or self.default_project_path or "").strip()
+        if not path:
+            raise ValueError("project_path is required")
+        return path
+
+    def _project_hash(self, project_path: str | None) -> str:
+        path = self._project_path(project_path)
+        return hashlib.sha256(path.encode("utf-8")).hexdigest()[:16]
+
+    def _row_to_memory(
+        self,
+        row: dict[str, Any],
+        *,
+        project_path_override: str | None = None,
+        score: float | None = None,
+    ) -> dict[str, Any]:
+        tags = _parse_tags_csv(row.get("tags_csv") or "")
+        ph = row.get("project_hash") or ""
+        out: dict[str, Any] = {
+            "key": row.get("memory_key") or "",
+            "content": row.get("content") or "",
+            "category": row.get("category") or "",
+            "tags": tags,
+            "created_at": row.get("created_at") or "",
+            "updated_at": row.get("updated_at") or "",
+        }
+        if project_path_override is not None:
+            out["project_path"] = project_path_override
+        else:
+            out["project_path"] = row.get("project_path") or ""
+        if ph:
+            out["project_hash"] = ph
+        if score is not None:
+            out["score"] = round(score, 6)
+        return out
+
+    def _delete_by_row_id(self, row_id: str) -> None:
+        expr = f"row_id == {_esc_expr_str(row_id)}"
+        self.collection.delete(expr)
+
+    def _insert_row_dict(self, row: dict[str, Any]) -> None:
+        if self.milvus_hybrid:
+            from xiaoheideplog.bm25_sparse import bm25_encode_document
+
+            emb_in = _embedding_input_text(
+                row["memory_key"],
+                row["content"],
+                _parse_tags_csv(row.get("tags_csv") or ""),
+            )
+            row = dict(row)
+            row[SPARSE_FIELD] = bm25_encode_document(emb_in)
+        self.collection.insert([row])
+
+    def save_memory(
+        self,
+        key: str,
+        content: str,
+        category: str,
+        tags: list[str] | None = None,
+        project_path: str | None = None,
+    ) -> dict[str, Any]:
+        normalized_key = key.strip()
+        if not normalized_key:
+            raise ValueError("key cannot be empty")
+        if category not in CATEGORIES:
+            raise ValueError(f"category must be one of {sorted(CATEGORIES)}")
+
+        proj_hash = self._project_hash(project_path)
+        path = self._project_path(project_path)
+        row_id = stable_row_id(proj_hash, normalized_key)
+        tags_u = sorted({t.strip() for t in (tags or []) if t.strip()})
+        tags_csv = ",".join(tags_u)
+        now = utc_now_iso()
+
+        existing = self.load_row(proj_hash, normalized_key)
+        created_at = existing.get("created_at", now) if existing else now
+
+        emb_in = _embedding_input_text(normalized_key, content, tags_u)
+        vector = embed_text(emb_in)
+        c = content if len(content) <= 65000 else content[:65000]
+
+        self._delete_by_row_id(row_id)
+        row = {
+            "row_id": row_id,
+            "project_hash": proj_hash,
+            "project_path": path,
+            "memory_key": normalized_key,
+            "category": category,
+            "content": c,
+            "tags_csv": tags_csv,
+            "created_at": created_at,
+            "updated_at": now,
+            "embedding": vector,
+        }
+        self._insert_row_dict(row)
+        self.collection.flush()
+        return self.load_memory(normalized_key, project_path=project_path) or {}
+
+    def load_row(self, project_hash: str, memory_key: str) -> dict[str, Any] | None:
+        rid = stable_row_id(project_hash, memory_key)
+        rows = self.collection.query(
+            expr=f"row_id == {_esc_expr_str(rid)}",
+            output_fields=[
+                "project_hash",
+                "project_path",
+                "memory_key",
+                "category",
+                "content",
+                "tags_csv",
+                "created_at",
+                "updated_at",
+            ],
+        )
+        return rows[0] if rows else None
+
+    def load_memory(
+        self, key: str, project_path: str | None = None
+    ) -> dict[str, Any] | None:
+        proj_hash = self._project_hash(project_path)
+        row = self.load_row(proj_hash, key.strip())
+        if not row:
+            return None
+        return self._row_to_memory(row)
+
+    def delete_memory(self, key: str, project_path: str | None = None) -> bool:
+        proj_hash = self._project_hash(project_path)
+        row = self.load_row(proj_hash, key.strip())
+        if not row:
+            return False
+        rid = stable_row_id(proj_hash, key.strip())
+        self._delete_by_row_id(rid)
+        self.collection.flush()
+        return True
+
+    def _base_expr(
+        self,
+        scope: str,
+        project_path: str | None,
+        category: str | None,
+    ) -> str:
+        parts: list[str] = ["row_id != ''"]
+        if category:
+            parts.append(f"category == {_esc_expr_str(category)}")
+        if scope == "project":
+            ph = self._project_hash(project_path)
+            parts.append(f"project_hash == {_esc_expr_str(ph)}")
+        return " and ".join(parts)
+
+    def list_memories(
+        self,
+        project_path: str | None = None,
+        category: str | None = None,
+        scope: str = "all",
+        *,
+        limit: int = 200,
+        offset: int = 0,
+    ) -> list[dict[str, Any]]:
+        s = (scope or "all").strip().lower()
+        if s not in ("all", "project"):
+            raise ValueError("scope must be 'all' or 'project'")
+        expr = self._base_expr(s, project_path, category)
+        rows = self.collection.query(
+            expr=expr,
+            output_fields=[
+                "project_hash",
+                "project_path",
+                "memory_key",
+                "category",
+                "content",
+                "tags_csv",
+                "created_at",
+                "updated_at",
+            ],
+            offset=max(0, offset),
+            limit=min(max(1, limit), 16384),
+        )
+        out: list[dict[str, Any]] = []
+        for row in rows:
+            item = self._row_to_memory(row)
+            out.append(item)
+        out.sort(
+            key=lambda r: (r.get("updated_at") or "", r.get("key") or ""),
+            reverse=True,
+        )
+        return out
+
+    def search_memories(
+        self,
+        query: str,
+        project_path: str | None = None,
+        category: str | None = None,
+        tags: list[str] | None = None,
+        scope: str = "all",
+        *,
+        top_k: int = 12,
+        min_score: float = -1.0,
+    ) -> dict[str, Any]:
+        s = (scope or "all").strip().lower()
+        if s not in ("all", "project"):
+            raise ValueError("scope must be 'all' or 'project'")
+        tag_need = {t.strip() for t in (tags or []) if t.strip()}
+        q = (query or "").strip()
+        base_expr = self._base_expr(s, project_path, category)
+        prefetch = max(self.hybrid_prefetch, top_k * 4)
+
+        if not q:
+            raw = self.list_memories(
+                project_path=project_path,
+                category=category,
+                scope=s,
+                limit=min(max(top_k * 2, 8), 64),
+                offset=0,
+            )
+            filtered: list[dict[str, Any]] = []
+            for item in raw:
+                if tag_need and not tag_need <= set(item.get("tags") or []):
+                    continue
+                filtered.append(item)
+            return {
+                "results": filtered[:top_k],
+                "retrieval_mode": "metadata_filter_only",
+            }
+
+        out_fields = [
+            "project_hash",
+            "project_path",
+            "memory_key",
+            "category",
+            "content",
+            "tags_csv",
+            "created_at",
+            "updated_at",
+        ]
+
+        if self.milvus_hybrid:
+            try:
+                from xiaoheideplog.bm25_sparse import bm25_encode_query
+
+                qvec = embed_text(q)
+                sparse_mat = bm25_encode_query(q)
+                dense_req = AnnSearchRequest(
+                    [qvec],
+                    "embedding",
+                    {"metric_type": "COSINE", "params": {"nprobe": 16}},
+                    prefetch,
+                    expr=base_expr,
+                )
+                sparse_req = AnnSearchRequest(
+                    sparse_mat,
+                    SPARSE_FIELD,
+                    {"metric_type": "IP"},
+                    prefetch,
+                    expr=base_expr,
+                )
+                res = self.collection.hybrid_search(
+                    [dense_req, sparse_req],
+                    RRFRanker(k=60),
+                    limit=prefetch,
+                    output_fields=out_fields,
+                )
+                matches = self._hits_to_memories(
+                    res[0],
+                    out_fields,
+                    tag_need,
+                    top_k,
+                    min_score,
+                    apply_min_score=False,
+                    hybrid_fusion=True,
+                )
+                return {
+                    "results": matches,
+                    "retrieval_mode": "milvus_hybrid_rrf_bm25_dense",
+                }
+            except Exception as e:
+                logger.warning("hybrid_search failed, falling back: %s", e)
+
+        qvec = embed_text(q)
+        search_params = {"metric_type": "COSINE", "params": {"nprobe": 16}}
+        res = self.collection.search(
+            data=[qvec],
+            anns_field="embedding",
+            param=search_params,
+            limit=prefetch,
+            expr=base_expr,
+            output_fields=out_fields,
+        )
+
+        by_id: dict[str, dict[str, Any]] = {}
+        dense_order: list[str] = []
+        for hit in res[0]:
+            row = {f: hit.get(f) for f in out_fields}
+            rid = hit.id
+            if rid is None:
+                continue
+            dist = float(hit.distance)
+            cos_sim = 1.0 - dist
+            if min_score >= 0.0 and cos_sim < min_score:
+                continue
+            mem = self._row_to_memory(row, score=cos_sim)
+            if tag_need and not tag_need <= set(mem.get("tags") or []):
+                continue
+            by_id[str(rid)] = mem
+            dense_order.append(str(rid))
+
+        if len(dense_order) < 2:
+            out = [by_id[i] for i in dense_order][:top_k]
+            return {"results": out, "retrieval_mode": "dense_vector_only"}
+
+        pairs: list[tuple[str, str]] = [
+            (rid, by_id[rid].get("content") or "") for rid in dense_order if rid in by_id
+        ]
+        lex_order = lexical_rank_by_token_overlap(q, pairs)
+        fused = rrf_merge_two_rankings(dense_order, lex_order, k=60)
+        ordered: list[dict[str, Any]] = []
+        for rid in fused:
+            if rid in by_id:
+                ordered.append(by_id[rid])
+            if len(ordered) >= top_k:
+                break
+        return {
+            "results": ordered,
+            "retrieval_mode": "dense_vector_plus_lexical_rrf",
+        }
+
+    def _hits_to_memories(
+        self,
+        hits: Any,
+        out_fields: list[str],
+        tag_need: set[str],
+        top_k: int,
+        min_score: float,
+        *,
+        apply_min_score: bool,
+        hybrid_fusion: bool = False,
+    ) -> list[dict[str, Any]]:
+        matches: list[dict[str, Any]] = []
+        for hit in hits:
+            row = {f: hit.get(f) for f in out_fields}
+            dist = float(hit.distance)
+            if hybrid_fusion:
+                rank_score = round(1.0 / (1.0 + abs(dist)), 6)
+                mem = self._row_to_memory(row, score=rank_score)
+            else:
+                cos_sim = 1.0 - dist
+                if apply_min_score and min_score >= 0.0 and cos_sim < min_score:
+                    continue
+                mem = self._row_to_memory(row, score=cos_sim)
+            if tag_need and not tag_need <= set(mem.get("tags") or []):
+                continue
+            matches.append(mem)
+            if len(matches) >= top_k:
+                break
+        return matches
+
+    def list_written_tags(
+        self, project_path: str | None = None, scope: str = "all"
+    ) -> list[str]:
+        s = (scope or "all").strip().lower()
+        if s not in ("all", "project"):
+            raise ValueError("scope must be 'all' or 'project'")
+        expr = self._base_expr(s, project_path, None)
+        seen: set[str] = set()
+        batch = 2000
+        off = 0
+        while True:
+            rows = self.collection.query(
+                expr=expr,
+                output_fields=["tags_csv"],
+                offset=off,
+                limit=batch,
+            )
+            if not rows:
+                break
+            for row in rows:
+                for t in _parse_tags_csv(row.get("tags_csv") or ""):
+                    seen.add(t)
+            if len(rows) < batch:
+                break
+            off += batch
+        return sorted(seen)
+
+    def project_overview(
+        self,
+        project_path: str | None = None,
+        *,
+        max_keys_preview: int = 40,
+    ) -> dict[str, Any]:
+        project = self._project_path(project_path)
+        proj_hash = self._project_hash(project_path)
+        cap = max(1, min(int(max_keys_preview), 500))
+        all_rows = self.collection.query(
+            expr=f"project_hash == {_esc_expr_str(proj_hash)}",
+            output_fields=["memory_key", "category"],
+            limit=16384,
+        )
+        keys = sorted({r.get("memory_key") for r in all_rows if r.get("memory_key")})
+        by_category: dict[str, dict[str, Any]] = {}
+        for cat in sorted(CATEGORIES):
+            kcnt = sum(
+                1
+                for r in all_rows
+                if r.get("category") == cat and r.get("memory_key")
+            )
+            by_category[cat] = {"count": kcnt}
+        preview_all = keys[:cap]
+        return {
+            "project_path": project,
+            "project_hash": proj_hash,
+            "total_memories": len(keys),
+            "keys_preview": preview_all,
+            "keys_omitted_count": max(0, len(keys) - len(preview_all)),
+            "by_category": by_category,
+            "note": "Token-safe overview. Use memory_semantic_search / memory_read for content.",
+        }
+
+    def upsert_from_migration(
+        self,
+        *,
+        project_hash: str,
+        project_path: str,
+        memory_key: str,
+        content: str,
+        category: str,
+        tags: list[str],
+        created_at: str,
+        updated_at: str,
+    ) -> None:
+        if category not in CATEGORIES:
+            raise ValueError(f"invalid category {category}")
+        row_id = stable_row_id(project_hash, memory_key)
+        tags_u = sorted({t.strip() for t in tags if t.strip()})
+        tags_csv = ",".join(tags_u)
+        emb_in = _embedding_input_text(memory_key, content, tags_u)
+        vector = embed_text(emb_in)
+        c = content if len(content) <= 65000 else content[:65000]
+        self._delete_by_row_id(row_id)
+        row = {
+            "row_id": row_id,
+            "project_hash": project_hash,
+            "project_path": project_path,
+            "memory_key": memory_key,
+            "category": category,
+            "content": c,
+            "tags_csv": tags_csv,
+            "created_at": created_at or utc_now_iso(),
+            "updated_at": updated_at or utc_now_iso(),
+            "embedding": vector,
+        }
+        self._insert_row_dict(row)
+
+    def flush(self) -> None:
+        self.collection.flush()
+
+    # ------------------------------------------------------------------
+    # Simplified global-scope API (no project_path isolation).
+    # These are used by the new server.py tools.
+    # ------------------------------------------------------------------
+
+    def save(
+        self,
+        key: str,
+        content: str,
+        category: str,
+        tags: list[str] | None = None,
+    ) -> dict[str, Any]:
+        """Write/update a memory in the global namespace."""
+        return self.save_memory(
+            key=key,
+            content=content,
+            category=category,
+            tags=tags,
+            project_path=GLOBAL_PATH,
+        )
+
+    def delete(self, key: str) -> int:
+        """Delete all records matching memory_key globally. Returns count deleted."""
+        normalized = key.strip()
+        rows = self.collection.query(
+            expr=f"memory_key == {_esc_expr_str(normalized)}",
+            output_fields=["row_id"],
+            limit=256,
+        )
+        if not rows:
+            return 0
+        for row in rows:
+            self._delete_by_row_id(row["row_id"])
+        self.collection.flush()
+        return len(rows)
+
+    def search(
+        self,
+        query: str,
+        top_k: int = 8,
+        category: str | None = None,
+        tags: list[str] | None = None,
+    ) -> dict[str, Any]:
+        """Hybrid semantic+lexical search across ALL records (global scope)."""
+        return self.search_memories(
+            query=query,
+            project_path=None,
+            category=category,
+            tags=tags,
+            scope="all",
+            top_k=top_k,
+            min_score=-1.0,
+        )
+
+    def count_all(self) -> int:
+        """Return total number of records in the collection."""
+        try:
+            rows = self.collection.query(
+                expr="row_id != ''",
+                output_fields=["row_id"],
+                limit=16384,
+            )
+            return len(rows)
+        except Exception:
+            return -1
+
+
+_MEM_RE = re.compile(r"^pm:([0-9a-f]{16}):mem:(.+)$")
+
+
+def parse_pm_mem_key(redis_key: str) -> tuple[str, str] | None:
+    m = _MEM_RE.match(redis_key)
+    if not m:
+        return None
+    return m.group(1), m.group(2)