fix: scope private-project workstream visibility to members, not admins

Workstreams attached to a private project were visible -- including their
conversation content -- to holders of admin.cluster.inspect / admin.coordinator
(both default builtin-admin permissions), defeating the project's confidentiality
boundary. Enforce that a private project's resources are visible only to people
IN the project (owner, workstream creator, or an explicit member), even for admins.

Surfaces closed:

- WorkstreamProjectVisibility bypass narrowed to service scope only (node->console
  machine plumbing, re-filtered per-user at the console edge). No human principal
  bypasses; admin.cluster.inspect gates the inspect surfaces, not tenancy. This
  flows to /dashboard, session listings, the attachment row-gate, cluster_workstreams,
  cluster_node_detail, and cluster_snapshot/SSE.
- cluster_ws_detail 404-masks a workstream in a private project the caller can't
  see; cluster_ws_live_bulk routes such ids to the denied list (no private-project
  oracle).
- Coordinator operator verbs (history/export/detail/send/approve/set_title/open/
  children/tasks/attachments) now enforce project tenancy: _coordinator_tenant_check
  on coord_endpoint_config, the gate in _resolve_coordinator_or_404 (children/tasks),
  the tenant_check now run in make_open_handler before rehydrate, and a
  project-visibility check in _coord_attachment_owner. admin.coordinator gates the
  surface cluster-wide, but a non-member is 404-masked. The tenant-check mirrors the
  manager-first + coordinator-kind ladder so kind-isolation is preserved.
- service scope is no longer user-assignable: admin_create_token and both
  turnstone-admin CLI mint paths reject it via reject_unassignable_scopes, so an
  admin.users holder cannot self-mint a service token and restore the bypass. Service
  scope is minted only by ServiceTokenManager / the JWT secret.
- The events/global node proxy (service-elevated cross-tenant firehose) is gated on
  admin.cluster.inspect so a plain authenticated user cannot reach it through the
  console proxy.

Updates the OpenAPI description, the row-gate/tenancy-filter docstrings, and adds
tests for every surface (visibility predicate + cluster detail/bulk + coordinator
history/export/children/open/attachments + events/global proxy + scope-mint
rejection); inverts the tests that pinned the old admin-bypass contract.

tests/test_console.py

@@ -1600,6 +1600,82 @@ class TestConsoleProxy:
             # browser's interactive UI 403-loops on every retry.
             assert sse_mock.await_args.kwargs.get("use_service_auth") is True
 
+    def test_proxy_events_global_403_without_cluster_inspect(self, mock_collector):
+        """A plain authenticated user (no service scope, no
+        admin.cluster.inspect) cannot reach the node's cross-tenant
+        firehose through the proxy: elevating to the console's service
+        identity would bypass per-user filtering, so the path is
+        operator-gated. _proxy_sse must NOT be reached."""
+        from unittest.mock import AsyncMock, patch
+
+        from starlette.responses import Response
+        from starlette.testclient import TestClient
+
+        from turnstone.console.server import _load_static, create_app
+        from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt
+
+        _load_static()
+        app = create_app(collector=mock_collector, jwt_secret=_TEST_JWT_SECRET)
+        user_jwt = create_jwt(
+            user_id="plain-user",
+            scopes=frozenset({"read"}),
+            source="test",
+            secret=_TEST_JWT_SECRET,
+            audience=JWT_AUD_CONSOLE,
+            permissions=frozenset(),
+        )
+        user_client = TestClient(
+            app,
+            raise_server_exceptions=False,
+            headers={"Authorization": f"Bearer {user_jwt}"},
+        )
+        with patch(
+            "turnstone.console.server._proxy_sse",
+            new_callable=AsyncMock,
+            return_value=Response("ok", status_code=200),
+        ) as sse_mock:
+            resp = user_client.get("/node/node-a/v1/api/events/global")
+        assert resp.status_code == 403
+        assert sse_mock.await_count == 0
+        user_client.close()
+
+    def test_proxy_events_global_allows_cluster_inspect(self, mock_collector):
+        """An operator holding admin.cluster.inspect passes the gate and
+        reaches the SSE proxy with the service token."""
+        from unittest.mock import AsyncMock, patch
+
+        from starlette.responses import Response
+        from starlette.testclient import TestClient
+
+        from turnstone.console.server import _load_static, create_app
+        from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt
+
+        _load_static()
+        app = create_app(collector=mock_collector, jwt_secret=_TEST_JWT_SECRET)
+        op_jwt = create_jwt(
+            user_id="operator",
+            scopes=frozenset({"read"}),
+            source="test",
+            secret=_TEST_JWT_SECRET,
+            audience=JWT_AUD_CONSOLE,
+            permissions=frozenset({"admin.cluster.inspect"}),
+        )
+        op_client = TestClient(
+            app,
+            raise_server_exceptions=False,
+            headers={"Authorization": f"Bearer {op_jwt}"},
+        )
+        with patch(
+            "turnstone.console.server._proxy_sse",
+            new_callable=AsyncMock,
+            return_value=Response("ok", status_code=200),
+        ) as sse_mock:
+            resp = op_client.get("/node/node-a/v1/api/events/global")
+        assert resp.status_code == 200
+        assert sse_mock.await_count == 1
+        assert sse_mock.await_args.kwargs.get("use_service_auth") is True
+        op_client.close()
+
     def test_proxy_api_per_ws_events_uses_user_auth_not_service(self, client, mock_collector):
         """Per-ws events route uses the user's re-minted JWT, not the
         service token — the upstream per-ws SSE handler scopes by

tests/test_coordinator_endpoints.py

@@ -42,6 +42,7 @@ from turnstone.console.server import (
     _coord_create_post_install,
     _coord_create_validate_request,
     _coord_saved_loaded_lookup,
+    _coordinator_tenant_check,
     _require_admin_coordinator,
     _require_coord_mgr,
     cluster_ws_detail,
@@ -83,15 +84,21 @@ def _coord_attach_owner(request, ws_id, mgr):
 
     Kind-strict — coord attachments can only be accessed for
     workstreams currently held by ``coord_mgr``; no storage fallback
-    so cross-kind ws_ids 404 instead of leaking through storage.
+    so cross-kind ws_ids 404 instead of leaking through storage. Also
+    project-tenancy-strict: mirrors ``_coord_attachment_owner`` so a
+    private-project coordinator's attachments 404-mask non-members.
     """
     from starlette.responses import JSONResponse
 
+    from turnstone.core.auth import WorkstreamProjectVisibility
     from turnstone.core.web_helpers import auth_user_id
 
     ws = mgr.get(ws_id)
     if ws is None:
         return "", JSONResponse({"error": "coordinator not found"}, status_code=404)
+    visibility = WorkstreamProjectVisibility.for_request(request)
+    if not visibility.ws_visible(getattr(ws, "project_id", "") or "", ws_owner=ws.user_id or ""):
+        return "", JSONResponse({"error": "coordinator not found"}, status_code=404)
     return ws.user_id or auth_user_id(request), None
 
 
@@ -101,7 +108,7 @@ def _coord_attach_owner(request, ws_id, mgr):
 _coord_endpoint_config = SessionEndpointConfig(
     permission_gate=_require_admin_coordinator,
     manager_lookup=_require_coord_mgr,
-    tenant_check=None,
+    tenant_check=_coordinator_tenant_check,
     not_found_label="coordinator not found",
     audit_action_prefix="coordinator",
     supports_attachments=True,
@@ -1408,6 +1415,110 @@ def test_history_any_admin_coordinator_caller_can_read(storage):
     assert resp.json()["ws_id"] == ws.id
 
 
+def test_history_private_project_hidden_from_non_member(storage):
+    # admin.coordinator gates the surface, but a coordinator in a private
+    # project the caller isn't a member of is 404-masked — the conversation
+    # does not leak to a non-member operator.
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.register_workstream(
+        "c" * 32, kind="coordinator", user_id="alice", project_id="proj-secret"
+    )
+    storage.save_message("c" * 32, "user", "secret plan")
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/workstreams/{'c' * 32}/history",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 404
+
+
+def test_history_private_project_visible_to_member(storage):
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.add_project_member("proj-secret", "member-bob")
+    storage.register_workstream(
+        "c" * 32, kind="coordinator", user_id="alice", project_id="proj-secret"
+    )
+    storage.save_message("c" * 32, "user", "secret plan")
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/workstreams/{'c' * 32}/history",
+        headers={"X-Test-User": "member-bob", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 200
+    assert any(m.get("content") == "secret plan" for m in resp.json()["messages"])
+
+
+def test_export_private_project_hidden_from_non_member(storage):
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.register_workstream(
+        "c" * 32, kind="coordinator", user_id="alice", project_id="proj-secret"
+    )
+    storage.save_message("c" * 32, "user", "secret plan")
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/workstreams/{'c' * 32}/export",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 404
+
+
+def test_children_private_project_hidden_from_non_member(storage):
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.register_workstream(
+        "c" * 32, kind="coordinator", user_id="alice", project_id="proj-secret"
+    )
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/workstreams/{'c' * 32}/children",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 404
+
+
+def test_open_private_project_hidden_from_non_member(storage):
+    # `open` rehydrates + returns the auto-titled name, so an ungated open is a
+    # private-project existence/metadata oracle AND an unauthorized resurrection.
+    # The tenant_check must fire before the already-loaded shortcut and mgr.open.
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.register_workstream(
+        "c" * 32, kind="coordinator", user_id="alice", project_id="proj-secret"
+    )
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/workstreams/{'c' * 32}/open",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 404
+
+
+def test_coord_attachments_private_project_hidden_from_non_member(storage):
+    # Attachment list/serve resolves the owner as the coord owner and only
+    # enforced cross-kind before — a non-member operator could enumerate and
+    # download the owner's staged blobs. Now 404-masked by project tenancy.
+    storage.create_project("proj-secret", "Secret", "alice")
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="alice", project_id="proj-secret")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/workstreams/{ws.id}/attachments",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 404
+
+
+def test_coord_attachments_private_project_visible_to_member(storage):
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.add_project_member("proj-secret", "member-bob")
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="alice", project_id="proj-secret")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/workstreams/{ws.id}/attachments",
+        headers={"X-Test-User": "member-bob", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 200
+
+
 def test_history_serves_storage_only_workstream(storage):
     """Persisted-but-not-loaded coordinators (closed / evicted) are still
     readable via /history without rehydrating. Mirrors the pre-lift
@@ -2108,6 +2219,10 @@ def test_open_any_admin_coordinator_caller_succeeds_in_memory(storage):
 
 def test_open_rehydrates_when_not_in_memory(storage, monkeypatch):
     mgr = _build_mgr(storage)
+    # The tenancy gate resolves the row from storage before rehydrating, so a
+    # legitimately-openable coordinator must exist there (it always does in
+    # production — open rehydrates a persisted row).
+    storage.register_workstream("coord-rehy", kind="coordinator", user_id="user-1")
     rehydrated = MagicMock()
     rehydrated.id = "coord-rehy"
     rehydrated.name = "rehydrated"
@@ -2141,6 +2256,7 @@ def test_open_503_on_coord_mgr_unavailable(storage):
 
 def test_open_correlation_id_on_factory_failure(storage, monkeypatch):
     mgr = _build_mgr(storage)
+    storage.register_workstream("bad-ws", kind="coordinator", user_id="user-1")
     monkeypatch.setattr(mgr, "open", MagicMock(side_effect=RuntimeError("boom")))
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
     resp = client.post("/v1/api/workstreams/bad-ws/open", headers=_COORD_HEADERS)
@@ -2151,6 +2267,7 @@ def test_open_correlation_id_on_factory_failure(storage, monkeypatch):
 def test_open_503_when_open_raises_value_error(storage, monkeypatch):
     """ValueError from the factory surfaces as 503 with the remediation text."""
     mgr = _build_mgr(storage)
+    storage.register_workstream("bad-ws", kind="coordinator", user_id="user-1")
     monkeypatch.setattr(mgr, "open", MagicMock(side_effect=ValueError("coord registry missing")))
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
     resp = client.post("/v1/api/workstreams/bad-ws/open", headers=_COORD_HEADERS)
@@ -2316,7 +2433,8 @@ def test_cluster_inspect_invalid_ws_id_400(storage):
 
 
 def test_cluster_inspect_any_inspect_caller_sees_detail(storage):
-    # Trusted-team visibility: admin.cluster.inspect sees every row.
+    # A project-less workstream has no tenancy to enforce, so any
+    # admin.cluster.inspect caller sees it (trusted-team default).
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -2328,6 +2446,46 @@ def test_cluster_inspect_any_inspect_caller_sees_detail(storage):
     assert resp.json()["persisted"]["ws_id"] == ws.id
 
 
+def test_cluster_inspect_private_project_hidden_from_non_member(storage):
+    # admin.cluster.inspect gates the surface, but a workstream in a
+    # private project the caller isn't a member of is masked as 404 —
+    # no private-project oracle even for a cluster admin.
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.register_workstream(
+        "c" * 32,
+        node_id="console",
+        user_id="alice",
+        kind="coordinator",
+        project_id="proj-secret",
+    )
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/cluster/ws/{'c' * 32}/detail",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.cluster.inspect"},
+    )
+    assert resp.status_code == 404
+
+
+def test_cluster_inspect_private_project_visible_to_member(storage):
+    # A project member (even a non-owner) still sees the persisted row.
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.add_project_member("proj-secret", "member-bob")
+    storage.register_workstream(
+        "c" * 32,
+        node_id="console",
+        user_id="alice",
+        kind="coordinator",
+        project_id="proj-secret",
+    )
+    client = _make_client(storage, coord_mgr=_build_mgr(storage), registry=_fake_registry())
+    resp = client.get(
+        f"/v1/api/cluster/ws/{'c' * 32}/detail",
+        headers={"X-Test-User": "member-bob", "X-Test-Perms": "admin.cluster.inspect"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["persisted"]["ws_id"] == "c" * 32
+
+
 def test_cluster_inspect_coordinator_self_path(storage):
     """A coordinator row returns live from the in-process manager."""
     mgr = _build_mgr(storage)

tests/test_phase6_endpoints.py

@@ -179,10 +179,10 @@ def test_bulk_live_admin_bypass_returns_live(storage):
 
 
 def test_bulk_live_cluster_wide_visibility(storage):
-    """Trusted-team visibility: any ``admin.cluster.inspect`` caller
-    sees every row in ``results``.  ``denied`` is reserved for ids
-    that don't correspond to a persisted workstream (no existence
-    oracle for unknown ids)."""
+    """A project-less workstream has no tenancy to enforce, so any
+    ``admin.cluster.inspect`` caller sees it in ``results``.  ``denied``
+    is reserved for ids that don't correspond to a persisted workstream
+    (no existence oracle for unknown ids)."""
     ws_id = "b" * 32
     _seed_workstream(storage, ws_id=ws_id, node_id="node-a", user_id="stranger")
     client = _make_client(storage, coord_mgr=_build_mgr(storage))
@@ -196,6 +196,48 @@ def test_bulk_live_cluster_wide_visibility(storage):
     assert body["denied"] == []
 
 
+def test_bulk_live_private_project_row_routes_to_denied(storage):
+    """A workstream in a private project the caller isn't a member of
+    routes to ``denied``, not ``results`` — a cluster admin gets no
+    private-project oracle from the bulk surface either."""
+    storage.create_project("proj-secret", "Secret", "alice")
+    ws_id = "c" * 32
+    storage.register_workstream(ws_id, node_id="node-a", user_id="alice", project_id="proj-secret")
+    client = _make_client(storage, coord_mgr=_build_mgr(storage))
+    resp = client.get(
+        f"/v1/api/cluster/ws/live?ids={ws_id}",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.cluster.inspect"},
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["results"] == {}
+    assert body["denied"] == [ws_id]
+
+
+def test_bulk_live_private_project_row_visible_to_member(storage):
+    """A project member sees the row (routes to ``results``); the live
+    block is null only because the coordinator row isn't loaded."""
+    storage.create_project("proj-secret", "Secret", "alice")
+    storage.add_project_member("proj-secret", "member-bob")
+    ws_id = "c" * 32
+    storage.register_workstream(
+        ws_id,
+        node_id="console",
+        user_id="alice",
+        kind="coordinator",
+        project_id="proj-secret",
+    )
+    client = _make_client(storage, coord_mgr=_build_mgr(storage))
+    resp = client.get(
+        f"/v1/api/cluster/ws/live?ids={ws_id}",
+        headers={"X-Test-User": "member-bob", "X-Test-Perms": "admin.cluster.inspect"},
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert ws_id in body["results"]
+    assert body["denied"] == []
+
+
 def test_bulk_live_unknown_ids_route_to_denied(storage):
     """Unknown ids (not in storage) land in ``denied`` so the endpoint
     can't be used as an existence oracle."""

tests/test_project_workstream_visibility.py

@@ -127,10 +127,14 @@ class TestWsVisiblePredicate:
         assert storage.get_project.call_count == 1
 
     def test_for_request_bypass_rules(self) -> None:
+        # Only service scope bypasses (node→console machine plumbing,
+        # re-filtered per-user at the console edge).
         assert WorkstreamProjectVisibility.for_request(
             _request_for("bob", scopes=("service",))
         )._bypass
-        assert WorkstreamProjectVisibility.for_request(
+        # admin.cluster.inspect gates the inspect *surfaces* but does NOT
+        # bypass private-project tenancy — the admin filters as themselves.
+        assert not WorkstreamProjectVisibility.for_request(
             _request_for("bob", permissions=("admin.cluster.inspect",))
         )._bypass
         assert not WorkstreamProjectVisibility.for_request(_request_for("bob"))._bypass
@@ -214,15 +218,17 @@ class TestResolveWorkstreamOwnerProjectGate:
         assert err is None
         assert owner == "bob"
 
-    def test_admin_inspect_bypasses(self, tmp_db: str) -> None:
+    def test_admin_inspect_does_not_bypass(self, tmp_db: str) -> None:
+        # A permitted admin (admin.cluster.inspect) who isn't the owner /
+        # creator / member of a private project is still 403'd at the row
+        # gate — the permission gates the inspect surface, not the tenancy.
         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"
+        assert err is not None and err.status_code == 403
 
     def test_missing_ws_still_404s(self, tmp_db: str) -> None:
         from turnstone.core.web_helpers import resolve_workstream_owner

tests/test_server_authz.py

@@ -57,6 +57,39 @@ def _auth(
     return {"Authorization": f"Bearer {_make_jwt(user, scopes=scopes, permissions=permissions)}"}
 
 
+class TestAssignableScopes:
+    """``service`` scope is a cross-tenant bypass and must never be
+    GRANTED via a user-facing token mint (admin API or CLI) — otherwise an
+    ``admin.users`` holder could self-mint it and see every private
+    project's workstreams.  Both mint paths route through
+    :func:`reject_unassignable_scopes`."""
+
+    def test_service_scope_rejected(self) -> None:
+        from turnstone.core.auth import reject_unassignable_scopes
+
+        assert reject_unassignable_scopes("service") is not None
+        assert reject_unassignable_scopes("read,service") is not None
+        assert reject_unassignable_scopes("read,write,approve,service") is not None
+
+    def test_service_not_in_assignable_set(self) -> None:
+        from turnstone.core.auth import ASSIGNABLE_SCOPES, VALID_SCOPES
+
+        assert "service" in VALID_SCOPES  # still a valid runtime scope
+        assert "service" not in ASSIGNABLE_SCOPES  # but not user-assignable
+
+    def test_ordinary_scopes_accepted(self) -> None:
+        from turnstone.core.auth import reject_unassignable_scopes
+
+        assert reject_unassignable_scopes("read") is None
+        assert reject_unassignable_scopes("read,write,approve") is None
+
+    def test_empty_and_unknown_rejected(self) -> None:
+        from turnstone.core.auth import reject_unassignable_scopes
+
+        assert reject_unassignable_scopes("") is not None
+        assert reject_unassignable_scopes("bogus") is not None
+
+
 # ---------------------------------------------------------------------------
 # FakeUI / FakeSession doubles — match the shape the create handler expects
 # ---------------------------------------------------------------------------

turnstone/admin.py

@@ -78,7 +78,13 @@ def _cmd_create_user(args: argparse.Namespace) -> None:
     print(f"  Name: {args.name}")
 
     if args.token:
+        from turnstone.core.auth import reject_unassignable_scopes
+
         scopes = args.scopes or "read,write,approve"
+        scope_err = reject_unassignable_scopes(scopes)
+        if scope_err is not None:
+            print(f"Error: {scope_err}", file=sys.stderr)
+            sys.exit(1)
         raw = generate_token()
         tid = uuid.uuid4().hex
         storage.create_api_token(
@@ -96,7 +102,12 @@ def _cmd_create_user(args: argparse.Namespace) -> None:
 
 
 def _cmd_create_token(args: argparse.Namespace) -> None:
-    from turnstone.core.auth import generate_token, hash_token, token_prefix
+    from turnstone.core.auth import (
+        generate_token,
+        hash_token,
+        reject_unassignable_scopes,
+        token_prefix,
+    )
 
     storage = _get_storage(args)
 
@@ -104,6 +115,12 @@ def _cmd_create_token(args: argparse.Namespace) -> None:
         print(f"Error: user {args.user} not found", file=sys.stderr)
         sys.exit(1)
 
+    scopes = args.scopes or "read,write,approve"
+    scope_err = reject_unassignable_scopes(scopes)
+    if scope_err is not None:
+        print(f"Error: {scope_err}", file=sys.stderr)
+        sys.exit(1)
+
     expires = None
     if args.expires_days:
         from datetime import UTC, datetime, timedelta
@@ -120,12 +137,12 @@ def _cmd_create_token(args: argparse.Namespace) -> None:
         token_prefix=token_prefix(raw),
         user_id=args.user,
         name=args.name or "",
-        scopes=args.scopes,
+        scopes=scopes,
         expires=expires,
     )
     print(f"Token: {raw}")
     print(f"  ID: {tid}")
-    print(f"  Scopes: {args.scopes}")
+    print(f"  Scopes: {scopes}")
     if expires:
         print(f"  Expires: {expires}")
     print("  (Save this token now — it cannot be retrieved again)")

turnstone/api/console_spec.py

@@ -1572,9 +1572,12 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
             '``kind="coordinator"`` rows), and the tail of the message '
             "history.  Gated on the ``admin.cluster.inspect`` permission "
             "(granted to ``builtin-admin`` via migration 040; revoke or "
-            "reassign to a custom role for tighter control).  ``live`` "
-            "is null on node unreachability / 5xx so callers can degrade "
-            "gracefully."
+            "reassign to a custom role for tighter control).  A workstream "
+            "attached to a *private* project stays confidential to its "
+            "members: a permitted caller who isn't its owner / creator / "
+            "project member gets a 404 (same masking as an unknown id).  "
+            "``live`` is null on node unreachability / 5xx so callers can "
+            "degrade gracefully."
         ),
         response_model=ClusterWsDetailResponse,
         query_params=[

turnstone/console/server.py

@@ -1323,7 +1323,7 @@ async def cluster_ws_detail(request: Request) -> JSONResponse:
     404 masks ownership failures (match :func:`make_detail_handler`).
     Correlation-id masks unexpected exceptions in the merge path.
     """
-    from turnstone.core.auth import require_permission
+    from turnstone.core.auth import WorkstreamProjectVisibility, require_permission
     from turnstone.core.web_helpers import require_storage_or_503
 
     err = require_permission(request, "admin.cluster.inspect")
@@ -1337,6 +1337,12 @@ async def cluster_ws_detail(request: Request) -> JSONResponse:
     if not _VALID_WS_ID_RE.match(ws_id):
         return JSONResponse({"error": "invalid ws_id"}, status_code=400)
 
+    # ``admin.cluster.inspect`` gates the surface, but a workstream attached
+    # to a private project stays confidential to its members — a permitted
+    # admin who isn't the owner/creator/member sees a 404, same existence
+    # masking as an unknown ws_id below (no private-project oracle).
+    visibility = WorkstreamProjectVisibility.for_request(request, storage=storage)
+
     # Accept either ``?limit=`` (the canonical name used by
     # the lifted history factory and the list_workstreams tool) or the
     # transitional ``?message_limit=`` from earlier phase-3 drafts.
@@ -1378,6 +1384,14 @@ async def cluster_ws_detail(request: Request) -> JSONResponse:
     if row is None:
         return JSONResponse({"error": "workstream not found"}, status_code=404)
 
+    # Private-project tenancy — the memoized predicate may resolve a project
+    # row + membership from storage, so judge it off the event loop.
+    ws_visible = await asyncio.to_thread(
+        visibility.ws_visible, row.get("project_id") or "", row.get("user_id") or ""
+    )
+    if not ws_visible:
+        return JSONResponse({"error": "workstream not found"}, status_code=404)
+
     try:
         live = await _fetch_live_block(request, row, ws_id)
     except Exception:
@@ -1430,14 +1444,15 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
     ``cluster_ws_detail`` so node-dashboard cache behaviour, coordinator
     in-process snapshots, and ownership masking stay consistent.
 
-    Permission + ownership semantics match ``cluster_ws_detail``:
-    gated on ``admin.cluster.inspect`` and rows the caller doesn't
-    own surface in ``denied`` rather than ``results`` (so the endpoint
-    can't be used as an existence oracle).  Missing ids also route to
+    Permission + tenancy semantics match ``cluster_ws_detail``:
+    gated on ``admin.cluster.inspect``, and rows the caller can't see —
+    private-project workstreams they don't own / aren't a member of —
+    surface in ``denied`` rather than ``results`` (so the endpoint can't
+    be used as a private-project oracle).  Missing ids also route to
     ``denied`` for the same reason.  ``ids`` over the cap is truncated
     with ``truncated=true`` so the model / frontend knows to paginate.
     """
-    from turnstone.core.auth import require_permission
+    from turnstone.core.auth import WorkstreamProjectVisibility, require_permission
     from turnstone.core.web_helpers import require_storage_or_503
 
     err = require_permission(request, "admin.cluster.inspect")
@@ -1446,6 +1461,7 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
     storage, err503 = require_storage_or_503(request)
     if err503 is not None:
         return err503
+    visibility = WorkstreamProjectVisibility.for_request(request, storage=storage)
 
     raw_ids = request.query_params.get("ids", "") or ""
     # Split on comma; strip whitespace; drop empty / invalid entries.
@@ -1484,17 +1500,27 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
         )
 
     results: dict[str, dict[str, Any] | None] = {}
-    denied: list[str] = []
-    owned_rows: list[tuple[str, dict[str, Any]]] = []
-    for wid in cleaned:
-        row = rows.get(wid)
-        if row is None:
-            # Missing rows route to ``denied`` rather than ``results``
-            # so the endpoint can't be used as an existence oracle for
-            # ids outside the caller's knowledge.
-            denied.append(wid)
-            continue
-        owned_rows.append((wid, row))
+
+    def _partition() -> tuple[list[tuple[str, dict[str, Any]]], list[str]]:
+        # Missing rows AND private-project rows the caller isn't a member of
+        # both route to ``denied`` rather than ``results`` — neither an
+        # existence oracle for unknown ids nor a private-project oracle for
+        # workstreams the admin can't see. The tenancy predicate resolves
+        # project rows + membership from storage, so this runs off the
+        # event loop.
+        visible: list[tuple[str, dict[str, Any]]] = []
+        hidden: list[str] = []
+        for wid in cleaned:
+            row = rows.get(wid)
+            if row is None or not visibility.ws_visible(
+                row.get("project_id") or "", row.get("user_id") or ""
+            ):
+                hidden.append(wid)
+                continue
+            visible.append((wid, row))
+        return visible, hidden
+
+    owned_rows, denied = await asyncio.to_thread(_partition)
 
     # Fetch live blocks concurrently — ``_fetch_live_block`` already
     # routes node-backed reads through the per-node dashboard cache,
@@ -1604,10 +1630,11 @@ class _ClusterTenancyFilter:
 
     def __init__(self, visibility: Any) -> None:
         self._vis = visibility
-        # Bypass principals (service scope / admin.cluster.inspect) get
-        # the payload UNTOUCHED — no row drops, and crucially no
-        # overview recompute (their header should reflect the
-        # collector's own aggregates).
+        # Bypass principals (service scope only — the collector/machine
+        # plumbing) get the payload UNTOUCHED — no row drops, and crucially
+        # no overview recompute (their header should reflect the
+        # collector's own aggregates). Human admins are NOT bypass; they
+        # see the same private-project filtering as any other user.
         self._bypass = bool(getattr(visibility, "bypass", False))
         self._hidden: set[str] = set()
         # wid -> (project_id, ws_owner) awaiting a definitive verdict.
@@ -3091,6 +3118,18 @@ async def proxy_api(request: Request) -> Response:
         # service identity instead. Per-ws + bare events stay on
         # the user's identity for upstream audit attribution.
         use_service = path == "events/global"
+        if use_service:
+            # Elevating to the console's SERVICE identity bypasses the
+            # node's per-user filtering, so the raw cross-tenant firehose
+            # must be operator-gated — otherwise any authenticated user
+            # could read every tenant's (incl. private-project) workstream
+            # inventory through the node proxy. admin.cluster.inspect is
+            # the same permission the cluster-inspect surfaces use.
+            from turnstone.core.auth import require_permission
+
+            perm_err = require_permission(request, "admin.cluster.inspect")
+            if perm_err is not None:
+                return perm_err
         return await _proxy_sse(
             request,
             server_url,
@@ -3363,14 +3402,20 @@ async def _resolve_coordinator_or_404(
     Centralises the manager-first, storage-fallback, 404-mask ladder
     used by the coord-only verbs (``coordinator_children`` /
     ``coordinator_tasks``).  The shared verbs (history, detail, ...)
-    inline the same ladder via :func:`make_history_handler` /
-    :func:`make_detail_handler`.  Turnstone is a
-    trusted-team tool — ``user_id`` is metadata, not an access
-    boundary, so this helper no longer gates on row ownership; scope
-    auth (``admin.coordinator``) upstream is the gate.
+    route through :func:`_coordinator_tenant_check` instead (wired onto
+    ``coord_endpoint_config.tenant_check``).  Turnstone is a trusted-team
+    tool — ``user_id`` is metadata, not an ownership boundary, so this
+    helper does not gate on row ownership; ``admin.coordinator`` upstream
+    gates the surface.  It DOES enforce project tenancy, though: a
+    coordinator attached to a PRIVATE project stays confidential to its
+    members, so a non-member (even an ``admin.coordinator`` holder) is
+    404-masked, same as a missing row.
     """
     del user_id  # retained in signature for caller-site clarity; not consulted here
+    from turnstone.core.auth import WorkstreamProjectVisibility
+
     miss = JSONResponse({"error": "coordinator not found"}, status_code=404)
+    visibility = WorkstreamProjectVisibility.for_request(request, storage=storage)
     ws = coord_mgr.get(ws_id) if coord_mgr is not None else None
     if ws is None:
         if storage is None:
@@ -3386,10 +3431,61 @@ async def _resolve_coordinator_or_404(
             return None, miss
         if row is None or row.get("kind") != WorkstreamKind.COORDINATOR:
             return None, miss
+        # Project tenancy — the predicate may resolve a project row +
+        # membership, so judge it off the event loop.
+        if not await asyncio.to_thread(
+            visibility.ws_visible, row.get("project_id") or "", row.get("user_id") or ""
+        ):
+            return None, miss
         return None, None
+    if not await asyncio.to_thread(
+        visibility.ws_visible, getattr(ws, "project_id", "") or "", ws.user_id or ""
+    ):
+        return None, miss
     return ws, None
 
 
+def _coordinator_tenant_check(request: Request, ws_id: str, mgr: Any) -> JSONResponse | None:
+    """Project-tenancy gate for the lifted coordinator verbs.
+
+    Wired onto ``coord_endpoint_config.tenant_check`` (invoked SYNC in a
+    thread) so history / export / detail / set_title / send / approve / …
+    all enforce it.  ``admin.coordinator`` gates the coordinator surface
+    cluster-wide, so row OWNERSHIP is not enforced (any operator may drive
+    any coordinator) — but a coordinator attached to a PRIVATE project stays
+    confidential to its members.
+
+    Sync mirror of :func:`_resolve_coordinator_or_404`'s manager-first,
+    storage-fallback, coordinator-kind ladder (the in-memory manager is the
+    existence/kind authority; a storage row covers saved/closed
+    coordinators) plus the project-visibility gate.  Everything that fails —
+    unknown id, wrong kind, or a private project the caller can't see —
+    404-masks identically, so the surface is neither an existence nor a
+    private-project oracle.  Reusing the manager-first + kind ladder also
+    preserves the coord kind-isolation that :func:`make_set_title_handler`
+    previously got from the ``tenant_check is None`` manager-lookup guard.
+    """
+    from turnstone.core.auth import WorkstreamProjectVisibility
+    from turnstone.core.memory import get_workstream_row
+
+    miss = JSONResponse({"error": "coordinator not found"}, status_code=404)
+    ws = mgr.get(ws_id) if mgr is not None else None
+    if ws is not None:
+        # coord_mgr only holds coordinators, so kind is implied.
+        project_id = getattr(ws, "project_id", "") or ""
+        owner = ws.user_id or ""
+    else:
+        row = get_workstream_row(ws_id)
+        if row is None or row.get("kind") != WorkstreamKind.COORDINATOR:
+            return miss
+        project_id = row.get("project_id") or ""
+        owner = row.get("user_id") or ""
+    visibility = WorkstreamProjectVisibility.for_request(request)
+    if not visibility.ws_visible(project_id, ws_owner=owner):
+        return miss
+    return None
+
+
 def _auth_user_id(request: Request) -> str:
     """Thin shim over :func:`turnstone.core.web_helpers.auth_user_id`.
 
@@ -5633,14 +5729,14 @@ async def admin_create_token(request: Request) -> JSONResponse:
     scopes = body.get("scopes", "read,write,approve")
     expires_days = body.get("expires_days")
 
-    # Validate scopes
-    from turnstone.core.auth import VALID_SCOPES
+    # Validate scopes — ``service`` is NOT user-assignable (it bypasses
+    # private-project tenancy; only ServiceTokenManager / the JWT secret
+    # may mint it).
+    from turnstone.core.auth import reject_unassignable_scopes
 
-    requested = {s.strip() for s in scopes.split(",") if s.strip()}
-    if not requested or not requested.issubset(VALID_SCOPES):
-        return JSONResponse(
-            {"error": "Invalid scopes (allowed: read, write, approve)"}, status_code=400
-        )
+    scope_err = reject_unassignable_scopes(scopes)
+    if scope_err is not None:
+        return JSONResponse({"error": scope_err}, status_code=400)
 
     expires: str | None = None
     if expires_days is not None:
@@ -13519,11 +13615,20 @@ def create_app(
         coordinators must be ``open``ed before they can accept
         attachment operations.
         """
+        from turnstone.core.auth import WorkstreamProjectVisibility
         from turnstone.core.web_helpers import auth_user_id
 
         ws = mgr.get(ws_id)
         if ws is None:
             return "", JSONResponse({"error": "coordinator not found"}, status_code=404)
+        # Private-project tenancy: a coordinator attached to a private project
+        # serves attachments only to its members — admin.coordinator gates the
+        # surface, not the tenancy. 404-mask non-members like the other verbs.
+        visibility = WorkstreamProjectVisibility.for_request(request)
+        if not visibility.ws_visible(
+            getattr(ws, "project_id", "") or "", ws_owner=ws.user_id or ""
+        ):
+            return "", JSONResponse({"error": "coordinator not found"}, status_code=404)
         return ws.user_id or auth_user_id(request), None
 
     from turnstone.core.attachments import classify_upload as _coord_classify_upload
@@ -13550,7 +13655,8 @@ def create_app(
     coord_endpoint_config = SessionEndpointConfig(
         permission_gate=_require_admin_coordinator,
         manager_lookup=_require_coord_mgr,
-        tenant_check=None,  # cluster-wide admin.coordinator gate covers it
+        # admin.coordinator gates the surface; this gates private-project tenancy.
+        tenant_check=_coordinator_tenant_check,
         not_found_label="coordinator not found",
         audit_action_prefix="coordinator",
         supports_attachments=True,

turnstone/core/auth.py

@@ -73,6 +73,28 @@ _MIN_SECRET_LENGTH = 32  # 256 bits minimum for HMAC-SHA256
 
 VALID_SCOPES: frozenset[str] = frozenset({"read", "write", "approve", "service"})
 
+# Scopes a principal may be GRANTED via a user-facing token mint (the admin
+# token API / ``turnstone-admin create-token``).  ``service`` is deliberately
+# excluded: it is a full cross-tenant bypass (see
+# :meth:`WorkstreamProjectVisibility.for_request`) and must only ever be
+# minted by :class:`ServiceTokenManager` / operators holding the JWT secret —
+# never assigned to a user, or an admin could self-grant it and see every
+# private project's workstreams.
+ASSIGNABLE_SCOPES: frozenset[str] = VALID_SCOPES - frozenset({"service"})
+
+
+def reject_unassignable_scopes(scopes_csv: str) -> str | None:
+    """Validate a user-supplied comma-separated scope string for token mints.
+
+    Returns an error message when the request is empty or names any scope
+    outside :data:`ASSIGNABLE_SCOPES` (notably ``service``), else ``None``.
+    Shared by the admin token API and the CLI so the rule can't drift.
+    """
+    requested = {s.strip() for s in scopes_csv.split(",") if s.strip()}
+    if not requested or not requested.issubset(ASSIGNABLE_SCOPES):
+        return "Invalid scopes (allowed: read, write, approve)"
+    return None
+
 
 def jwt_version_slot() -> str:
     """Return ``major.minor`` from ``__version__`` for JWT version claims.
@@ -267,11 +289,13 @@ class WorkstreamProjectVisibility:
 
     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).
+    * ``bypass`` instances see everything — but ONLY service-scope callers
+      (the collector and other cluster machinery must never be blinded at
+      the node edge; user-facing filtering happens per-principal at the
+      console edge). No human principal bypasses: a private project's
+      workstreams are confidential even from admins — ``admin.cluster.inspect``
+      still gates the cluster-inspect *surfaces*, but a permitted admin only
+      sees the private-project rows they own or are a member of.
     * 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).
@@ -295,22 +319,23 @@ class WorkstreamProjectVisibility:
     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.
+        Only service-scoped tokens get a bypass instance (node→console
+        machine plumbing, re-filtered per-user at the console edge);
+        everyone else — admins included — filters as themselves. An admin
+        holding ``admin.cluster.inspect`` reaches the cluster-inspect
+        surfaces but still only sees private-project workstreams they own
+        or belong to.
         """
         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"))
-        )
+        bypass = bool(auth is not None and auth.has_scope("service"))
         return cls(uid, bypass=bypass, storage=storage)
 
     @property
     def bypass(self) -> bool:
-        """True when this principal sees everything (service scope /
-        ``admin.cluster.inspect``) — callers that transform payloads
-        (not just drop rows) use this to leave them untouched."""
+        """True when this principal sees everything (service scope only) —
+        callers that transform payloads (not just drop rows) use this to
+        leave them untouched."""
         return self._bypass
 
     def _resolve_storage(self) -> Any:

turnstone/core/session_routes.py

@@ -1735,6 +1735,16 @@ def make_open_handler(
                 return JSONResponse({"error": cfg.not_found_label}, status_code=404)
             ws_id = resolved
 
+        # Tenancy gate BEFORE the already-loaded shortcut and before
+        # ``mgr.open`` rehydrates — otherwise ``open`` is a private-project
+        # existence/metadata oracle (it returns the auto-titled name) and an
+        # unauthorized resurrection of a closed private workstream into the
+        # pool. Interactive wires ownership, coord wires project tenancy.
+        if cfg.tenant_check is not None:
+            err_tenant = await asyncio.to_thread(cfg.tenant_check, request, ws_id, mgr)
+            if err_tenant is not None:
+                return err_tenant
+
         # Already-loaded shortcut — both kinds return the same
         # ``{ws_id, name, already_loaded: true}`` shape.
         existing = mgr.get(ws_id)

turnstone/core/web_helpers.py

@@ -320,8 +320,8 @@ def resolve_workstream_owner(
     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
+    creator, and service-scope callers — everyone else, admins included,
+    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