fix: resolve config duplication, dead code, and ChromaDB singleton

- cw_tools.py: import CW_MCP_SERVER_DIR and CW_CLIENT_ID from config.py
  instead of reading os.getenv() directly (was duplicating config source)
- cw_tools.py: remove unused _run() sync wrapper (dead code, never called)
- vector_store.py: singleton ChromaDB PersistentClient (was creating new
  client per query — wasteful and unnecessary)
- vector_store.py: consistent pathlib usage (was mixing os.path + glob
  with pathlib.Path from config)

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Author: jace-ryan

server/tools/cw_tools.py

@@ -30,17 +30,15 @@
 
 from config import (
     CW_BASE_URL,
+    CW_CLIENT_ID,
     CW_COMPANY_ID,
+    CW_MCP_SERVER_DIR,
     CW_PRIVATE_KEY,
     CW_PUBLIC_KEY,
 )
 
 logger = logging.getLogger(__name__)
 
-# Path to your built cw-mcp-server (must contain dist/index.js after `npm run build`)
-CW_MCP_SERVER_DIR: str = os.getenv("CW_MCP_SERVER_DIR", "")
-CW_CLIENT_ID: str = os.getenv("CW_CLIENT_ID", "")
-
 # ---------------------------------------------------------------------------
 # MCP client session (lazy singleton)
 # ---------------------------------------------------------------------------
@@ -144,20 +142,6 @@ async def _close_session() -> None:
     _session = _client_ctx = _session_ctx = None
 
 
-# ---------------------------------------------------------------------------
-# Synchronous wrappers (FastAPI endpoints are async but tool callers may not be)
-# ---------------------------------------------------------------------------
-
-def _run(coro):
-    """Run an async coroutine from sync context."""
-    try:
-        loop = asyncio.get_running_loop()
-        # Already in an async context — just return the coroutine for await
-        return coro
-    except RuntimeError:
-        return asyncio.run(coro)
-
-
 # ---------------------------------------------------------------------------
 # Public tool functions — IRP-relevant subset of the 73 tools
 # ---------------------------------------------------------------------------

server/vector_store.py

@@ -7,8 +7,8 @@
 
 from __future__ import annotations
 
-import glob
 import os
+from pathlib import Path
 from typing import Any
 
 import chromadb
@@ -20,14 +20,19 @@
 
 _embedding_fn = embedding_functions.DefaultEmbeddingFunction()
 
+_client: chromadb.PersistentClient | None = None
+
 
 def _get_client() -> chromadb.PersistentClient:
-    os.makedirs(CHROMA_DIR, exist_ok=True)
-    return chromadb.PersistentClient(path=str(CHROMA_DIR))
+    global _client
+    if _client is None:
+        os.makedirs(CHROMA_DIR, exist_ok=True)
+        _client = chromadb.PersistentClient(path=str(CHROMA_DIR))
+    return _client
 
 
-def _get_collection(client: chromadb.PersistentClient) -> Any:
-    return client.get_or_create_collection(
+def _get_collection() -> Any:
+    return _get_client().get_or_create_collection(
         name=COLLECTION_NAME,
         embedding_function=_embedding_fn,
         metadata={"hnsw:space": "cosine"},
@@ -51,9 +56,8 @@ def ingest_playbooks(playbooks_dir: str | None = None) -> dict[str, Any]:
 
     Returns a summary dict with ``files_ingested``, ``total_chunks``, and ``details``.
     """
-    src = playbooks_dir or str(PLAYBOOKS_DIR)
-    client = _get_client()
-    collection = _get_collection(client)
+    src = Path(playbooks_dir) if playbooks_dir else PLAYBOOKS_DIR
+    collection = _get_collection()
 
     # Wipe existing data for a clean re-index.
     try:
@@ -63,27 +67,26 @@ def ingest_playbooks(playbooks_dir: str | None = None) -> dict[str, Any]:
     except Exception:
         pass
 
-    md_files = sorted(glob.glob(os.path.join(src, "*.md")))
+    skip = {"README.md", "full-irp-template.md"}
+    md_files = sorted(src.glob("*.md"))
     total_chunks = 0
     details: list[dict[str, Any]] = []
 
     for filepath in md_files:
-        filename = os.path.basename(filepath)
-        if filename in ("README.md", "full-irp-template.md"):
+        if filepath.name in skip:
             continue
 
-        with open(filepath) as f:
-            content = f.read()
+        content = filepath.read_text()
         if not content.strip():
             continue
 
-        playbook_type = filename.removesuffix(".md")
+        playbook_type = filepath.stem
         chunks = chunk_document(content)
 
         ids = [f"{playbook_type}__chunk_{i}" for i in range(len(chunks))]
         metadatas = [
             {
-                "source_file": filename,
+                "source_file": filepath.name,
                 "playbook_type": playbook_type,
                 "chunk_index": i,
                 "total_chunks": len(chunks),
@@ -93,7 +96,7 @@ def ingest_playbooks(playbooks_dir: str | None = None) -> dict[str, Any]:
 
         collection.add(ids=ids, documents=chunks, metadatas=metadatas)
         total_chunks += len(chunks)
-        details.append({"file": filename, "chunks": len(chunks)})
+        details.append({"file": filepath.name, "chunks": len(chunks)})
 
     return {
         "files_ingested": len(details),
@@ -104,8 +107,7 @@ def ingest_playbooks(playbooks_dir: str | None = None) -> dict[str, Any]:
 
 def search_playbooks(query: str, n_results: int = 5) -> list[dict[str, Any]]:
     """Semantic search — returns the *n_results* closest playbook chunks."""
-    client = _get_client()
-    collection = _get_collection(client)
+    collection = _get_collection()
 
     results = collection.query(query_texts=[query], n_results=n_results)
 
@@ -124,8 +126,7 @@ def search_playbooks(query: str, n_results: int = 5) -> list[dict[str, Any]]:
 
 def list_playbooks() -> list[dict[str, Any]]:
     """Return de-duplicated metadata for every ingested playbook type."""
-    client = _get_client()
-    collection = _get_collection(client)
+    collection = _get_collection()
 
     seen: dict[str, dict[str, Any]] = {}
     for meta in collection.get()["metadatas"]: