fix(storage): scope conversation-history search by project tenancy

search_history / search_history_recent searched every workstream's rows
regardless of who asked. Pre-projects that matched the trusted-team
deployment shape; with private projects (062) it became a cross-tenant
read — the recall tool and /history returned private-project rows to
non-members.

Both methods take a keyword-only user_id (protocol, sqlite, postgresql)
scoped by one portable SQL predicate (HISTORY_VISIBILITY_SCOPE_SQL)
mirroring WorkstreamProjectVisibility: a row hides only when its
workstream links to an existing private project and the user is neither
the workstream creator, the project owner, nor a member. Applied in SQL
so limit/offset pagination stays honest; COALESCE guards the
NULL-creator row, which plain <> would leak.

The recall tool pins the scope identity at prepare time (the mcp_user_id
discipline) and fails loudly on an unpinned item; /history scopes to the
acting user; user_id=None (single-user CLI lanes) stays unscoped.

Tests: cross-backend visibility matrix, ws_visible parity pin,
marker-exclusion composition, LIKE-fallback path, prepare-pin plumbing.

tests/test_search_history_visibility.py

@@ -0,0 +1,205 @@
+"""Tenancy scoping for conversation-history search (recall tool + /history).
+
+``search_history`` / ``search_history_recent`` used to search every
+workstream's rows regardless of who asked — with private projects
+(migration 062) that is a cross-tenant read.  The SQL predicate
+(``HISTORY_VISIBILITY_SCOPE_SQL``) mirrors ``WorkstreamProjectVisibility``
+(core.auth), THE statement of the tenancy rule: a row is hidden only when
+its workstream links to an EXISTING project whose visibility is private and
+the searcher is neither the workstream creator, the project owner, nor a
+member.  Covered here:
+
+- unscoped (``user_id=None``) stays tenant-wide — single-user CLI back-compat;
+- trusted-team default: no-project rows are visible across users;
+- private project: hidden from strangers; visible to the workstream creator,
+  the project owner, and members — in both search and recent;
+- public project and dangling project link stay visible;
+- a NULL-creator workstream in a private project hides (COALESCE guard);
+- compaction markers stay excluded under scoping;
+- the sqlite LIKE fallback path applies the same predicate;
+- parity: SQL verdicts match ``ws_visible`` across the case matrix, so the
+  two statements of the rule cannot drift silently;
+- session plumbing: ``_prepare_recall`` pins the scope at prepare time,
+  ``_exec_recall`` searches with the pinned identity and refuses to run
+  unpinned.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from tests._session_helpers import make_session
+from turnstone.core.auth import WorkstreamProjectVisibility
+
+NEEDLE = "zebrafinch"
+
+
+def _ws(st, ws_id: str, owner: str | None, project_id: str | None = None) -> str:
+    st.register_workstream(
+        ws_id, user_id=owner, title="t", kind="interactive", project_id=project_id
+    )
+    st.save_message(ws_id, "user", f"{NEEDLE} in {ws_id}")
+    return ws_id
+
+
+def _found(st, user_id: str | None) -> set[str]:
+    return {r[1] for r in st.search_history(NEEDLE, limit=50, user_id=user_id)}
+
+
+def _recent(st, user_id: str | None) -> set[str]:
+    return {r[1] for r in st.search_history_recent(limit=50, user_id=user_id)}
+
+
+@pytest.fixture
+def world(storage_backend):
+    """One of each visibility case.
+
+    - ``ws_none``      — no project link (alice's)
+    - ``ws_dangling``  — links a project that does not exist (bob's)
+    - ``ws_public``    — public project, owned by alice
+    - ``ws_priv_own``  — private project ``P`` (owner alice), ws created by alice
+    - ``ws_priv_mem``  — private project ``P``, ws created by member bob
+    - ``ws_priv_other``— private project ``Q`` (owner dave, no members)
+    """
+    st = storage_backend
+    st.create_project("pub", "Pub", owner_id="alice", visibility="public")
+    st.create_project("P", "P", owner_id="alice", visibility="private")
+    st.create_project("Q", "Q", owner_id="dave", visibility="private")
+    st.add_project_member("P", "bob")
+    _ws(st, "ws_none", "alice")
+    _ws(st, "ws_dangling", "bob", project_id="ghost")
+    _ws(st, "ws_public", "alice", project_id="pub")
+    _ws(st, "ws_priv_own", "alice", project_id="P")
+    _ws(st, "ws_priv_mem", "bob", project_id="P")
+    _ws(st, "ws_priv_other", "dave", project_id="Q")
+    return st
+
+
+ALL_WS = {"ws_none", "ws_dangling", "ws_public", "ws_priv_own", "ws_priv_mem", "ws_priv_other"}
+
+
+class TestSearchHistoryScope:
+    def test_unscoped_stays_tenant_wide(self, world):
+        """CLI back-compat: ``user_id=None`` applies no filter."""
+        assert _found(world, None) == ALL_WS
+        assert _recent(world, None) == ALL_WS
+
+    def test_stranger_loses_only_private_rows(self, world):
+        """Trusted-team default: everything visible except other people's
+        private-project workstreams."""
+        expected = ALL_WS - {"ws_priv_own", "ws_priv_mem", "ws_priv_other"}
+        assert _found(world, "carol") == expected
+        assert _recent(world, "carol") == expected
+
+    def test_project_owner_sees_all_project_rows(self, world):
+        """alice owns P: sees bob's ws in P too; still not dave's Q."""
+        assert _found(world, "alice") == ALL_WS - {"ws_priv_other"}
+
+    def test_member_sees_project_rows(self, world):
+        """bob is a member of P: sees alice's ws in P; still not Q."""
+        assert _found(world, "bob") == ALL_WS - {"ws_priv_other"}
+
+    def test_ws_creator_sees_own_row_in_private_project(self, world):
+        """dave is neither owner nor member of P — but Q's rows are his."""
+        assert "ws_priv_other" in _found(world, "dave")
+
+    def test_null_creator_private_ws_hides(self, storage_backend):
+        """A NULL-creator ws in a private project must hide, not leak: plain
+        ``<>`` goes NULL against a NULL creator and would drop the row from
+        the hide-subquery (the COALESCE guard in the predicate)."""
+        st = storage_backend
+        st.create_project("P", "P", owner_id="alice", visibility="private")
+        _ws(st, "ws_orphan_creator", None, project_id="P")
+        assert _found(st, "carol") == set()
+        assert _found(st, "alice") == {"ws_orphan_creator"}  # project owner
+
+    def test_markers_stay_excluded_under_scope(self, world):
+        """The compaction-marker exclusion composes with the tenancy scope."""
+        world.save_message(
+            "ws_none",
+            "assistant",
+            f"{NEEDLE} SUMMARY",
+            source="compaction",
+            meta='{"watermark": 1}',
+        )
+        rows = world.search_history(NEEDLE, limit=50, user_id="alice")
+        assert not any("SUMMARY" in (r[3] or "") for r in rows)
+
+    def test_like_fallback_applies_same_predicate(self, world):
+        """The sqlite non-FTS path must scope identically."""
+        if not hasattr(world, "_fts5_available"):
+            pytest.skip("LIKE fallback is sqlite-only")
+        world._fts5_available = False
+        expected = ALL_WS - {"ws_priv_own", "ws_priv_mem", "ws_priv_other"}
+        assert _found(world, "carol") == expected
+
+
+class TestParityWithWsVisible:
+    """The SQL predicate and ``WorkstreamProjectVisibility`` are two
+    statements of one rule; this pins them together so neither can drift
+    without failing here."""
+
+    # (ws_id, creator, project_id) — mirrors the ``world`` fixture rows.
+    MATRIX = [
+        ("ws_none", "alice", None),
+        ("ws_dangling", "bob", "ghost"),
+        ("ws_public", "alice", "pub"),
+        ("ws_priv_own", "alice", "P"),
+        ("ws_priv_mem", "bob", "P"),
+        ("ws_priv_other", "dave", "Q"),
+    ]
+
+    @pytest.mark.parametrize("searcher", ["alice", "bob", "carol", "dave"])
+    def test_sql_matches_python_predicate(self, world, searcher):
+        vis = WorkstreamProjectVisibility(searcher, storage=world)
+        expected = {
+            ws_id
+            for ws_id, creator, project_id in self.MATRIX
+            if vis.ws_visible(project_id, ws_owner=creator or "")
+        }
+        assert _found(world, searcher) == expected
+        assert _recent(world, searcher) == expected
+
+
+class TestRecallScopePlumbing:
+    def _recorder(self, calls):
+        def fake_search_history(query, limit=20, offset=0, *, user_id=None):
+            calls.append(user_id)
+            return []
+
+        return fake_search_history
+
+    def test_prepare_pins_owner_without_acting_user(self):
+        session = make_session(user_id="owner")
+        item = session._prepare_recall("c1", {"query": "x"})
+        assert item["scope_user_id"] == "owner"
+
+    def test_prepare_pins_acting_user_over_owner(self):
+        session = make_session(user_id="owner")
+        session.bind_acting_user("driver")
+        item = session._prepare_recall("c1", {"query": "x"})
+        assert item["scope_user_id"] == "driver"
+
+    def test_prepare_pins_none_for_single_user_lanes(self):
+        session = make_session()  # user_id defaults to "" — CLI lane
+        item = session._prepare_recall("c1", {"query": "x"})
+        assert item["scope_user_id"] is None
+
+    def test_exec_searches_as_pinned_user(self, monkeypatch):
+        calls: list[str | None] = []
+        monkeypatch.setattr("turnstone.core.session.search_history", self._recorder(calls))
+        session = make_session(user_id="owner")
+        item = session._prepare_recall("c1", {"query": "x"})
+        session._exec_recall(item)
+        assert calls == ["owner"]
+
+    def test_exec_refuses_unpinned_item(self, monkeypatch):
+        """Fail loudly rather than fall back to a tenant-wide search."""
+        calls: list[str | None] = []
+        monkeypatch.setattr("turnstone.core.session.search_history", self._recorder(calls))
+        session = make_session(user_id="owner")
+        item = session._prepare_recall("c1", {"query": "x"})
+        del item["scope_user_id"]
+        with pytest.raises(KeyError):
+            session._exec_recall(item)
+        assert calls == []

turnstone/core/memory.py

@@ -641,19 +641,28 @@ def update_workstream_title(ws_id: str, title: str) -> None:
 # -- Conversation search -------------------------------------------------------
 
 
-def search_history(query: str, limit: int = 20, offset: int = 0) -> list[Any]:
-    """Search conversation history."""
+def search_history(
+    query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None
+) -> list[Any]:
+    """Search conversation history.
+
+    ``user_id`` scopes rows by project tenancy (private-project workstreams
+    hidden unless creator/owner/member — see
+    :meth:`StorageBackend.search_history`); ``None`` = unscoped, for
+    single-user lanes only.
+    """
     try:
-        return get_storage().search_history(query, limit, offset)
+        return get_storage().search_history(query, limit, offset, user_id=user_id)
     except Exception:
         log.warning("Failed to search history", exc_info=True)
         return []
 
 
-def search_history_recent(limit: int = 20) -> list[Any]:
-    """Return most recent conversation messages."""
+def search_history_recent(limit: int = 20, *, user_id: str | None = None) -> list[Any]:
+    """Return most recent conversation messages, tenancy-scoped like
+    :func:`search_history`."""
     try:
-        return get_storage().search_history_recent(limit)
+        return get_storage().search_history_recent(limit, user_id=user_id)
     except Exception:
         log.warning("Failed to search recent history", exc_info=True)
         return []

turnstone/core/session.py

@@ -4502,6 +4502,20 @@ class ChatSession:
         """
         return self._acting_user_id or self._mcp_user_id
 
+    def _history_scope_user_id(self) -> str | None:
+        """Identity that scopes conversation-history reads (recall tool,
+        ``/history``).
+
+        The acting user when one is bound — on a shared workstream the
+        search runs with the visibility of whoever is driving the turn —
+        otherwise the session owner.  ``None`` (CLI / eval / internal
+        single-user lanes) leaves history unscoped.  Deliberately no
+        admin/service bypass: the model-facing recall tool always reads as
+        a plain user, even when an admin is driving — bypass is a surface
+        property (cluster inspect), not a principal property.
+        """
+        return self._acting_user_id or self._user_id or None
+
     def bind_acting_user(self, user_id: str) -> None:
         """Bind the authenticated initiator of the current turn.
 
@@ -11862,6 +11876,11 @@ class ChatSession:
             "query": query,
             "limit": max(1, min(limit, 50)),
             "offset": max(0, offset),
+            # Pin the tenancy scope at prepare time: an item that sits in
+            # the queue must search as the user whose turn requested it,
+            # not whoever binds the session later (same discipline as
+            # ``mcp_user_id`` in ``_prepare_mcp_tool``).
+            "scope_user_id": self._history_scope_user_id(),
         }
 
     # -- skill prepare/execute -------------------------------------------------
@@ -13557,11 +13576,13 @@ class ChatSession:
         return call_id, msg
 
     def _exec_recall(self, item: dict[str, Any]) -> tuple[str, str]:
-        """Search conversation history."""
+        """Search conversation history, scoped to the prepare-time user."""
         call_id = item["call_id"]
         query, limit, offset = item["query"], item["limit"], item.get("offset", 0)
 
-        conv_rows = search_history(query, limit, offset)
+        # KeyError on a missing pin is deliberate — an unpinned item must
+        # fail loudly, not fall back to an unscoped (tenant-wide) search.
+        conv_rows = search_history(query, limit, offset, user_id=item["scope_user_id"])
         if conv_rows:
             lines = []
             for ts, sid, role, content, tool_name in conv_rows:
@@ -14399,7 +14420,7 @@ class ChatSession:
         elif cmd == "/history":
             query = arg.strip() if arg else None
             if query:
-                rows = search_history(query, limit=20)
+                rows = search_history(query, limit=20, user_id=self._history_scope_user_id())
                 if not rows:
                     self.ui.on_info(f"No results for {query!r}")
                 else:
@@ -14411,7 +14432,7 @@ class ChatSession:
                     self.ui.on_info("\n".join(lines))
             else:
                 # Show recent conversations (last 20 messages)
-                rows = search_history_recent(limit=20)
+                rows = search_history_recent(limit=20, user_id=self._history_scope_user_id())
                 if not rows:
                     self.ui.on_info("No conversation history yet.")
                 else:

turnstone/core/storage/_postgresql.py

@@ -80,6 +80,9 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     HEURISTIC_RULE_MUTABLE as _HEURISTIC_RULE_MUTABLE,
 )
+from turnstone.core.storage._utils import (
+    HISTORY_VISIBILITY_SCOPE_SQL as _HISTORY_SCOPE_SQL,
+)
 from turnstone.core.storage._utils import (
     LIKE_ESCAPE as _LIKE_ESCAPE,
 )
@@ -1199,11 +1202,19 @@ class PostgreSQLBackend:
 
     # -- Conversation search ---------------------------------------------------
 
-    def search_history(self, query: str, limit: int = 20, offset: int = 0) -> list[Any]:
+    def search_history(
+        self, query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None
+    ) -> list[Any]:
         if not query or not query.strip():
             return []
         capped = min(int(limit), 100)
         capped_offset = max(0, int(offset))
+        # Project-tenancy scope (see HISTORY_VISIBILITY_SCOPE_SQL): applied in
+        # SQL, not post-filtered in Python, so limit/offset pagination stays
+        # honest — a page never silently shrinks because hidden rows were
+        # fetched then dropped.
+        scope_sql = _HISTORY_SCOPE_SQL if user_id is not None else ""
+        scope_params = {"scope_user": user_id} if user_id is not None else {}
         with self._conn() as conn:
             # Use PostgreSQL full-text search if search_vector column exists
             try:
@@ -1218,7 +1229,8 @@ class PostgreSQLBackend:
                             # summary artifacts); IS DISTINCT FROM is NULL-safe so
                             # normal rows (_source NULL) are not dropped.
                             "AND c._source IS DISTINCT FROM :compaction_source "
-                            "ORDER BY ts_rank(to_tsvector('english', COALESCE(c.content, '')), "
+                            + scope_sql
+                            + "ORDER BY ts_rank(to_tsvector('english', COALESCE(c.content, '')), "
                             "   plainto_tsquery('english', :query)) DESC "
                             "LIMIT :limit OFFSET :offset"
                         ),
@@ -1227,6 +1239,7 @@ class PostgreSQLBackend:
                             "compaction_source": _COMPACTION_SOURCE,
                             "limit": capped,
                             "offset": capped_offset,
+                            **scope_params,
                         },
                     ).fetchall()
                 )
@@ -1235,32 +1248,37 @@ class PostgreSQLBackend:
                 return list(
                     conn.execute(
                         sa.text(
-                            "SELECT timestamp, ws_id, role, content, tool_name "
-                            "FROM conversations WHERE content ILIKE :pattern "
-                            "AND _source IS DISTINCT FROM :compaction_source "
-                            "ORDER BY timestamp DESC LIMIT :limit OFFSET :offset"
+                            "SELECT c.timestamp, c.ws_id, c.role, c.content, c.tool_name "
+                            "FROM conversations c WHERE c.content ILIKE :pattern "
+                            "AND c._source IS DISTINCT FROM :compaction_source "
+                            + scope_sql
+                            + "ORDER BY c.timestamp DESC LIMIT :limit OFFSET :offset"
                         ),
                         {
                             "pattern": f"%{query}%",
                             "compaction_source": _COMPACTION_SOURCE,
                             "limit": capped,
                             "offset": capped_offset,
+                            **scope_params,
                         },
                     ).fetchall()
                 )
 
-    def search_history_recent(self, limit: int = 20) -> list[Any]:
+    def search_history_recent(self, limit: int = 20, *, user_id: str | None = None) -> list[Any]:
         capped = min(limit, 100)
+        scope_sql = _HISTORY_SCOPE_SQL if user_id is not None else ""
+        scope_params = {"scope_user": user_id} if user_id is not None else {}
         with self._conn() as conn:
             return list(
                 conn.execute(
                     sa.text(
-                        "SELECT timestamp, ws_id, role, content, tool_name "
-                        "FROM conversations "
-                        "WHERE _source IS DISTINCT FROM :compaction_source "
-                        "ORDER BY timestamp DESC LIMIT :limit"
+                        "SELECT c.timestamp, c.ws_id, c.role, c.content, c.tool_name "
+                        "FROM conversations c "
+                        "WHERE c._source IS DISTINCT FROM :compaction_source "
+                        + scope_sql
+                        + "ORDER BY c.timestamp DESC LIMIT :limit"
                     ),
-                    {"limit": capped, "compaction_source": _COMPACTION_SOURCE},
+                    {"limit": capped, "compaction_source": _COMPACTION_SOURCE, **scope_params},
                 ).fetchall()
             )
 

turnstone/core/storage/_protocol.py

@@ -777,12 +777,29 @@ class StorageBackend(Protocol):
 
     # -- Conversation search ---------------------------------------------------
 
-    def search_history(self, query: str, limit: int = 20, offset: int = 0) -> list[Any]:
-        """Search conversation history. Returns (timestamp, ws_id, role, content, tool_name)."""
+    def search_history(
+        self, query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None
+    ) -> list[Any]:
+        """Search conversation history. Returns (timestamp, ws_id, role, content, tool_name).
+
+        ``user_id`` scopes results by project tenancy: rows are dropped when
+        their workstream sits in an existing PRIVATE project and *user_id* is
+        neither the workstream creator, the project owner, nor a member.
+        Everything else — no project link, dangling link, non-private project
+        — stays visible (trusted-team default).  The SQL predicate mirrors
+        ``WorkstreamProjectVisibility`` in ``core.auth`` (THE statement of the
+        rule); ``tests/test_search_history_visibility.py`` pins the parity.
+        ``None`` (default) applies no scoping — correct only for single-user
+        lanes (local CLI); authenticated surfaces MUST pass the acting user.
+        """
         ...
 
-    def search_history_recent(self, limit: int = 20) -> list[Any]:
-        """Return most recent conversation messages."""
+    def search_history_recent(self, limit: int = 20, *, user_id: str | None = None) -> list[Any]:
+        """Return most recent conversation messages.
+
+        ``user_id`` scopes rows by project tenancy exactly as in
+        :meth:`search_history`; ``None`` applies no scoping.
+        """
         ...
 
     # -- User identity operations -----------------------------------------------

turnstone/core/storage/_sqlite.py

@@ -80,6 +80,9 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     HEURISTIC_RULE_MUTABLE as _HEURISTIC_RULE_MUTABLE,
 )
+from turnstone.core.storage._utils import (
+    HISTORY_VISIBILITY_SCOPE_SQL as _HISTORY_SCOPE_SQL,
+)
 from turnstone.core.storage._utils import (
     LIKE_ESCAPE as _LIKE_ESCAPE,
 )
@@ -1374,11 +1377,19 @@ class SQLiteBackend:
 
     # -- Conversation search ---------------------------------------------------
 
-    def search_history(self, query: str, limit: int = 20, offset: int = 0) -> list[Any]:
+    def search_history(
+        self, query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None
+    ) -> list[Any]:
         if not query or not query.strip():
             return []
         capped = min(int(limit), 100)
         capped_offset = max(0, int(offset))
+        # Project-tenancy scope (see HISTORY_VISIBILITY_SCOPE_SQL): applied in
+        # SQL, not post-filtered in Python, so limit/offset pagination stays
+        # honest — a page never silently shrinks because hidden rows were
+        # fetched then dropped.
+        scope_sql = _HISTORY_SCOPE_SQL if user_id is not None else ""
+        scope_params = {"scope_user": user_id} if user_id is not None else {}
         with self._conn() as conn:
             if self._fts5_available:
                 return list(
@@ -1392,45 +1403,52 @@ class SQLiteBackend:
                             # summary artifacts); normal rows store _source NULL,
                             # so the filter must be NULL-safe or it drops everything.
                             "AND (c._source IS NULL OR c._source <> :compaction_source) "
-                            "ORDER BY f.rank ASC LIMIT :limit OFFSET :offset"
+                            + scope_sql
+                            + "ORDER BY f.rank ASC LIMIT :limit OFFSET :offset"
                         ),
                         {
                             "query": _fts5_query(query),
                             "compaction_source": _COMPACTION_SOURCE,
                             "limit": capped,
                             "offset": capped_offset,
+                            **scope_params,
                         },
                     ).fetchall()
                 )
             return list(
                 conn.execute(
                     sa.text(
-                        "SELECT timestamp, ws_id, role, content, tool_name "
-                        "FROM conversations WHERE content LIKE :pattern ESCAPE '\\' "
-                        "AND (_source IS NULL OR _source <> :compaction_source) "
-                        "ORDER BY timestamp DESC LIMIT :limit OFFSET :offset"
+                        "SELECT c.timestamp, c.ws_id, c.role, c.content, c.tool_name "
+                        "FROM conversations c WHERE c.content LIKE :pattern ESCAPE '\\' "
+                        "AND (c._source IS NULL OR c._source <> :compaction_source) "
+                        + scope_sql
+                        + "ORDER BY c.timestamp DESC LIMIT :limit OFFSET :offset"
                     ),
                     {
                         "pattern": f"%{_escape_like(query)}%",
                         "compaction_source": _COMPACTION_SOURCE,
                         "limit": capped,
                         "offset": capped_offset,
+                        **scope_params,
                     },
                 ).fetchall()
             )
 
-    def search_history_recent(self, limit: int = 20) -> list[Any]:
+    def search_history_recent(self, limit: int = 20, *, user_id: str | None = None) -> list[Any]:
         capped = min(limit, 100)
+        scope_sql = _HISTORY_SCOPE_SQL if user_id is not None else ""
+        scope_params = {"scope_user": user_id} if user_id is not None else {}
         with self._conn() as conn:
             return list(
                 conn.execute(
                     sa.text(
-                        "SELECT timestamp, ws_id, role, content, tool_name "
-                        "FROM conversations "
-                        "WHERE (_source IS NULL OR _source <> :compaction_source) "
-                        "ORDER BY timestamp DESC LIMIT :limit"
+                        "SELECT c.timestamp, c.ws_id, c.role, c.content, c.tool_name "
+                        "FROM conversations c "
+                        "WHERE (c._source IS NULL OR c._source <> :compaction_source) "
+                        + scope_sql
+                        + "ORDER BY c.timestamp DESC LIMIT :limit"
                     ),
-                    {"limit": capped, "compaction_source": _COMPACTION_SOURCE},
+                    {"limit": capped, "compaction_source": _COMPACTION_SOURCE, **scope_params},
                 ).fetchall()
             )
 

turnstone/core/storage/_utils.py

@@ -1002,6 +1002,38 @@ def recover_trajectory(turns: list[Turn]) -> list[Turn]:
 COMPACTION_SOURCE = "compaction"
 COMPACTION_SUMMARY_LABEL = "[Conversation summary]"
 
+# ---------------------------------------------------------------------------
+# History-search tenancy scope
+# ---------------------------------------------------------------------------
+# SQL mirror of ``WorkstreamProjectVisibility`` (core.auth) — THE statement of
+# who may see a workstream's rows.  A conversation row is hidden from
+# ``:scope_user`` only when its workstream links to an EXISTING project whose
+# visibility is 'private' and the user is neither the workstream creator, the
+# project owner, nor a member.  No project link, a dangling link (project row
+# deleted), and non-private projects all stay visible — the trusted-team
+# default.  ``COALESCE(w.user_id, '')`` makes a NULL creator hide (not leak):
+# plain ``<>`` would go NULL and drop the row from the hide-subquery.  Callers
+# never pass an empty ``:scope_user`` (empty scopes to None = unscoped), so
+# the COALESCE sentinel cannot collide with a real principal.  Portable across
+# SQLite and PostgreSQL; expects the conversations table aliased ``c``.
+# ``tests/test_search_history_visibility.py`` pins parity with the Python
+# predicate — change either side only in lockstep.
+
+HISTORY_VISIBILITY_SCOPE_SQL = (
+    "AND NOT EXISTS ("
+    "    SELECT 1 FROM workstreams w"
+    "    JOIN projects p ON p.project_id = w.project_id"
+    "    WHERE w.ws_id = c.ws_id"
+    "      AND p.visibility = 'private'"
+    "      AND COALESCE(w.user_id, '') <> :scope_user"
+    "      AND p.owner_id <> :scope_user"
+    "      AND NOT EXISTS ("
+    "          SELECT 1 FROM project_members pm"
+    "          WHERE pm.project_id = w.project_id AND pm.user_id = :scope_user"
+    "      )"
+    ") "
+)
+
 
 def _is_compaction_marker(row: Any) -> bool:
     """True when a stored row is a compaction checkpoint marker (``_source`` = row index 7)."""