feat(projects): enforce private-project workstream visibility server-side

Workstreams attached to a private project were listed and reachable for
every authenticated user — only the scope tier was checked. Add a
tenancy predicate (WorkstreamProjectVisibility: private → project
owner/members, the workstream's own creator, service scope, or
admin.cluster.inspect; public/dangling/no project → unchanged
trusted-team visibility; membership itself is the grant — deliberately
NOT gated on the project.read capability, which guards the management
API) and apply it at every surface:

- listings: saved sessions (project_id + owner tail-appended to
  list_workstreams_with_history on both backends), active list, node
  dashboard, console cluster list (pre-pagination via a collector
  row_filter so totals stay honest), node detail
- console tier-1 SSE: per-connection snapshot filtering + a hidden-set
  for sparse follow-up events; ws_created project lookups run on the
  executor, membership changes take effect on reconnect
- row access: resolve_workstream_owner 403s private-project rows for
  non-members, covering every interactive ws-scoped verb via
  tenant_check (console coordinator lane stays on its privileged
  admin.coordinator gate)
- create: ensure_project_attachable gates explicit and parent-inherited
  project_id on both create validators (unknown project 400s instead of
  minting a dangling link)

tests/test_project_workstream_visibility.py

@@ -0,0 +1,294 @@
+"""Private-project workstream visibility enforcement.
+
+Covers the tenancy predicate (:class:`WorkstreamProjectVisibility`), the
+create-time attach gate (:func:`ensure_project_attachable`), the row-access
+gate in :func:`resolve_workstream_owner`, and the saved-list filter in
+``_collect_saved_rows`` — the choke points that keep workstreams attached
+to a private project out of non-members' listings and 403 their direct
+access.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from turnstone.core.auth import (
+    WorkstreamProjectVisibility,
+    ensure_project_attachable,
+)
+
+pytestmark = pytest.mark.anyio
+
+
+def _fake_storage(
+    *,
+    visibility: str = "private",
+    owner: str = "alice",
+    members: tuple[str, ...] = (),
+    missing: bool = False,
+) -> MagicMock:
+    storage = MagicMock()
+    if missing:
+        storage.get_project.return_value = None
+    else:
+        storage.get_project.return_value = {
+            "project_id": "p1",
+            "name": "P1",
+            "owner_id": owner,
+            "visibility": visibility,
+            "state": "active",
+        }
+    storage.is_project_member.side_effect = lambda pid, uid: uid in members
+    return storage
+
+
+class _FakeAuth:
+    def __init__(
+        self,
+        user_id: str,
+        scopes: tuple[str, ...] = (),
+        permissions: tuple[str, ...] = (),
+    ) -> None:
+        self.user_id = user_id
+        self._scopes = set(scopes)
+        self._permissions = set(permissions)
+
+    def has_scope(self, scope: str) -> bool:
+        return scope in self._scopes
+
+    def has_permission(self, permission: str) -> bool:
+        return permission in self._permissions
+
+
+def _request_for(
+    uid: str,
+    scopes: tuple[str, ...] = (),
+    permissions: tuple[str, ...] = (),
+) -> Any:
+    return SimpleNamespace(state=SimpleNamespace(auth_result=_FakeAuth(uid, scopes, permissions)))
+
+
+class TestWsVisiblePredicate:
+    def test_no_project_always_visible(self) -> None:
+        vis = WorkstreamProjectVisibility("bob", storage=_fake_storage())
+        assert vis.ws_visible(None)
+        assert vis.ws_visible("")
+
+    def test_dangling_project_visible(self) -> None:
+        # Project deletion leaves ws links behind — no row, no privacy.
+        vis = WorkstreamProjectVisibility("bob", storage=_fake_storage(missing=True))
+        assert vis.ws_visible("p1")
+
+    def test_public_project_visible_to_anyone(self) -> None:
+        vis = WorkstreamProjectVisibility("bob", storage=_fake_storage(visibility="public"))
+        assert vis.ws_visible("p1")
+
+    def test_private_hidden_from_non_member(self) -> None:
+        vis = WorkstreamProjectVisibility("bob", storage=_fake_storage())
+        assert not vis.ws_visible("p1")
+
+    def test_private_visible_to_project_owner(self) -> None:
+        vis = WorkstreamProjectVisibility("alice", storage=_fake_storage())
+        assert vis.ws_visible("p1")
+
+    def test_private_visible_to_member(self) -> None:
+        vis = WorkstreamProjectVisibility("bob", storage=_fake_storage(members=("bob",)))
+        assert vis.ws_visible("p1")
+
+    def test_private_visible_to_ws_creator(self) -> None:
+        # A workstream's own creator never loses sight of it, even after
+        # a membership revoke leaves a legacy private-project link.
+        vis = WorkstreamProjectVisibility("bob", storage=_fake_storage())
+        assert vis.ws_visible("p1", ws_owner="bob")
+
+    def test_private_hidden_from_anonymous(self) -> None:
+        vis = WorkstreamProjectVisibility("", storage=_fake_storage())
+        assert not vis.ws_visible("p1")
+
+    def test_bypass_sees_everything(self) -> None:
+        vis = WorkstreamProjectVisibility("bob", bypass=True, storage=_fake_storage())
+        assert vis.ws_visible("p1")
+
+    def test_storage_error_fails_closed(self) -> None:
+        storage = MagicMock()
+        storage.get_project.side_effect = RuntimeError("db down")
+        vis = WorkstreamProjectVisibility("bob", storage=storage)
+        assert not vis.ws_visible("p1")
+
+    def test_project_rows_memoized(self) -> None:
+        storage = _fake_storage(visibility="public")
+        vis = WorkstreamProjectVisibility("bob", storage=storage)
+        assert vis.ws_visible("p1")
+        assert vis.ws_visible("p1")
+        assert storage.get_project.call_count == 1
+
+    def test_for_request_bypass_rules(self) -> None:
+        assert WorkstreamProjectVisibility.for_request(
+            _request_for("bob", scopes=("service",))
+        )._bypass
+        assert WorkstreamProjectVisibility.for_request(
+            _request_for("bob", permissions=("admin.cluster.inspect",))
+        )._bypass
+        assert not WorkstreamProjectVisibility.for_request(_request_for("bob"))._bypass
+
+
+class TestEnsureProjectAttachable:
+    def test_no_project_allowed(self) -> None:
+        assert ensure_project_attachable("bob", "", storage=_fake_storage()) is None
+
+    def test_unknown_project_is_400(self) -> None:
+        denied = ensure_project_attachable("bob", "p1", storage=_fake_storage(missing=True))
+        assert denied is not None and denied[0] == 400
+
+    def test_public_project_allowed(self) -> None:
+        assert (
+            ensure_project_attachable("bob", "p1", storage=_fake_storage(visibility="public"))
+            is None
+        )
+
+    def test_private_member_and_owner_allowed(self) -> None:
+        assert (
+            ensure_project_attachable("bob", "p1", storage=_fake_storage(members=("bob",))) is None
+        )
+        assert ensure_project_attachable("alice", "p1", storage=_fake_storage()) is None
+
+    def test_private_non_member_is_403(self) -> None:
+        denied = ensure_project_attachable("bob", "p1", storage=_fake_storage())
+        assert denied is not None and denied[0] == 403
+
+    def test_anonymous_private_is_403(self) -> None:
+        denied = ensure_project_attachable("", "p1", storage=_fake_storage())
+        assert denied is not None and denied[0] == 403
+
+    def test_storage_error_fails_closed(self) -> None:
+        storage = MagicMock()
+        storage.get_project.side_effect = RuntimeError("db down")
+        denied = ensure_project_attachable("bob", "p1", storage=storage)
+        assert denied is not None and denied[0] == 403
+
+
+class TestResolveWorkstreamOwnerProjectGate:
+    """Integration against the real (ephemeral) storage: the row-access
+    gate every interactive ws-scoped verb inherits via tenant_check."""
+
+    def _seed(self, *, member: bool) -> None:
+        from turnstone.core.memory import register_workstream
+        from turnstone.core.storage import get_storage
+
+        storage = get_storage()
+        storage.create_project("p1", "Secret", "alice")
+        if member:
+            storage.add_project_member("p1", "bob")
+        register_workstream("ws-priv", user_id="alice", project_id="p1")
+
+    def test_non_member_gets_403(self, tmp_db: str) -> None:
+        from turnstone.core.web_helpers import resolve_workstream_owner
+
+        self._seed(member=False)
+        owner, err = resolve_workstream_owner(_request_for("bob"), "ws-priv")
+        assert err is not None and err.status_code == 403
+
+    def test_member_resolves_owner(self, tmp_db: str) -> None:
+        from turnstone.core.web_helpers import resolve_workstream_owner
+
+        self._seed(member=True)
+        owner, err = resolve_workstream_owner(_request_for("bob"), "ws-priv")
+        assert err is None
+        assert owner == "alice"
+
+    def test_ws_creator_bypasses(self, tmp_db: str) -> None:
+        from turnstone.core.memory import register_workstream
+        from turnstone.core.storage import get_storage
+        from turnstone.core.web_helpers import resolve_workstream_owner
+
+        storage = get_storage()
+        storage.create_project("p1", "Secret", "alice")
+        # bob created a ws in alice's private project, then lost access —
+        # bob still reaches his own workstream.
+        register_workstream("ws-bob", user_id="bob", project_id="p1")
+        owner, err = resolve_workstream_owner(_request_for("bob"), "ws-bob")
+        assert err is None
+        assert owner == "bob"
+
+    def test_admin_inspect_bypasses(self, tmp_db: str) -> None:
+        from turnstone.core.web_helpers import resolve_workstream_owner
+
+        self._seed(member=False)
+        owner, err = resolve_workstream_owner(
+            _request_for("bob", permissions=("admin.cluster.inspect",)), "ws-priv"
+        )
+        assert err is None
+        assert owner == "alice"
+
+    def test_missing_ws_still_404s(self, tmp_db: str) -> None:
+        from turnstone.core.web_helpers import resolve_workstream_owner
+
+        owner, err = resolve_workstream_owner(_request_for("bob"), "nope")
+        assert err is not None and err.status_code == 404
+
+    def test_public_project_ws_resolves(self, tmp_db: str) -> None:
+        from turnstone.core.memory import register_workstream
+        from turnstone.core.storage import get_storage
+        from turnstone.core.web_helpers import resolve_workstream_owner
+
+        storage = get_storage()
+        storage.create_project("p1", "Open", "alice")
+        storage.update_project("p1", visibility="public")
+        register_workstream("ws-pub", user_id="alice", project_id="p1")
+        owner, err = resolve_workstream_owner(_request_for("bob"), "ws-pub")
+        assert err is None
+        assert owner == "alice"
+
+
+class TestSavedListFilter:
+    """The saved-sessions collector drops private-project rows server-side
+    and carries project_id on surviving rows (real ephemeral DB)."""
+
+    async def test_saved_rows_filtered_and_carry_project_id(self, tmp_db: str) -> None:
+        from turnstone.core.memory import register_workstream, save_message
+        from turnstone.core.session_routes import (
+            SessionEndpointConfig,
+            _collect_saved_rows,
+        )
+        from turnstone.core.storage import get_storage
+        from turnstone.core.workstream import WorkstreamKind
+
+        storage = get_storage()
+        storage.create_project("p1", "Secret", "alice")
+        storage.create_project("p2", "Open", "alice")
+        storage.update_project("p2", visibility="public")
+
+        register_workstream("ws-plain", user_id="alice")
+        register_workstream("ws-priv", user_id="alice", project_id="p1")
+        register_workstream("ws-pub", user_id="alice", project_id="p2")
+        register_workstream("ws-own", user_id="bob", project_id="p1")
+        for wid in ("ws-plain", "ws-priv", "ws-pub", "ws-own"):
+            save_message(wid, "user", "hello")
+
+        cfg = SessionEndpointConfig(
+            permission_gate=None,
+            manager_lookup=lambda request: (None, None),
+            tenant_check=None,
+            not_found_label="Workstream not found",
+            audit_action_prefix="workstream",
+            list_kind=WorkstreamKind.INTERACTIVE,
+            saved_state_filter=None,
+            saved_loaded_lookup=None,
+        )
+
+        rows = await _collect_saved_rows(cfg, _request_for("bob"))
+        ids = {r["ws_id"] for r in rows}
+        # bob: no membership in p1 — alice's private ws is dropped; the
+        # public-project ws, the project-less ws, and bob's own
+        # private-project ws all survive.
+        assert ids == {"ws-plain", "ws-pub", "ws-own"}
+        by_id = {r["ws_id"]: r for r in rows}
+        assert by_id["ws-pub"]["project_id"] == "p2"
+        assert by_id["ws-plain"]["project_id"] is None
+
+        rows_alice = await _collect_saved_rows(cfg, _request_for("alice"))
+        assert {r["ws_id"] for r in rows_alice} == {"ws-plain", "ws-priv", "ws-pub", "ws-own"}

tests/test_saved_handler_unified.py

@@ -6,7 +6,7 @@ for its L-shell dashboard, plus a regression guard for the single-kind
 :func:`turnstone.core.session_routes._collect_saved_rows`.
 
 Storage is mocked (``list_workstreams_with_history`` is patched to
-return synthetic 15-tuples) — no real or dev database is touched. The
+return synthetic 17-tuples) — no real or dev database is touched. The
 request is a :class:`unittest.mock.MagicMock`, matching how the
 body-level coordinator endpoint tests build request stubs; the saved
 path only reads ``request`` to pass it to ``saved_loaded_lookup`` /
@@ -41,7 +41,7 @@ pytestmark = pytest.mark.anyio
 # Column order from list_workstreams_with_history (keep in sync with the
 # storage SELECT): ws_id, alias, title, name, created, updated,
 # message_count, node_id, state, kind, model_alias, launch_skill,
-# child_count, context_tokens, context_window.
+# child_count, context_tokens, context_window, project_id, owner.
 def _row(
     ws_id: str,
     *,
@@ -49,8 +49,10 @@ def _row(
     kind: str,
     state: str = "closed",
     name: str | None = None,
+    project_id: str | None = None,
+    owner: str | None = None,
 ) -> tuple[Any, ...]:
-    """Build a synthetic storage row (15-tuple) for one workstream."""
+    """Build a synthetic storage row (17-tuple) for one workstream."""
     return (
         ws_id,
         None,  # alias
@@ -67,6 +69,8 @@ def _row(
         0,  # child_count
         1000,  # context_tokens
         4000,  # context_window
+        project_id,  # project_id
+        owner,  # owner user_id
     )
 
 
@@ -343,6 +347,7 @@ async def test_single_kind_saved_unchanged(monkeypatch: pytest.MonkeyPatch) -> N
         "child_count",
         "context_tokens",
         "context_ratio",
+        "project_id",
     }
 
 

turnstone/console/collector.py

@@ -932,6 +932,7 @@ class ClusterCollector:
         page: int = 1,
         per_page: int = 50,
         extra_rows: list[dict[str, Any]] | None = None,
+        row_filter: Callable[[dict[str, Any]], bool] | None = None,
     ) -> tuple[list[dict[str, Any]], int]:
         """Return filtered, sorted, paginated workstreams + total count.
 
@@ -939,6 +940,10 @@ class ClusterCollector:
         filter / sort / paginate — used by callers that contribute
         console-local rows (e.g. coordinator workstreams) that aren't
         tracked on any node's SSE stream.
+
+        ``row_filter`` (when provided) runs against the merged,
+        UNPAGINATED pool so dropped rows never skew ``total`` or page
+        boundaries — the private-project tenancy filter rides here.
         """
         with self._lock:
             all_ws = []
@@ -955,6 +960,10 @@ class ClusterCollector:
             if extra_rows:
                 all_ws.extend(dict(r) for r in extra_rows)
 
+        # Row-level tenancy filter first — before pagination math.
+        if row_filter is not None:
+            all_ws = [ws for ws in all_ws if row_filter(ws)]
+
         # Filter
         if state:
             all_ws = [ws for ws in all_ws if ws.get("state") == state]

turnstone/console/server.py

@@ -995,6 +995,8 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
 
 
 async def cluster_workstreams(request: Request) -> JSONResponse:
+    from turnstone.core.auth import WorkstreamProjectVisibility
+
     collector: ClusterCollector = request.app.state.collector
     params = dict(request.query_params)
     state = params.get("state")
@@ -1004,6 +1006,7 @@ async def cluster_workstreams(request: Request) -> JSONResponse:
     page = _parse_int(params, "page", 1, minimum=1)
     per_page = _parse_int(params, "per_page", 50, minimum=1, maximum=200)
     extra_rows = _coordinator_rows(request)
+    visibility = WorkstreamProjectVisibility.for_request(request)
     ws_list, total = collector.get_workstreams(
         state=state,
         node=node,
@@ -1012,6 +1015,9 @@ async def cluster_workstreams(request: Request) -> JSONResponse:
         page=page,
         per_page=per_page,
         extra_rows=extra_rows,
+        row_filter=lambda ws: visibility.ws_visible(
+            ws.get("project_id") or "", ws_owner=ws.get("user_id") or ""
+        ),
     )
     pages = math.ceil(total / per_page) if per_page > 0 else 0
     return JSONResponse(
@@ -1506,6 +1512,8 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
 
 
 async def cluster_node_detail(request: Request) -> JSONResponse:
+    from turnstone.core.auth import WorkstreamProjectVisibility
+
     collector: ClusterCollector = request.app.state.collector
     node_id = request.path_params["node_id"]
     nv = _validate_node_id(node_id)
@@ -1514,6 +1522,13 @@ async def cluster_node_detail(request: Request) -> JSONResponse:
     detail = collector.get_node_detail(node_id)
     if not detail:
         return JSONResponse({"error": "Node not found"}, status_code=404)
+    # Private-project tenancy — same predicate as the cluster list.
+    visibility = WorkstreamProjectVisibility.for_request(request)
+    detail["workstreams"] = [
+        ws
+        for ws in detail.get("workstreams", [])
+        if visibility.ws_visible(ws.get("project_id") or "", ws_owner=ws.get("user_id") or "")
+    ]
 
     # Attach metadata if available
     import json as _nd_json
@@ -1564,12 +1579,61 @@ async def cluster_snapshot(request: Request) -> JSONResponse:
 
 
 async def cluster_events_sse(request: Request) -> Response:
+    from turnstone.core.auth import WorkstreamProjectVisibility
+
     err = _collector_scope_error(request)
     if err is not None:
         return err
     collector: ClusterCollector = request.app.state.collector
     client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=2000)
 
+    # Per-connection private-project tenancy. The snapshot carries full
+    # rows (project_id + user_id); follow-up events are sparse (usually
+    # just ws_id), so invisible workstreams are recorded in ``hidden``
+    # at snapshot/ws_created time and every later event naming them is
+    # swallowed. Access changes (membership grant/revoke) take effect
+    # on reconnect — same resolve-once precedent as session
+    # construction. The visibility instance memoizes project rows, so
+    # the steady-state per-event cost is a set lookup.
+    visibility = WorkstreamProjectVisibility.for_request(request)
+    hidden: set[str] = set()
+
+    def _row_ws_id(ws: dict[str, Any]) -> str:
+        return str(ws.get("ws_id") or ws.get("id") or "")
+
+    def _filter_snapshot(snap: dict[str, Any]) -> dict[str, Any]:
+        for node in snap.get("nodes", []):
+            kept: list[dict[str, Any]] = []
+            for ws in node.get("workstreams", []):
+                if visibility.ws_visible(
+                    ws.get("project_id") or "", ws_owner=ws.get("user_id") or ""
+                ):
+                    kept.append(ws)
+                else:
+                    wid = _row_ws_id(ws)
+                    if wid:
+                        hidden.add(wid)
+            node["workstreams"] = kept
+        return snap
+
+    def _event_visible(event: dict[str, Any]) -> bool:
+        etype = event.get("type")
+        wid = str(event.get("ws_id") or "")
+        if etype == "ws_created":
+            if not visibility.ws_visible(
+                event.get("project_id") or "", ws_owner=event.get("user_id") or ""
+            ):
+                if wid:
+                    hidden.add(wid)
+                return False
+            hidden.discard(wid)
+            return True
+        if wid and wid in hidden:
+            if etype == "ws_closed":
+                hidden.discard(wid)
+            return False
+        return True
+
     async def event_generator() -> AsyncGenerator[dict[str, str], None]:
         loop = asyncio.get_running_loop()
         try:
@@ -1578,6 +1642,9 @@ async def cluster_events_sse(request: Request) -> Response:
                 None, collector.get_snapshot_and_register, client_queue
             )
             snap["type"] = "snapshot"
+            # Executor: the snapshot filter resolves project rows from
+            # storage — never block the event loop on DB I/O.
+            snap = await loop.run_in_executor(None, _filter_snapshot, snap)
             yield {"data": json.dumps(snap)}
 
             while True:
@@ -1585,7 +1652,15 @@ async def cluster_events_sse(request: Request) -> Response:
                     event = await loop.run_in_executor(
                         None, functools.partial(client_queue.get, timeout=5)
                     )
-                    yield {"data": json.dumps(event)}
+                    # Only ws_created can touch storage (project lookup);
+                    # every other event type is a pure set-membership
+                    # check and stays on the loop.
+                    if event.get("type") == "ws_created":
+                        visible = await loop.run_in_executor(None, _event_visible, event)
+                    else:
+                        visible = _event_visible(event)
+                    if visible:
+                        yield {"data": json.dumps(event)}
                 except queue.Empty:
                     pass  # poll timeout, retry
                 if await request.is_disconnected():
@@ -3393,6 +3468,19 @@ async def _coord_create_validate_request(
     """
     if not uid:
         return JSONResponse({"error": "authentication required"}, status_code=401)
+    # Project attach gate — same rule as the interactive validator: a
+    # private project accepts new workstreams only from its owner or
+    # members, and a nonexistent project_id 400s rather than minting a
+    # dangling link.
+    project_raw = body.get("project_id")
+    attach_pid = (project_raw.strip() if isinstance(project_raw, str) else "") or ""
+    if attach_pid:
+        from turnstone.core.auth import ensure_project_attachable
+
+        denied = ensure_project_attachable(uid, attach_pid)
+        if denied is not None:
+            status, message = denied
+            return JSONResponse({"error": message}, status_code=status)
     return None
 
 

turnstone/core/auth.py

@@ -253,6 +253,156 @@ def user_can_access_project(
     return acc.can_write if write else acc.can_read
 
 
+class WorkstreamProjectVisibility:
+    """Per-request memoized visibility predicate for project-scoped workstreams.
+
+    Answers "may *user_id* see a workstream attached to *project_id*?" for
+    listing filters and the row-access gate. Distinct from
+    :func:`user_can_access_project` on purpose: that composes the RBAC
+    capability (``project.read``, admin-default) with the ACL and gates the
+    project *management* surfaces, whereas workstream visibility is a
+    tenancy question — an explicit ``project_members`` row (or ownership)
+    IS the grant, no capability required, or members without ``project.read``
+    would lose sight of their own shared workstreams.
+
+    Rules (first match wins):
+
+    * ``bypass`` instances see everything — service-scope callers (the
+      collector and other cluster machinery must never be blinded at the
+      node edge; user-facing filtering happens at the console edge) and
+      holders of ``admin.cluster.inspect`` (the existing cluster-wide
+      workstream-inspect surface).
+    * No / dangling ``project_id`` → visible (a deleted project leaves the
+      link behind by design — no row, no privacy to enforce).
+    * Non-private visibility → visible (trusted-team default).
+    * Private → the workstream's own creator, the project owner, or a
+      project member; anonymous callers fail closed.
+    * A storage failure while resolving a project fails closed (treated
+      as private-and-not-a-member) rather than leaking on a blip.
+
+    Project rows and membership verdicts are memoized per instance —
+    construct one per request/connection, not per row.
+    """
+
+    def __init__(self, user_id: str, *, bypass: bool = False, storage: Any = None) -> None:
+        self._user_id = user_id or ""
+        self._bypass = bypass
+        self._storage = storage
+        self._projects: dict[str, dict[str, Any] | None] = {}
+        self._member: dict[str, bool] = {}
+
+    @classmethod
+    def for_request(cls, request: Any, *, storage: Any = None) -> WorkstreamProjectVisibility:
+        """Build a filter for an HTTP request's authenticated principal.
+
+        Service-scoped tokens and ``admin.cluster.inspect`` holders get a
+        bypass instance; everyone else filters as themselves.
+        """
+        auth: AuthResult | None = getattr(getattr(request, "state", None), "auth_result", None)
+        uid = str(getattr(auth, "user_id", "") or "")
+        bypass = bool(
+            auth is not None
+            and (auth.has_scope("service") or auth.has_permission("admin.cluster.inspect"))
+        )
+        return cls(uid, bypass=bypass, storage=storage)
+
+    def _resolve_storage(self) -> Any:
+        if self._storage is None:
+            from turnstone.core.storage._registry import get_storage
+
+            self._storage = get_storage()
+        return self._storage
+
+    def _project(self, project_id: str) -> dict[str, Any] | None:
+        if project_id not in self._projects:
+            storage = self._resolve_storage()
+            if storage is None:
+                raise RuntimeError("storage unavailable for project visibility check")
+            self._projects[project_id] = storage.get_project(project_id)
+        return self._projects[project_id]
+
+    def _is_member(self, project_id: str) -> bool:
+        if project_id not in self._member:
+            storage = self._resolve_storage()
+            self._member[project_id] = bool(
+                storage is not None and storage.is_project_member(project_id, self._user_id)
+            )
+        return self._member[project_id]
+
+    def ws_visible(self, project_id: str | None, ws_owner: str = "") -> bool:
+        """Apply the class rules to one workstream row."""
+        if self._bypass:
+            return True
+        pid = (project_id or "").strip()
+        if not pid:
+            return True
+        if ws_owner and ws_owner == self._user_id:
+            return True
+        try:
+            project = self._project(pid)
+            if project is None:
+                return True
+            if (project.get("visibility") or "private") != "private":
+                return True
+            if not self._user_id:
+                return False
+            if project.get("owner_id") == self._user_id:
+                return True
+            return self._is_member(pid)
+        except Exception:
+            log.warning(
+                "workstream project-visibility check failed user=%s project=%s — failing closed",
+                self._user_id,
+                pid,
+            )
+            return False
+
+
+def ensure_project_attachable(
+    user_id: str,
+    project_id: str,
+    *,
+    storage: Any = None,
+) -> tuple[int, str] | None:
+    """Gate attaching a NEW workstream to *project_id* at create time.
+
+    Returns ``None`` when the attach is allowed, else ``(status_code,
+    message)`` for the handler to surface. Stricter than
+    :meth:`WorkstreamProjectVisibility.ws_visible` in one way — a
+    nonexistent project is a 400 (a dangling link on an EXISTING row is
+    tolerated because project deletion leaves links behind, but minting
+    a fresh dangling link is a caller error) — and shares its tenancy
+    rule: private projects accept workstreams only from their owner or
+    members; public/active-or-archived projects accept from anyone
+    (memory writes stay member-gated at the session layer).
+
+    Fail-closed: empty ``user_id`` or a storage failure denies with 403.
+    """
+    pid = (project_id or "").strip()
+    if not pid:
+        return None
+    if storage is None:
+        from turnstone.core.storage._registry import get_storage
+
+        storage = get_storage()
+    if storage is None:
+        return (403, "project access could not be verified")
+    try:
+        project = storage.get_project(pid)
+        if project is None:
+            return (400, "unknown project_id")
+        if (project.get("visibility") or "private") != "private":
+            return None
+        if user_id and (
+            project.get("owner_id") == user_id or bool(storage.is_project_member(pid, user_id))
+        ):
+            return None
+        return (403, "cannot attach a workstream to a private project you don't belong to")
+    except Exception:
+        log.warning("project attach check failed user=%s project=%s — failing closed", user_id, pid)
+        return (403, "project access could not be verified")
+
+
 def _permissions_to_scopes(permissions: set[str]) -> frozenset[str]:
     """Derive legacy scopes from a granular permission set."""
     scopes: set[str] = set()

turnstone/core/memory.py

@@ -614,6 +614,20 @@ def get_workstream_owner(ws_id: str) -> str | None:
         return None
 
 
+def get_workstream_row(ws_id: str) -> dict[str, Any] | None:
+    """Return the full workstreams row dict, or None when missing/unreadable.
+
+    Same fail-soft shape as :func:`get_workstream_owner` — access gates
+    treat ``None`` as not-found, so a storage blip degrades to a 404
+    rather than a 500.
+    """
+    try:
+        return get_storage().get_workstream(ws_id)
+    except Exception:
+        log.warning("Failed to get workstream row ws=%s", ws_id, exc_info=True)
+        return None
+
+
 def update_workstream_title(ws_id: str, title: str) -> None:
     """Set or update the auto-generated title for a workstream."""
     try:

turnstone/core/session_routes.py

@@ -2646,12 +2646,20 @@ def make_list_handler(cfg: SessionEndpointConfig) -> Handler:
         resolve_titles = cfg.list_resolve_titles
 
         def _build_rows() -> list[dict[str, Any]]:
+            from turnstone.core.auth import WorkstreamProjectVisibility
+
+            visibility = WorkstreamProjectVisibility.for_request(request)
             wss = mgr.list_all()
             titles: dict[str, str | None] = {}
             if resolve_titles is not None and wss:
                 titles = resolve_titles([ws.id for ws in wss])
             rows: list[dict[str, Any]] = []
             for ws in wss:
+                project_id = getattr(ws, "project_id", "") or ""
+                # Private-project tenancy — drop rows the requester may
+                # not see (same predicate as the saved list).
+                if not visibility.ws_visible(project_id, ws_owner=ws.user_id or ""):
+                    continue
                 title = titles.get(ws.id) or ws.name
                 rows.append(
                     {
@@ -2661,6 +2669,7 @@ def make_list_handler(cfg: SessionEndpointConfig) -> Handler:
                         "kind": ws.kind,
                         "parent_ws_id": ws.parent_ws_id,
                         "user_id": ws.user_id,
+                        "project_id": project_id or None,
                     }
                 )
             return rows
@@ -2720,12 +2729,17 @@ async def _collect_saved_rows(
     # Column order from list_workstreams_with_history (keep in sync with
     # the storage SELECT): ws_id, alias, title, name, created, updated,
     # message_count, node_id, state, kind, model_alias, launch_skill,
-    # child_count, context_tokens, context_window.  The occupancy ratio
-    # is derived here (Python float division) rather than in SQL so the
-    # NULL / zero-window cases stay obvious and identical across backends.
-    # context_window is NULL for model aliases absent from
-    # model_definitions (e.g. config.toml-only models), so context_ratio
-    # degrades to 0.0 there rather than reporting a bogus occupancy.
+    # child_count, context_tokens, context_window, project_id, owner.
+    # The occupancy ratio is derived here (Python float division) rather
+    # than in SQL so the NULL / zero-window cases stay obvious and
+    # identical across backends.  context_window is NULL for model
+    # aliases absent from model_definitions (e.g. config.toml-only
+    # models), so context_ratio degrades to 0.0 there rather than
+    # reporting a bogus occupancy.
+    from turnstone.core.auth import WorkstreamProjectVisibility
+
+    visibility = WorkstreamProjectVisibility.for_request(request)
+
     result: list[dict[str, Any]] = []
     for row in rows:
         (
@@ -2744,9 +2758,15 @@ async def _collect_saved_rows(
             child_count,
             context_tokens,
             context_window,
+            project_id,
+            owner,
         ) = row
         if wid in loaded:
             continue
+        # Private-project tenancy: rows the requester may not see are
+        # dropped server-side, not hidden client-side.
+        if not visibility.ws_visible(project_id, ws_owner=owner or ""):
+            continue
         ctx_tokens = context_tokens or 0
         context_ratio = (
             round(ctx_tokens / context_window, 3) if ctx_tokens and context_window else 0.0
@@ -2768,6 +2788,7 @@ async def _collect_saved_rows(
                 "child_count": child_count or 0,
                 "context_tokens": ctx_tokens,
                 "context_ratio": context_ratio,
+                "project_id": project_id or None,
             }
         )
     return result

turnstone/core/storage/_postgresql.py

@@ -622,7 +622,7 @@ class PostgreSQLBackend:
                         "(SELECT ue.prompt_tokens FROM usage_events ue "
                         " WHERE ue.ws_id = w.ws_id "
                         " ORDER BY ue.timestamp DESC LIMIT 1), "
-                        "md.context_window "
+                        "md.context_window, w.project_id, w.user_id "
                         "FROM workstreams w "
                         "LEFT JOIN workstream_config wcm "
                         "  ON wcm.ws_id = w.ws_id AND wcm.key = 'model_alias' "

turnstone/core/storage/_sqlite.py

@@ -726,7 +726,7 @@ class SQLiteBackend:
                         "(SELECT ue.prompt_tokens FROM usage_events ue "
                         " WHERE ue.ws_id = w.ws_id "
                         " ORDER BY ue.timestamp DESC LIMIT 1), "
-                        "md.context_window "
+                        "md.context_window, w.project_id, w.user_id "
                         "FROM workstreams w "
                         "LEFT JOIN workstream_config wcm "
                         "  ON wcm.ws_id = w.ws_id AND wcm.key = 'model_alias' "

turnstone/core/web_helpers.py

@@ -316,17 +316,22 @@ def resolve_workstream_owner(
     """Resolve ``ws_id`` to its owner; 404 when the row doesn't exist.
 
     Turnstone is a trusted-team tool: scope-level auth (e.g.
-    ``admin.workstreams`` / ``admin.coordinator``) is the only gate;
-    row-level ownership is not enforced here. Returns
+    ``admin.coordinator``) is the primary gate and row-level OWNERSHIP
+    is still not enforced here. The one row-level check this performs
+    is PROJECT tenancy: a workstream attached to a *private* project is
+    only reachable by the project's owner/members, the workstream's own
+    creator, service-scope callers, and ``admin.cluster.inspect``
+    holders — everyone else gets a 403 (see
+    :class:`turnstone.core.auth.WorkstreamProjectVisibility`). Returns
     ``(owner_user_id, None)`` on success — the persisted owner id,
     which attachments should be filed under so existing storage shape
     is preserved. Falls back to the caller's own uid when the row has
     no recorded owner.
 
     When ``mgr`` is provided and the workstream is live in memory,
-    trust its cached ``user_id`` instead of round-tripping storage —
-    keeps in-memory-only handlers functional during transient DB
-    outages and trims the hot-path by one query.
+    trust its cached ``user_id`` / ``project_id`` instead of
+    round-tripping storage — keeps in-memory-only handlers functional
+    during transient DB outages and trims the hot-path by one query.
 
     ``not_found_label`` is the message body the 404 carries — the
     interactive surface uses "Workstream not found"; coord uses
@@ -334,20 +339,36 @@ def resolve_workstream_owner(
     """
     from starlette.responses import JSONResponse as _JSONResponse
 
+    from turnstone.core.auth import WorkstreamProjectVisibility
+
     caller = auth_user_id(request)
+    owner: str | None = None
+    project_id = ""
 
     if mgr is not None:
         ws_mem = mgr.get(ws_id)
         if ws_mem is not None:
-            return ws_mem.user_id or caller, None
-        # Not in memory — fall through to storage so persisted-but-not-
-        # loaded rows still resolve.
+            owner = ws_mem.user_id or ""
+            project_id = getattr(ws_mem, "project_id", "") or ""
 
-    from turnstone.core.memory import get_workstream_owner
-
-    owner = get_workstream_owner(ws_id)
     if owner is None:
-        return "", _JSONResponse({"error": not_found_label}, status_code=404)
+        # Not in memory — storage resolves persisted-but-not-loaded rows.
+        from turnstone.core.memory import get_workstream_row
+
+        row = get_workstream_row(ws_id)
+        if row is None:
+            return "", _JSONResponse({"error": not_found_label}, status_code=404)
+        owner = row.get("user_id") or ""
+        project_id = row.get("project_id") or ""
+
+    if project_id:
+        visibility = WorkstreamProjectVisibility.for_request(request)
+        if not visibility.ws_visible(project_id, ws_owner=owner):
+            return "", _JSONResponse(
+                {"error": "Forbidden: workstream belongs to a private project"},
+                status_code=403,
+            )
+
     return owner or caller, None
 
 

turnstone/server.py

@@ -1065,12 +1065,19 @@ async def global_events_sse(request: Request) -> Response:
 
 async def dashboard(request: Request) -> JSONResponse:
     """GET /v1/api/dashboard — enriched workstream data + aggregate stats."""
+    from turnstone.core.auth import WorkstreamProjectVisibility
     from turnstone.core.memory import get_workstream_display_name
 
     mgr: SessionManager = request.app.state.workstreams
-    # No per-user filter — see list_workstreams above for the rationale
-    # (trusted-team deployment shape; mutations stay owner-gated).
-    wss = mgr.list_all()
+    # No per-user OWNER filter (trusted-team deployment shape) — but
+    # private-project rows are dropped for non-members, same predicate
+    # as every other listing surface.
+    visibility = WorkstreamProjectVisibility.for_request(request)
+    wss = [
+        ws
+        for ws in mgr.list_all()
+        if visibility.ws_visible(getattr(ws, "project_id", "") or "", ws_owner=ws.user_id or "")
+    ]
     total_tokens = 0
     total_tool_calls = 0
     active_count = 0
@@ -1936,6 +1943,20 @@ async def _interactive_create_validate_request(
         # (which forwards the body verbatim).
         if not (body.get("project_id") or "") and parent_row.get("project_id"):
             body["project_id"] = parent_row.get("project_id")
+    # Project attach gate (explicit or parent-inherited): a private
+    # project accepts new workstreams only from its owner/members, and a
+    # nonexistent project_id is a caller error rather than a silent
+    # dangling link. Re-checking the inherited value is deliberate — a
+    # coordinator owner whose membership was revoked fails the child
+    # spawn loudly here instead of minting rows they can no longer see.
+    attach_pid = str(body.get("project_id") or "")
+    if attach_pid:
+        from turnstone.core.auth import ensure_project_attachable
+
+        denied = ensure_project_attachable(uid, attach_pid)
+        if denied is not None:
+            status, message = denied
+            return JSONResponse({"error": message}, status_code=status)
     notify_targets_raw = body.get("notify_targets", "[]")
     if isinstance(notify_targets_raw, list):
         notify_targets_raw = json.dumps(notify_targets_raw)