test: refuse SQLite connections into a checkout's data/ directory (#355)

src/pinky_daemon/agent_registry.py

@@ -2803,7 +2803,11 @@ def _safe(content: str, source: str) -> str | None:
         # Inject learned user profiles (from dream consolidation)
         try:
             from pinky_daemon.user_profile_store import UserProfileStore
-            profile_store = UserProfileStore()
+            # Sibling of this registry's own DB rather than the default
+            # relative path, which would resolve against the daemon's cwd.
+            profile_store = UserProfileStore(
+                db_path=str(Path(self._db_path).parent / "user_profiles.db"),
+            )
             known_users = profile_store.get_all_users()
             profile_sections = []
             for uid in known_users:

src/pinky_daemon/api.py

@@ -4295,7 +4295,11 @@ async def _trigger_librarian() -> None:
         from pinky_daemon.voice_routes import set_dependencies as _voice_set_deps
         from pinky_daemon.voice_store import VoiceStore
 
-        _voice_store = VoiceStore(db_path="data/voice_calls.db")
+        # Sibling of the main DB, not a hardcoded relative path: every other
+        # store here derives from ``db_path``, and a literal "data/..." makes
+        # an API created with a custom db_path write to the *cwd's* data dir
+        # anyway. Keeping the basename means the deployed file is unchanged.
+        _voice_store = VoiceStore(db_path=str(Path(db_path).parent / "voice_calls.db"))
         _voice_base_url = (
             agents.get_setting("PINKY_BASE_URL")
             or os.environ.get("PINKY_BASE_URL", "")
@@ -12058,7 +12062,10 @@ async def set_owner_profile(req: OwnerProfileRequest):
 
     from pinky_daemon.user_profile_store import UserProfileStore
 
-    user_profiles = UserProfileStore()
+    # Sibling of the main DB — see the VoiceStore note above.
+    user_profiles = UserProfileStore(
+        db_path=str(Path(db_path).parent / "user_profiles.db"),
+    )
 
     from pinky_daemon.routes.user_profiles import router as _user_profiles_router
     from pinky_daemon.routes.user_profiles import set_dependencies as _user_profiles_set_deps

src/pinky_daemon/dream_runner.py

@@ -656,7 +656,11 @@ def _extract_user_profiles(self, dream_output: str) -> int:
         # Lazy import to avoid circular deps
         from pinky_daemon.user_profile_store import ProfileEntry, UserProfileStore
 
-        store = UserProfileStore()
+        # Sibling of the dream state DB rather than the default relative
+        # path, which would resolve against the daemon's cwd.
+        store = UserProfileStore(
+            db_path=str(Path(self._db_path).parent / "user_profiles.db"),
+        )
         count = 0
 
         valid_categories = {
@@ -718,7 +722,11 @@ def _extract_user_relationships(self, dream_output: str) -> int:
 
         from pinky_daemon.user_profile_store import Relationship, UserProfileStore
 
-        store = UserProfileStore()
+        # Sibling of the dream state DB rather than the default relative
+        # path, which would resolve against the daemon's cwd.
+        store = UserProfileStore(
+            db_path=str(Path(self._db_path).parent / "user_profiles.db"),
+        )
         rels = []
         for rd in rels_data:
             from_id = rd.get("from_chat_id", "")

tests/conftest.py

@@ -20,11 +20,20 @@
 ``pytestmark`` in that file). For those, the conftest leaves
 ``TestClient`` alone and the test sets up its own auth state via
 ``monkeypatch.setenv`` + ``/auth/setup``.
+
+Production-database guard (#355):
+
+``_forbid_production_db_writes`` wraps ``sqlite3.connect`` for the whole
+session and raises if a test opens a database under a checkout's
+``data/`` directory. See that fixture's docstring for the why.
 """
 
 from __future__ import annotations
 
 import os
+import sqlite3
+from pathlib import Path
+from urllib.parse import unquote, urlparse
 
 import pytest
 from fastapi.testclient import TestClient
@@ -43,6 +52,90 @@ def pytest_configure(config: pytest.Config) -> None:
     )
 
 
+def _sqlite_target_path(database, uri: bool) -> Path | None:
+    """Best-effort resolution of a ``sqlite3.connect`` target to a real file.
+
+    Returns ``None`` for anything that is not an on-disk database
+    (``:memory:``, empty/temporary databases, ``mode=memory`` URIs, and
+    exotic argument types we don't want to guess about).
+    """
+    if isinstance(database, (bytes, bytearray)):
+        database = os.fsdecode(database)
+    if not isinstance(database, (str, os.PathLike)):
+        return None
+
+    target = os.fspath(database)
+    if target in ("", ":memory:"):
+        return None
+
+    if uri or target.startswith("file:"):
+        parsed = urlparse(target)
+        if parsed.scheme != "file":
+            return None
+        if "mode=memory" in (parsed.query or ""):
+            return None
+        target = unquote(parsed.path)
+        if target in ("", ":memory:"):
+            return None
+
+    try:
+        return Path(target).resolve()
+    except (OSError, ValueError):  # pragma: no cover - defensive
+        return None
+
+
+@pytest.fixture(autouse=True, scope="session")
+def _forbid_production_db_writes():
+    """Fail loudly if a test opens a SQLite file under a checkout's ``data/``.
+
+    Roughly two dozen stores default to a *relative* ``db_path``
+    (``data/tasks.db``, ``data/agents.db``, ...). A test that instantiates
+    one without an explicit path therefore reads and writes whatever
+    ``./data`` happens to be — which, when pytest is run from a live
+    deployment checkout, is the production database.
+
+    Rather than rewrite the defaults (invasive, production code) or chdir
+    the whole session into a tmpdir (breaks tests that rely on
+    repo-relative paths), this wraps ``sqlite3.connect`` for the duration
+    of the test session and raises on any target under
+    ``<cwd>/data`` or ``<repo root>/data``. The fix for a test that trips
+    it is always the same: pass an explicit ``db_path`` under ``tmp_path``.
+
+    Known limitation: this only covers connections opened in the pytest
+    process. Code executed in a subprocess is unaffected.
+    """
+    repo_data = (Path(__file__).resolve().parent.parent / "data").resolve()
+    original_connect = sqlite3.connect
+
+    def guarded_connect(database=None, *args, **kwargs):
+        uri = kwargs.get("uri", False)
+        target = _sqlite_target_path(database, uri)
+        if target is not None:
+            # cwd is recomputed per call and checked alongside the repo
+            # root because the two can differ: running pytest from a live
+            # deployment against another checkout's tests resolves the
+            # stores' relative defaults against the *deployment's* data/.
+            # A test that chdirs elsewhere first is refused too — the rule
+            # is simply "pass an explicit path", with no exceptions to
+            # reason about.
+            for guarded in {repo_data, (Path.cwd() / "data").resolve()}:
+                if target == guarded or target.is_relative_to(guarded):
+                    raise RuntimeError(
+                        f"Test tried to open a SQLite database inside a checkout's "
+                        f"data/ directory: {target}\n"
+                        f"That is the production database when pytest runs from a "
+                        f"live deployment. Pass an explicit db_path under tmp_path "
+                        f"instead of relying on the store's relative default."
+                    )
+        return original_connect(database, *args, **kwargs)
+
+    sqlite3.connect = guarded_connect
+    try:
+        yield
+    finally:
+        sqlite3.connect = original_connect
+
+
 @pytest.fixture(autouse=True, scope="session")
 def _ensure_test_session_secret():
     """Make sure PINKY_SESSION_SECRET is set for the whole test run.

tests/test_conftest_db_guard.py

@@ -0,0 +1,63 @@
+"""Tests for the conftest guard that keeps the suite off production DBs (#355)."""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+import pytest
+
+REPO_DATA = Path(__file__).resolve().parent.parent / "data"
+
+
+def test_relative_store_default_is_rejected():
+    """The exact shape ~24 stores use as their default db_path."""
+    with pytest.raises(RuntimeError, match="production database"):
+        sqlite3.connect("data/tasks.db")
+
+
+def test_absolute_path_into_repo_data_is_rejected():
+    with pytest.raises(RuntimeError, match="production database"):
+        sqlite3.connect(str(REPO_DATA / "agents" / "engineer" / "memory.db"))
+
+
+def test_path_object_is_rejected():
+    with pytest.raises(RuntimeError, match="production database"):
+        sqlite3.connect(REPO_DATA / "tasks.db")
+
+
+def test_readonly_uri_into_data_is_rejected():
+    """auth.py opens its DB via a ``file:...?mode=ro`` URI."""
+    with pytest.raises(RuntimeError, match="production database"):
+        sqlite3.connect("file:data/agents.db?mode=ro", uri=True)
+
+
+def test_tmp_path_is_allowed(tmp_path):
+    conn = sqlite3.connect(tmp_path / "fine.db")
+    conn.close()
+    assert (tmp_path / "fine.db").exists()
+
+
+def test_tmp_data_dir_is_allowed(tmp_path):
+    """A ``data/`` dir that isn't a checkout's is legitimate."""
+    (tmp_path / "data").mkdir()
+    conn = sqlite3.connect(tmp_path / "data" / "fine.db")
+    conn.close()
+
+
+def test_in_memory_is_allowed():
+    sqlite3.connect(":memory:").close()
+    sqlite3.connect("file::memory:?cache=shared", uri=True).close()
+
+
+def test_relative_data_path_rejected_even_after_chdir(tmp_path, monkeypatch):
+    """cwd is re-read per call, and ``<cwd>/data`` is guarded either way.
+
+    Chdir'ing into a tmpdir first is not an escape hatch: the rule is
+    "pass an explicit path", so a relative ``data/`` default stays refused.
+    """
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "data").mkdir()
+    with pytest.raises(RuntimeError, match="production database"):
+        sqlite3.connect("data/tasks.db")
+    assert not (tmp_path / "data" / "tasks.db").exists()

tests/test_daemon.py

@@ -474,14 +474,14 @@ def test_none_value(self):
 
 
 class TestDaemon:
-    def test_create_daemon(self):
-        config = DaemonConfig()
+    def test_create_daemon(self, tmp_path):
+        config = DaemonConfig(working_dir=str(tmp_path))
         with patch("pinky_daemon.claude_runner._find_claude_binary", return_value="/usr/bin/claude"):
             daemon = Daemon(config)
         assert daemon.is_running is False
 
-    def test_stats(self):
-        config = DaemonConfig()
+    def test_stats(self, tmp_path):
+        config = DaemonConfig(working_dir=str(tmp_path))
         with patch("pinky_daemon.claude_runner._find_claude_binary", return_value="/usr/bin/claude"):
             daemon = Daemon(config)
         stats = daemon.stats