fix(test): isolate the test suite from a live B12 install

CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Fixed
 
+- Running the test suite on a machine with B12 installed no longer touches the live install. Installer tests used to rebind the live `com.b12.mcp.daemon` launchd job to a temporary plist and stop the live embed daemon, and a test session wrote a summary into the live database. A root `conftest.py` now isolates `HOME`, the daemon sockets and `launchctl`/`crontab` before any test module loads.
+- `B12_EMBED_RUNTIME_DIR` and `B12_MCP_DAEMON_SOCK` are now honored by every hook and script. About 30 call sites, plus the installer's socket wait and uninstall, still used hardcoded `/tmp/b12-*` paths, so a custom runtime dir split clients from the daemon. The installer now also writes custom values into the launchd job. Defaults are unchanged.
 - Hook watchdogs no longer hold the host's stdout pipe open after the hook exits. The orphaned watchdog `sleep` delayed EOF until the timer expired, and Claude Code then refused hook output that injected context, so proactively surfaced memories and tag corrections were dropped.
 - The MCP server and daemon no longer flood stderr with SDK validation dumps for newer-protocol client probes (`server/discover`, pre-initialization `roots/list_changed`) or with one INFO line per request. Each known probe is logged once as a single line; all other warnings are unchanged. In daemon mode this stderr is an unrotated launchd log that had grown by megabytes per day.
 - Installer now migrates launchd jobs that still run B12 Python scripts with `/usr/bin/python3` (Python 3.9 on macOS) to the B12 venv interpreter. Older consolidation jobs failed on every run because the scripts require Python 3.11+. Jobs that were unloaded or disabled stay off, and `/bin/bash` jobs are untouched.

CONTRIBUTING.md

@@ -94,6 +94,13 @@ echo "retrieval exit=$?"
 
 The CI workflow at `.github/workflows/ci.yml` runs the same checks on every PR.
 
+The pytest suite is safe to run on a machine where B12 is installed: the root
+`conftest.py` points `HOME`, `B12_EMBED_RUNTIME_DIR` and `B12_MCP_DAEMON_SOCK` at
+a private temp tree and puts no-op `launchctl` and `crontab` stubs first on
+`PATH` before any test module is imported. Keep new code on those knobs (never a
+hardcoded `/tmp/b12-*` path or a direct `launchctl` call outside `install.sh`),
+or a test run can reach the live daemons and database.
+
 ## PR conventions
 
 - **Title format**: `feat(scope): short summary` / `fix(scope): ...` / `docs(scope): ...` / `chore(scope): ...`.

conftest.py

@@ -0,0 +1,84 @@
+"""Keep the whole test suite away from a developer's live B12 install.
+
+The memory DB (~/Library/Application Support/mcp-memory), ~/.B12 and
+~/Library/LaunchAgents all derive from $HOME, while the launchd job domain,
+the daemon sockets and PID files under /tmp and the user crontab are shared by
+the whole account. Running the suite on a machine with B12 installed once
+rebound the live com.b12.mcp.daemon to a pytest temp plist, stopped the live
+embed daemon and wrote a session summary into the live database.
+
+pytest_configure runs before any test module is imported, so module-level
+paths (DB_PATH, socket paths) already see the isolated values:
+
+- HOME, the embed runtime dir and the MCP daemon socket point into a private
+  temp tree; inherited B12_* settings are dropped.
+- No-op `launchctl` and `crontab` stubs come first on PATH, so an installer run
+  can never load, unload or schedule anything for the real account.
+- The Hugging Face cache keeps its real location so model files are not
+  downloaded again, and B12_TEST_VENV_PYTHON points at the installed venv
+  interpreter for tests that only need to run it.
+"""
+from __future__ import annotations
+
+import os
+import pathlib
+import shutil
+import tempfile
+
+_LAUNCHCTL = """#!/bin/sh
+# Test isolation stub: never touch the real launchd domain.
+echo "launchctl $*" >> "{log}"
+# `launchctl list <label>` reports "not loaded" so callers never reload a job.
+if [ "$1" = list ] && [ -n "$2" ]; then exit 113; fi
+exit 0
+"""
+
+_CRONTAB = """#!/bin/sh
+# Test isolation stub: never read or replace the real user crontab.
+echo "crontab $*" >> "{log}"
+if [ "$1" = - ]; then cat >/dev/null; fi
+exit 0
+"""
+
+
+_ISOLATION_BASE: pathlib.Path | None = None
+
+
+def pytest_configure(config):
+    global _ISOLATION_BASE
+    real_home = os.path.expanduser("~")
+    # Short base under /tmp keeps AF_UNIX socket paths below the 104-byte limit.
+    base = pathlib.Path(tempfile.mkdtemp(prefix="b12t-", dir="/tmp" if os.path.isdir("/tmp") else None))
+    _ISOLATION_BASE = base
+    home, runtime, stubs = base / "home", base / "run", base / "bin"
+    for path in (home, runtime, stubs):
+        path.mkdir()
+
+    log = base / "stub-calls.log"
+    for name, body in (("launchctl", _LAUNCHCTL), ("crontab", _CRONTAB)):
+        stub = stubs / name
+        stub.write_text(body.format(log=log))
+        stub.chmod(0o755)
+
+    real_venv_python = os.path.join(real_home, ".local", "b12-venv", "bin", "python3")
+    hf_cache = os.path.join(real_home, ".cache", "huggingface")
+    if "HF_HOME" not in os.environ and os.path.isdir(hf_cache):
+        os.environ["HF_HOME"] = hf_cache
+
+    for key in [k for k in os.environ if k.startswith("B12_")]:
+        del os.environ[key]
+    os.environ["HOME"] = str(home)
+    # On native Windows expanduser() reads USERPROFILE, not HOME.
+    os.environ["USERPROFILE"] = str(home)
+    os.environ["B12_EMBED_RUNTIME_DIR"] = str(runtime)
+    os.environ["B12_MCP_DAEMON_SOCK"] = str(runtime / "b12-mcp.sock")
+    os.environ["B12_TEST_STUB_LOG"] = str(log)
+    # Tests that need the installed interpreter (not its data) can still find it.
+    if os.path.exists(real_venv_python):
+        os.environ["B12_TEST_VENV_PYTHON"] = real_venv_python
+    os.environ["PATH"] = f"{stubs}{os.pathsep}{os.environ.get('PATH', '')}"
+
+
+def pytest_unconfigure(config):
+    if _ISOLATION_BASE is not None:
+        shutil.rmtree(_ISOLATION_BASE, ignore_errors=True)

hooks/memory-checkpoint.sh

@@ -294,7 +294,7 @@ if not prefix_classified:
     # ── ML classify via daemon (LogReg head over embeddings) ─────
     import socket as _sock
     _uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-    _daemon_sock = f"/tmp/b12-embed-{_uid}.sock"
+    _daemon_sock = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.sock")
 
     def _daemon_classify(text):
         """Call daemon classify op. Returns type str or None."""

hooks/memory-precompact.sh

@@ -198,7 +198,7 @@ fi
 
 VENV_PYTHON="$HOME/.local/b12-venv/bin/python3"
 _UID=$(id -u 2>/dev/null || echo $$)
-EMBED_SOCK="/tmp/b12-embed-${_UID}.sock"
+EMBED_SOCK="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.sock"
 
 if [ -f "$STAGING_DIR/precompact-${SESSION_ID}.txt" ] && [ -x "$VENV_PYTHON" ]; then
   "$VENV_PYTHON" - "$STAGING_DIR/precompact-${SESSION_ID}.txt" "$PROJECT_NAME" "$CWD" "$EMBED_SOCK" << 'PCPYEOF'

hooks/memory-proactive-surface.sh

@@ -208,8 +208,8 @@ fi
 
 # ── Daemon helpers ──────────────────────────────────────────
 _UID=$(id -u 2>/dev/null || echo $$)
-EMBED_SOCK="/tmp/b12-embed-${_UID}.sock"
-EMBED_PID="/tmp/b12-embed-${_UID}.pid"
+EMBED_SOCK="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.sock"
+EMBED_PID="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.pid"
 
 daemon_alive() {
   [ -S "$EMBED_SOCK" ] && [ -f "$EMBED_PID" ] && \

hooks/memory-quality-audit.sh

@@ -76,8 +76,8 @@ if fix_mode:
 # ── Daemon helper for orphan embedding regeneration ──────────
 _uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
 # Hardcode /tmp/ — macOS TMPDIR varies per session
-_DAEMON_SOCK = f"/tmp/b12-embed-{_uid}.sock"
-_DAEMON_PID = f"/tmp/b12-embed-{_uid}.pid"
+_DAEMON_SOCK = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.sock")
+_DAEMON_PID = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.pid")
 
 def daemon_alive():
     if not os.path.exists(_DAEMON_SOCK) or not os.path.exists(_DAEMON_PID):

hooks/memory-retrieval.sh

@@ -159,8 +159,8 @@ FEEDBACK_DIR="$B12_BASE/memory-logs"
 # ── Embedding daemon helpers (Phase 1) ───────────────────────
 _UID=$(id -u 2>/dev/null || echo $$)
 # Hardcode /tmp/ — macOS TMPDIR varies per session, causing mismatch with daemon
-EMBED_SOCK="/tmp/b12-embed-${_UID}.sock"
-EMBED_PID="/tmp/b12-embed-${_UID}.pid"
+EMBED_SOCK="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.sock"
+EMBED_PID="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.pid"
 
 daemon_alive() {
   [ -S "$EMBED_SOCK" ] && [ -f "$EMBED_PID" ] && \

hooks/memory-session-end.sh

@@ -806,8 +806,8 @@ if not os.path.exists(DB_PATH):
 # Try embedding daemon before loading SentenceTransformer (~4.5s savings)
 _uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
 # Hardcode /tmp/ — macOS TMPDIR varies per session
-_DAEMON_SOCK = f"/tmp/b12-embed-{_uid}.sock"
-_DAEMON_PID = f"/tmp/b12-embed-{_uid}.pid"
+_DAEMON_SOCK = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.sock")
+_DAEMON_PID = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.pid")
 _USE_DAEMON = False
 
 if os.path.exists(_DAEMON_SOCK) and os.path.exists(_DAEMON_PID):

hooks/memory-session-start.sh

@@ -35,8 +35,8 @@ done
 # Singleton: daemon uses fcntl.flock() — only one instance can run.
 _UID=$(id -u 2>/dev/null || echo $$)
 # Hardcode /tmp/ — macOS TMPDIR varies per session, causing mismatch with daemon
-EMBED_SOCK="/tmp/b12-embed-${_UID}.sock"
-EMBED_LOCK="/tmp/b12-embed-${_UID}.lock"
+EMBED_SOCK="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.sock"
+EMBED_LOCK="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.lock"
 VENV_PYTHON="$HOME/.local/b12-venv/bin/python3"
 B12_SCRIPTS="${B12_HOOK_DIR:-$HOME/.B12/hooks}/scripts"
 DAEMON_SCRIPT="$B12_SCRIPTS/embed_daemon.py"

hooks/memory-subagent-start.sh

@@ -51,7 +51,7 @@ fi
 # Recall query: agent-type + first 200 chars of task (when present).
 QUERY=$(printf 'agent:%s %s' "$AGENT_TYPE" "${TASK:0:200}")
 _UID=$(id -u 2>/dev/null || echo $$)
-SOCK="/tmp/b12-embed-${_UID}.sock"
+SOCK="${B12_EMBED_RUNTIME_DIR:-/tmp}/b12-embed-${_UID}.sock"
 
 # Platform-aware DB path via shared resolver (Darwin / Linux / WSL /
 # Windows-bash). Codex review PR #44 P2 originated the inline switch

install.sh

@@ -3512,10 +3512,20 @@ install_mcp_daemon() {
     -e "s|B12_HOME_DATA_DIR|$DATA_DIR|g" \
     "$TEMPLATE" > "$PLIST_DEST"
 
+  # Carry custom runtime paths into the job, so the daemon binds the socket
+  # this function waits for below. Default installs need nothing here.
+  local _var _val
+  for _var in B12_MCP_DAEMON_SOCK B12_EMBED_RUNTIME_DIR; do
+    _val="${!_var:-}"
+    [ -n "$_val" ] || continue
+    /usr/libexec/PlistBuddy -c "Add :EnvironmentVariables:$_var string $_val" "$PLIST_DEST" 2>/dev/null \
+      || /usr/libexec/PlistBuddy -c "Set :EnvironmentVariables:$_var $_val" "$PLIST_DEST"
+  done
+
   launchctl load "$PLIST_DEST"
 
   # Brief wait for the socket to appear (cold start ~5-10s)
-  local SOCK="/tmp/b12-mcp-$(id -u).sock"
+  local SOCK="${B12_MCP_DAEMON_SOCK:-/tmp/b12-mcp-$(id -u).sock}"
   local i=0
   while [ $i -lt 20 ] && [ ! -S "$SOCK" ]; do
     sleep 0.5
@@ -3546,7 +3556,8 @@ uninstall_mcp_daemon() {
     info "MCP daemon was not installed; nothing to remove."
   fi
   # Best-effort socket cleanup (daemon should have done this on SIGTERM)
-  rm -f "/tmp/b12-mcp-$(id -u).sock" "/tmp/b12-mcp-$(id -u).pid" 2>/dev/null || true
+  local _sock="${B12_MCP_DAEMON_SOCK:-/tmp/b12-mcp-$(id -u).sock}"
+  rm -f "$_sock" "${_sock%.sock}.pid" 2>/dev/null || true
 }
 
 # Reload the long-lived launchd daemon IFF it is already running, so an upgrade

scripts/b12_health.py

@@ -57,7 +57,7 @@
 REQUIRED_TABLES = ["memories", "memory_embeddings", "memory_fts"]
 
 _UID = os.getuid() if hasattr(os, "getuid") else os.getpid()
-SOCK_PATH = f"/tmp/b12-embed-{_UID}.sock"
+SOCK_PATH = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.sock")
 
 CLAUDE_JSON_PATH = _HOME / ".claude.json"
 
@@ -294,8 +294,8 @@ def _daemon_pids() -> set[int]:
     """Discover live MCP/embed daemons from PID files plus process scan."""
     pids: set[int] = set()
     for path in (
-        Path(f"/tmp/b12-mcp-{_UID}.pid"),
-        Path(f"/tmp/b12-embed-{_UID}.pid"),
+        Path(os.path.splitext(os.environ.get("B12_MCP_DAEMON_SOCK", f"/tmp/b12-mcp-{_UID}.sock"))[0] + ".pid"),
+        Path(os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.pid")),
     ):
         try:
             pids.add(int(path.read_text(encoding="utf-8").strip()))

scripts/b12_mcp_daemon.py

@@ -62,7 +62,7 @@ def process_executable_path(pid=None):  # fail-open
 
 _UID = os.getuid() if hasattr(os, "getuid") else os.getpid()
 SOCK_PATH = os.environ.get("B12_MCP_DAEMON_SOCK", f"/tmp/b12-mcp-{_UID}.sock")
-PID_PATH = f"/tmp/b12-mcp-{_UID}.pid"
+PID_PATH = os.path.splitext(SOCK_PATH)[0] + ".pid"  # next to the socket; /tmp/b12-mcp-<UID>.pid by default
 
 _B12_DATA_DIR = os.environ.get("B12_DATA_DIR", os.path.expanduser("~/.B12"))
 LOG_DIR = os.path.join(_B12_DATA_DIR, "memory-logs")

scripts/codex_session_end.py

@@ -120,7 +120,7 @@ def _get_embedding_daemon(text):
     """Get embedding via embed daemon Unix socket."""
     import socket
     _uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-    sock_path = f"/tmp/b12-embed-{_uid}.sock"
+    sock_path = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.sock")
     if not os.path.exists(sock_path):
         return None
     try:

scripts/consolidation_engine.py

@@ -84,8 +84,8 @@ def get_db_path() -> str:
 STRENGTH_CAP = 5.0
 
 _UID = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-DAEMON_SOCK = f"/tmp/b12-embed-{_UID}.sock"
-DAEMON_PID = f"/tmp/b12-embed-{_UID}.pid"
+DAEMON_SOCK = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.sock")
+DAEMON_PID = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.pid")
 
 
 # ── Result dataclass ─────────────────────────────────────────────

scripts/contradiction_resolver.py

@@ -151,7 +151,7 @@ def _encode_via_daemon(text):
     import socket as _sock
     import base64 as _b64
     uid = os.getuid() if hasattr(os, "getuid") else os.getpid()
-    sock_path = f"/tmp/b12-embed-{uid}.sock"
+    sock_path = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{uid}.sock")
     if not os.path.exists(sock_path):
         return None
     s = None

scripts/embedding_backfill.py

@@ -34,7 +34,7 @@ def get_db_path():
 def daemon_request(texts):
     """Request embeddings from the embed daemon via Unix socket."""
     _uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-    sock_path = f"/tmp/b12-embed-{_uid}.sock"
+    sock_path = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.sock")
     if not os.path.exists(sock_path):
         return None
     try:
@@ -244,7 +244,7 @@ def main():
 
     # Check daemon is running
     _uid2 = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-    sock_path = f"/tmp/b12-embed-{_uid2}.sock"
+    sock_path = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid2}.sock")
     if not os.path.exists(sock_path):
         print("Embed daemon not running. Start a Claude Code session first,")
         print("or run: ~/.local/b12-venv/bin/python3 scripts/embed_daemon.py &")

scripts/graph_enrich.py

@@ -50,8 +50,8 @@
 LOG_DIR = os.path.join(os.environ.get('B12_DATA_DIR', os.path.expanduser('~/.B12')), 'memory-logs')
 _UID = os.getuid() if hasattr(os, 'getuid') else os.getpid()
 # Hardcode /tmp/ — macOS TMPDIR varies per session
-DAEMON_SOCK = f"/tmp/b12-embed-{_UID}.sock"
-DAEMON_PID = f"/tmp/b12-embed-{_UID}.pid"
+DAEMON_SOCK = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.sock")
+DAEMON_PID = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.pid")
 
 MAX_MEMORIES_PER_RUN = 50
 NEIGHBOR_K = 5

scripts/hook_adapter.py

@@ -306,7 +306,7 @@ def _log_tool_usage(self, tool_name: str, tool_input: dict):
     def _daemon_request(op: str, **kwargs) -> dict | None:
         """Send a request to the embed daemon."""
         _uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-        sock_path = f"/tmp/b12-embed-{_uid}.sock"
+        sock_path = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_uid}.sock")
         if not os.path.exists(sock_path):
             return None
         try:

scripts/memory_refine.py

@@ -15,7 +15,7 @@ def scrub_pii(value: str) -> str:
 
 # Daemon socket path (same as b12_mcp_server.py)
 _UID = os.getuid() if hasattr(os, 'getuid') else os.getpid()
-SOCK_PATH = f"/tmp/b12-embed-{_UID}.sock"
+SOCK_PATH = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{_UID}.sock")
 
 
 def daemon_request(op: str, **kwargs) -> dict | None:

scripts/surfacing_engine.py

@@ -276,7 +276,7 @@ def _daemon_search(query: str, db_path: str, limit: int = 20) -> list[dict]:
     """
     uid = os.getuid() if hasattr(os, 'getuid') else os.getpid()
     # Hardcode /tmp/ — macOS TMPDIR varies per session, causing socket mismatch
-    sock_path = f"/tmp/b12-embed-{uid}.sock"
+    sock_path = os.path.join(os.environ.get("B12_EMBED_RUNTIME_DIR", "/tmp"), f"b12-embed-{uid}.sock")
 
     try:
         s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)