Compare commits

...

2 Commits

Author SHA1 Message Date
Alexander Whitestone
f28843878e fix: disable chromadb telemetry in nexus clients (#1427)
Some checks failed
CI / test (pull_request) Failing after 1m7s
CI / validate (pull_request) Failing after 57s
Review Approval Gate / verify-review (pull_request) Failing after 8s
2026-04-17 01:45:37 -04:00
Alexander Whitestone
69cc254baf wip: add chroma telemetry regression tests for #1427 2026-04-17 01:36:12 -04:00
5 changed files with 86 additions and 4 deletions

View File

@@ -53,13 +53,28 @@ def _get_client(palace_path: Path):
"Run: pip install chromadb (or: pip install mempalace)"
) from exc
try:
from chromadb.config import Settings # type: ignore
except Exception: # pragma: no cover - supports MagicMock-based tests
Settings = getattr(getattr(chromadb, "config", None), "Settings", None)
if Settings is None:
Settings = getattr(chromadb, "Settings", None)
if Settings is None:
raise MemPalaceUnavailable(
"ChromaDB Settings API unavailable. "
"Upgrade chromadb or verify the install."
)
if not palace_path.exists():
raise MemPalaceUnavailable(
f"Palace directory not found: {palace_path}\n"
"Run 'mempalace mine' to initialise the palace."
)
return chromadb.PersistentClient(path=str(palace_path))
return chromadb.PersistentClient(
path=str(palace_path),
settings=Settings(anonymized_telemetry=False),
)
def search_memories(

View File

@@ -5,6 +5,7 @@ Filters and ranks content by Hermes/Timmy relevance
"""
import chromadb
from chromadb.config import Settings
from chromadb.utils import embedding_functions
from typing import List, Dict, Any
import json
@@ -26,7 +27,10 @@ HERMES_CONTEXT = [
class RelevanceEngine:
def __init__(self, collection_name: str = "deep_dive"):
self.client = chromadb.PersistentClient(path="./chroma_db")
self.client = chromadb.PersistentClient(
path="./chroma_db",
settings=Settings(anonymized_telemetry=False),
)
self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2"
)

View File

@@ -16,6 +16,7 @@ from pathlib import Path
try:
import chromadb
from chromadb.config import Settings
except ImportError:
print("ERROR: chromadb not installed")
sys.exit(1)
@@ -34,7 +35,10 @@ VIOLATION_KEYWORDS = [
def audit(palace_path: Path):
violations = []
client = chromadb.PersistentClient(path=str(palace_path))
client = chromadb.PersistentClient(
path=str(palace_path),
settings=Settings(anonymized_telemetry=False),
)
try:
col = client.get_collection("mempalace_drawers")
except Exception as e:

View File

@@ -10,6 +10,7 @@ import json
import sys
from pathlib import Path
import chromadb
from chromadb.config import Settings
PALACE_PATH = "/root/wizards/bezalel/.mempalace/palace"
FLEET_INCOMING = "/var/lib/mempalace/fleet/incoming"
@@ -18,7 +19,10 @@ DOCS_PER_ROOM = 5
def main():
client = chromadb.PersistentClient(path=PALACE_PATH)
client = chromadb.PersistentClient(
path=PALACE_PATH,
settings=Settings(anonymized_telemetry=False),
)
col = client.get_collection("mempalace_drawers")
# Discover rooms in this wing

View File

@@ -0,0 +1,55 @@
"""Regression tests for ChromaDB telemetry hardening.
Issue #1427: ChromaDB defaults to anonymous telemetry unless explicitly disabled.
These tests ensure every direct PersistentClient() call opts out.
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
from nexus.mempalace.searcher import _get_client
PROJECT_ROOT = Path(__file__).parent.parent
DIRECT_CLIENT_FILES = [
PROJECT_ROOT / "nexus/mempalace/searcher.py",
PROJECT_ROOT / "scripts/mempalace_export.py",
PROJECT_ROOT / "scripts/audit_mempalace_privacy.py",
PROJECT_ROOT / "scaffold/deep-dive/relevance/relevance_engine.py",
]
def test_get_client_disables_chroma_telemetry(tmp_path):
mock_chroma = MagicMock()
mock_settings = MagicMock(name="Settings")
mock_chroma.PersistentClient.return_value = "CLIENT"
mock_settings.return_value = "SETTINGS"
with patch.dict(
"sys.modules",
{
"chromadb": mock_chroma,
"chromadb.config": MagicMock(Settings=mock_settings),
},
):
(tmp_path / "chroma.sqlite3").touch()
client = _get_client(tmp_path)
assert client == "CLIENT"
mock_settings.assert_called_once_with(anonymized_telemetry=False)
mock_chroma.PersistentClient.assert_called_once_with(path=str(tmp_path), settings="SETTINGS")
def test_all_direct_persistent_clients_explicitly_disable_telemetry():
missing = []
needle = "anonymized_telemetry=False"
for path in DIRECT_CLIENT_FILES:
text = path.read_text(encoding="utf-8")
if needle not in text:
missing.append(path.relative_to(PROJECT_ROOT).as_posix())
assert not missing, (
"Direct Chroma PersistentClient call(s) still missing explicit telemetry opt-out:\n"
+ "\n".join(f" - {item}" for item in missing)
)