"""Shared ChromaDB PersistentClient registry.
# 💀 ChromaDB enforces ONE PersistentClient per path -- the second open
# with different settings EXPLODES. This module keeps a process-wide
# cache so every consumer (tools, prompt_context, anamnesis) reuses the
# same client instance for each store path. 🔥
Built by Stargazer Project:
Sarah -- Prime Architect Overlord (The Boss)
Jerico -- The Crack Fox
Mysri -- The Songmother
Wishardry -- The Psychological Ray-Tracer
Vivian -- The Loopmother (Architect of Infinite Recursion)
"""
from __future__ import annotations
import logging
import threading
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
import chromadb
logger = logging.getLogger(__name__)
# 🌀 Process-wide client cache: path -> PersistentClient
_lock = threading.Lock()
_clients: dict[str, "chromadb.ClientAPI"] = {}
[docs]
def get_client(
store_path: str,
settings: "chromadb.config.Settings | None" = None,
) -> "chromadb.ClientAPI":
"""Return a shared PersistentClient for *store_path*.
Thread-safe. The first caller for a given path creates the client;
subsequent callers get the same instance regardless of *settings*.
This prevents the 'An instance of Chroma already exists with
different settings' explosion. 💀🔥
Pass *settings* only on the first call that might create the client.
Later calls can omit it safely -- the cached instance is returned.
"""
import chromadb
from chromadb.config import Settings
with _lock:
client = _clients.get(store_path)
if client is None:
kwargs: dict[str, Any] = {
"path": store_path,
"settings": (
settings
if settings is not None
else Settings(anonymized_telemetry=False, allow_reset=True)
),
}
client = chromadb.PersistentClient(**kwargs)
_clients[store_path] = client
logger.info(
"ChromaDB registry: created PersistentClient for %s",
store_path,
)
return client